diff --git a/backend/database/conversations.py b/backend/database/conversations.py index 75ca26eac58..69dd11880b5 100644 --- a/backend/database/conversations.py +++ b/backend/database/conversations.py @@ -898,7 +898,15 @@ def update_conversation(uid: str, conversation_id: str, update_data: dict) -> bo doc_level = doc_snapshot.to_dict().get('data_protection_level', 'standard') prepared_data = _prepare_conversation_for_write(update_data, uid, doc_level) - doc_ref.update(prepared_data) + try: + doc_ref.update(prepared_data) + except NotFound: + # The conversation was deleted between the existence read above and + # this commit. The contract of this function is to report a gone owner + # as False — not to raise — so callers like the pusher's private-cloud + # audio sync take their designed gone-owner path (stop syncing, release + # the audio budget) instead of logging an ERROR and retrying forever. + return False return True diff --git a/backend/docs/listen_pusher_pipeline.mdx b/backend/docs/listen_pusher_pipeline.mdx index d04ae71f3c2..e55fcefc42c 100644 --- a/backend/docs/listen_pusher_pipeline.mdx +++ b/backend/docs/listen_pusher_pipeline.mdx @@ -158,6 +158,13 @@ beforehand races the pusher's own registration write, and losing that race in the other direction would delete a recording that had just become referenceable. +When the private-cloud flush writes `audio_files` onto a conversation that a +mid-session delete removes between the flush's existence read and its commit, +`update_conversation` reports the owner as gone (returns `False`) instead of +raising: the flush then takes the same stop-syncing path as any other deleted +owner rather than surfacing `404 No document to update` as an ERROR on every +remaining flush of the session. + ## Capture-device provenance Every capture client with WebSocket-upgrade header support sends diff --git a/backend/docs/operational/aac-decode-failure-reporting.md b/backend/docs/operational/aac-decode-failure-reporting.md new file mode 100644 index 00000000000..c2098f6354f --- /dev/null +++ b/backend/docs/operational/aac-decode-failure-reporting.md @@ -0,0 +1,95 @@ +# AAC Decode Failures — typed failures, not silent native noise + +Date: 2026-09-01 · Scope: `utils/aac.py` (`AACDecoder.decode`) · +Guard tests: `tests/unit/test_aac_decode_failure_reporting.py` + +## What happened + +The GCP prod error feed (backend-listen) carried a family of +`ERROR:libav.aac:*` signatures continuously across 2026-08-30/31 — +combined ~130 events / 30 min at peaks: + +``` +ERROR:libav.aac:Channel element 1.7 is not allocated (×12–17 / 30 min) +ERROR:libav.aac:Reserved bit set. (×6–11 / 30 min) +ERROR:libav.aac:Error decoding AAC frame header. (×3–14 / 30 min) +ERROR:libav.aac:Number of bands (…) exceeds limit (…) (×11 / 30 min) +ERROR:libav.aac:Number of scalefactor bands in group (…) … (×5 / 30 min) +``` + +These are FFmpeg's **native codec diagnostics** for a frame its AAC decoder +rejected, emitted while `AACDecoder.decode` held the frame. Reproduced +locally against the pinned PyAV 12.0.0 by feeding the real decoder +payload-scrambled and truncated ADTS frames: every corrupt shape raises +`av.InvalidDataError`/`av.UndefinedError` **and** emits the corresponding +native `libav.aac` line — the prod signatures are our decode loop rejecting +corrupt/truncated client audio, frame after frame. + +## What was broken + +`AACDecoder.decode` caught `(EOFError, av.AVError)` and returned `b''`. Two +consequences, one contract violation: + +1. **The receiver's decode-failure contract never fired for AAC.** The + listen receiver (`routers/listen/receiver.py`) already owns undecodable- + frame reporting: `_record_decode_failure` logs a per-frame warning with + the codec's own message, payload size, and streak, and the one-shot + `record_fallback(component='silent_mic', …, outcome='exhausted')` at 50 + consecutive drops (1 s at the omi 20 ms frame cadence) — the contract + #11732 established when opus streams were dropping silently. All of it + hangs off the decoder *raising*. opuslib raises `OpusError`; the AAC + decoder swallowed, so a fully undecodable AAC stream recorded a whole + session with no transcript, no ring buffer, no mixed audio, and **no + fallback metric** — a fail-open branch with no operator signal (the + fallback-telemetry contract, `docs/agents/fallback-telemetry.md`). +2. **The only trace left was the context-free native line.** FFmpeg's + `ERROR:libav.aac:…` carries no uid, session, codec name, payload size, or + streak — strictly less information than the warning the receiver would + have logged — and it duplicated that warning per frame once the receiver + path existed, polluting the error feed the Loop S sensor watches. + +## The fix + +- `AACDecoder.decode` now **raises `AACDecodeError`** (message = the FFmpeg + error verbatim, cause = the original `av.AVError`) for frames the codec + rejects, and still returns `b''` only for benign no-output input (empty + payload, encoder priming). The receiver's existing `except Exception → + _record_decode_failure('aac', …)` handles the rest — no receiver change. +- `NativeDuplicateSuppressionFilter` on the `libav.aac` logger drops + FFmpeg's re-report **only while our own decode call is on the stack** + (thread-local flag, set/cleared in `finally`); every other `av` consumer's + native errors still flow. + +## Operator-visible after this lands + +- Per corrupt frame: `WARNING … Listen audio frame decode failed codec=aac + type=AACDecodeError bytes=N streak=M detail=` (uid/session + context via the existing warning shape). +- A fully undecodable stream: one `silent_mic` fallback metric per session — + the alertable signal that was missing. +- The `ERROR:libav.aac:*` family should disappear from the error feed; any + residual native lines would indicate a decode outside this module (speech + profiles, transcode), which keeps its own context. + +## Regression coverage + +`tests/unit/test_aac_decode_failure_reporting.py` drives the real +`AACDecoder` over real ADTS frames encoded in-process (real PyAV encoder → +real codec context) and the real `ListenReceiver.receive_data` loop; only +the websocket transport is scripted. Covers: corrupt/truncated frames raise +the typed error with FFmpeg detail (mono and stereo, header-only fragments); +clean frames decode; recovery after a corrupt frame and after a burst; +native duplicate suppressed inside our decode window with a negative control +proving the suppression is not vacuous; window-scope (an unrelated native +error right after still logs) and thread-locality; receiver streak +progression, warning shape, one-shot silent-mic fallback at the threshold, +streak reset on recovery, interleaved corrupt/clean streams, and the +`initialize_decoders` production wiring. + +Failure-Class: FC-typed-failure-collapsed-to-generic — instance fix; the +class was canonized by #11487 (proactive lane) and applied at the live-STT +boundary by #11732 (opus): a codec that already produces a typed failure +must not have it collapsed into silence by the caller. Here the collapse +happened one layer lower — the decoder itself swallowing `AVError` — so the +receiver's whole reporting contract (built for exactly this failure shape) +never engaged for AAC. diff --git a/backend/docs/operational/soniox-typed-rejections.md b/backend/docs/operational/soniox-typed-rejections.md new file mode 100644 index 00000000000..79f1902c543 --- /dev/null +++ b/backend/docs/operational/soniox-typed-rejections.md @@ -0,0 +1,65 @@ +# Soniox in-stream rejections: keep the type, split the severity + +Incident window: 2026-08-30/31 (backend-listen, Loop S sensor). +`ERROR:utils.stt.soniox:Soniox streaming error:` was the #4 error signature +(~52 events / 30 min) carrying three unrelated provider behaviors on one +free-text line: + +| Frame | Typed reason | Owner | Severity | +|---|---|---|---| +| 400 `invalid_request` "No audio received" | `soniox_idle_timeout` | this session's VAD pattern | WARNING | +| 402 `organization_balance_exhausted` | `soniox_account_state` | the provider/account | **ERROR** | +| 413 `max_duration_reached` | `soniox_rotation` | documented protocol rotation | WARNING | + +## What was broken + +- Every in-stream error frame — including two shapes that are the protocol + answering how the session was used — logged at ERROR, so a fleet of + VAD-starved sockets was indistinguishable from a serving outage. +- The socket laundered the typed frame into free-text `death_reason` + (`soniox error: 402 organization_balance_exhausted …`); every terminal funnel + (`terminate_live_stt_session`, the death monitor, the send path) collapsed it + to `connection_lost`, so `omi_live_stt_terminal_failures_total` carried no + cause distinction. +- `_fallback_failure_reason` matched neither `exhausted` nor `balance`, filing + the 402 as `provider_5xx` in fallback telemetry. +- `bounded_provider` did not know the live-path tokens `soniox` / + `deepgram_cloud`, so terminal-failure metrics reported `provider='unknown'`. +- A 402 provider still **accepts connects** while refusing every stream: + mid-session failover moved each dying session to the next provider and the + session survived, so the terminal path that feeds the selection circuit never + ran, and each new session was handed straight back to the refusing provider. + +## The contract now + +- `soniox_death_reason(error_code, error_type)` bounds the provider's own typed + frame into the terminal vocabulary; unknown shapes degrade to + `connection_lost` rather than growing per-message cardinality. +- The socket latches both the raw frame (`death_reason`, for logs) and the + bounded type (`typed_death_reason`); `GatedSTTSocket` proxies the typed + reason so the VAD gate cannot erase it. +- `live_stt_terminal_reason(socket, fallback)` lets every terminal funnel + report the provider's type instead of its own vantage point. +- Severity follows fault ownership: only a 402 account-state refusal stays at + ERROR (`Soniox streaming error:`); idle-timeout and rotation log + `Soniox stream closed:` at WARNING. Never mute a signature without + classifying fault origin first — see `ws-auth-rejection-severity.md` for the + same rule at the auth boundary. +- Fleet evidence: `note_typed_provider_death` at the failover seam (and + `soniox_account_state` in the terminal path) opens the provider's + process-local selection circuit for one cooldown. Session-scoped reasons + deliberately do not — an idle timeout is this session's VAD pattern, not a + provider fault. +- `_fallback_failure_reason` classifies `exhausted`/`balance` text as `quota`; + `bounded_provider` accepts the live-path provider tokens. + +## Regression coverage + +`tests/unit/test_soniox_typed_rejections.py` drives the real `SafeSonioxSocket` +receive loop over the exact prod frame shapes, the real `ListenReceiver` +failover and death-monitor paths, and the real terminal/send paths; only the +process-global circuit opener is patched. + +Failure-Class: FC-typed-failure-collapsed-to-generic (instance fix; class +canonized by #11487 in the proactive lane — the same collapse at the live-STT +boundary). diff --git a/backend/docs/operational/ws-auth-rejection-severity.md b/backend/docs/operational/ws-auth-rejection-severity.md new file mode 100644 index 00000000000..ee085c73d34 --- /dev/null +++ b/backend/docs/operational/ws-auth-rejection-severity.md @@ -0,0 +1,68 @@ +# WS Auth Rejection Severity — error-feed hygiene + +Date: 2026-08-31 · Scope: `utils/other/endpoints.py` (`_verify_ws_auth`) · +Guard tests: `tests/unit/test_ws_auth_rejection_logging.py` + +## What happened + +The GCP prod error feed (backend-listen) carried two of its top signatures +for 16+ consecutive hours on 2026-08-30/31: + +``` +ERROR:utils.other.endpoints:WebSocket auth failed: code=4001 error=Token expired, 1788101015 < 1788114388 (×9–47 / 30 min) +ERROR:utils.other.endpoints:WebSocket auth failed: code=4001 error=Certificate for key id 6ac9047f… not found. (×1–34 / 30 min) +``` + +Both are **client-caused rejections the server handled correctly**: + +- `Token expired` — samples show the presented Firebase ID tokens were 3.7h, + 7.5h and 18h past `exp` (tokens live 1h). Devices with suspended clocks or + long sleep/reconnect loops replaying a dead credential. +- `Certificate for key id 6ac9047f… not found` — every sample across 14h + names the *same* key id. Verified against Google's currently served x509 + set (`/robot/v1/metadata/x509/securetoken@system.gserviceaccount.com`) on + 2026-08-31: that kid is **retired** — absent from the fresh cert list. A + client cohort minted before the key rotation presents tokens no server can + verify; note `google.oauth2.id_token.verify_token` checks the `kid` against + the cert set *before* the `exp` check, so this fires regardless of expiry + and for arbitrary freshness. + +The rejections (and their close codes — 4001 "refresh" / 4004 "re-login") are +the auth protocol working. The **defect was classification**: every expected +client-caused rejection was logged at ERROR, so the stale-client reconnect +population was indistinguishable from a serving outage in the feed that pages +humans — during the same window a *real* outage (the Modulate STT 5xx storm) +had to be found inside it. + +## The rule (severity follows fault origin) + +In `_verify_ws_auth`, the narrow catch now routes logging through +`_log_ws_auth_rejection(close_code, error)`: + +| Verify outcome | Fault origin | Severity | Close code | +|---|---|---|---| +| `InvalidIdTokenError` (incl. `ExpiredIdTokenError`, `RevokedIdTokenError`) | client — Firebase evaluated and refused the token | **WARNING** | unchanged (4001/4004/1008) | +| `CertificateFetchError` | **server** — could not fetch Google's certs to even evaluate | **ERROR** | 4001 | +| unexpected exception | server | ERROR | 1008 | + +Close codes and reasons are untouched — clients already get the right +remediation hint. The warning message is `WebSocket auth rejected: …` +(still carries code + error text for debugging); server faults keep +`WebSocket auth failed: …`. + +## Applying the same rule elsewhere + +- Client-caused auth rejections should not be top error signatures. If a new + one appears, first establish fault origin from samples (stale `exp` deltas, + repeated retired `kid`s vs. cert-fetch/transport failures), then classify + severity accordingly — do not mute by signature-silencing. +- The message-derived branches in `_get_ws_auth_close` (matching + `'expired'`/`'certificate'` in the text) exist for exceptions crossing this + boundary without a typed class; they do not change fault-origin + classification. +- Related incident record and test fixtures: + `tests/unit/test_ws_auth_rejection_logging.py` replays the exact prod + message shapes. + +Failure-Class: FC-request-input-rejection-escapes-as-server-fault (instance +fix; class canonized by #11853). diff --git a/backend/routers/listen/receiver.py b/backend/routers/listen/receiver.py index 4f2c3d72d59..78fd267f946 100644 --- a/backend/routers/listen/receiver.py +++ b/backend/routers/listen/receiver.py @@ -47,7 +47,9 @@ flush_live_stt_buffer, live_stt_initialization_failure, live_stt_socket_is_dead, + live_stt_terminal_reason, live_stt_upstream_failure, + note_typed_provider_death, send_live_stt_audio, terminate_live_stt_session, ) @@ -520,6 +522,12 @@ async def _rebuild_stt_socket_locked(self) -> bool: self._stt_failed_providers.add(dead_provider) if len(self._stt_failed_providers) > MAX_STT_FAILOVERS: return False + # A provider-level typed rejection (Soniox 402 balance-exhausted) is + # fleet-level evidence the failover would otherwise swallow: the session + # survives on the next provider, so the terminal path that normally + # feeds the selection circuit never runs for it, and the NEXT session is + # handed straight back to the provider that refuses every stream. + note_typed_provider_death(self.stt_socket, dead_provider) service, language, model = get_stt_service_for_language( self.host.language, @@ -588,7 +596,7 @@ async def _monitor_stt_death(self) -> None: self.host.request.websocket, self.host.state, failure=live_stt_upstream_failure(self._serving_provider()), - reason='connection_lost', + reason=live_stt_terminal_reason(socket, 'connection_lost'), platform=self.host.client_device_context.platform, ) return diff --git a/backend/tests/unit/test_aac_decode_failure_reporting.py b/backend/tests/unit/test_aac_decode_failure_reporting.py new file mode 100644 index 00000000000..b6f8e8133e8 --- /dev/null +++ b/backend/tests/unit/test_aac_decode_failure_reporting.py @@ -0,0 +1,570 @@ +"""Regression: undecodable AAC frames must fail loudly, not silently. + +Failure-Class: FC-typed-failure-collapsed-to-generic — instance fix; the AAC +decoder collapsed every FFmpeg rejection into ``b''`` so the receiver's +decode-failure contract (per-frame warning, streak, ``silent_mic`` fallback) +never fired for AAC, leaving only FFmpeg's context-free +``ERROR:libav.aac:Channel element 1.7 is not allocated`` / ``Reserved bit +set.`` native lines (~130 events / 30 min across the ``libav.aac`` family, +Loop S sensor, 2026-08-30/31) as the trace of sessions that recorded with no +transcript, no ring buffer, and no fallback metric. Same collapse at the +audio-decode boundary as #11732 fixed at the opus boundary. + +These tests drive the real ``AACDecoder`` (real PyAV codec context, real +ADTS bytes encoded in-process) and the real ``ListenReceiver.receive_data`` +loop — only the websocket transport is scripted, the pattern +``test_listen_receiver_decode_failure_report.py`` established for opus. +Corruption shapes are the deterministic per-signature mutations validated +against the pinned PyAV 12.0.0 (see the operational note +``docs/operational/aac-decode-failure-reporting.md``). +""" + +import array +import logging +import math +import os +import tempfile +import threading +from types import SimpleNamespace + +import av +import numpy as np +import pytest + +from routers.listen import receiver as receiver_module +from routers.listen.receiver import DECODE_FAILURE_STREAK_ALERT, ListenReceiver +from utils.aac import AACDecodeError, AACDecoder, NativeDuplicateSuppressionFilter + +# --------------------------------------------------------------------------- +# Real ADTS frames, encoded in-process (no fixtures, no network). +# --------------------------------------------------------------------------- + + +def _encode_adts_frames(count: int = 10, rate: int = 16000, layout: str = 'mono', channels: int = 1) -> list: + """Real AAC/ADTS frames via a real PyAV encoder (mono/stereo, any rate).""" + with tempfile.NamedTemporaryFile(suffix='.aac', delete=False) as tmp: + path = tmp.name + try: + container = av.open(path, 'w', format='adts') + stream = container.add_stream('aac', rate=rate) + stream.layout = layout + for i in range(count): + n = 1024 + if channels == 1: + pcm = array.array('h', (int(10000 * math.sin(i * 300 * t / rate)) for t in range(n))) + arr = np.frombuffer(pcm, dtype='int16').reshape(1, -1) + else: + left = array.array('h', (int(9000 * math.sin(i * 200 * t / rate)) for t in range(n))) + right = array.array('h', (int(7000 * math.cos(i * 130 * t / rate)) for t in range(n))) + interleaved = array.array('h') + for l, r in zip(left, right): + interleaved.append(l) + interleaved.append(r) + arr = np.frombuffer(interleaved, dtype='int16').reshape(1, -1) + frame = av.AudioFrame.from_ndarray(arr, format='s16', layout=layout) + frame.sample_rate = rate + pkts = stream.encode(frame) + if pkts: + container.mux(pkts) + pkts = stream.close() + if pkts: + container.mux(pkts) + container.close() + with open(path, 'rb') as fh: + data = fh.read() + finally: + os.unlink(path) + + frames = [] + i = 0 + while i + 7 <= len(data): + if data[i] == 0xFF and (data[i + 1] & 0xF0) == 0xF0: + length = ((data[i + 3] & 0x03) << 11) | (data[i + 4] << 3) | (data[i + 5] >> 5) + frames.append(data[i : i + length]) + i += length + else: + i += 1 + return frames + + +def _capture_native_logs(): + """Handler capturing libav.aac records; returns (records, remover).""" + records: list = [] + + def _capture(record: logging.LogRecord) -> None: + records.append(record) + + handler = logging.Handler() + handler.emit = _capture # type: ignore[assignment, method-assign] # test-local capture + logger = logging.getLogger('libav.aac') + logger.addHandler(handler) + return records, lambda: logger.removeHandler(handler) + + +@pytest.fixture(scope='module') +def adts_frames(): + return _encode_adts_frames() + + +@pytest.fixture(scope='module') +def adts_frames_stereo(): + return _encode_adts_frames(count=6, layout='stereo', channels=2) + + +@pytest.fixture(scope='module') +def adts_frames_44k(): + return _encode_adts_frames(count=6, rate=44100) + + +# --------------------------------------------------------------------------- +# Deterministic per-signature corruption shapes (probe-validated) +# --------------------------------------------------------------------------- + + +def _corrupt_payload(frame: bytes) -> bytes: + """Valid ADTS header, scrambled payload — the shape behind the prod + ``channel element … is not allocated`` / ``Reserved bit set.`` lines.""" + return frame[:7] + bytes((b * 13 + 7) & 0xFF for b in frame[7:]) + + +def _bitflip_first_quarter(frame: bytes) -> bytes: + """One flipped byte in the first quarter — ``Number of bands (…) exceeds + limit (…)`` / ``channel element`` shapes; deterministic per frame.""" + pos = len(frame) // 4 + return frame[:pos] + bytes([frame[pos] ^ 0xFF]) + frame[pos + 1 :] + + +def _overwrite_with_ff_run(frame: bytes) -> bytes: + """16-byte 0xFF run after the header — ``Error decoding AAC frame + header.`` (a false sync-word pattern inside the payload).""" + return frame[:7] + b'\xff' * 16 + frame[23:] + + +def _truncate(frame: bytes, keep: float = 0.7) -> bytes: + return frame[: int(len(frame) * keep)] + + +# --------------------------------------------------------------------------- +# Real decoder: typed failure instead of silence +# --------------------------------------------------------------------------- + + +class TestAACDecoderRaisesTypedError: + + def test_corrupt_frame_raises_aac_decode_error(self, adts_frames): + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(_corrupt_payload(adts_frames[1])) + + def test_truncated_frame_raises_aac_decode_error(self, adts_frames): + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(_truncate(adts_frames[1])) + + def test_garbage_without_adts_sync_raises_aac_decode_error(self): + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(bytes(60)) + + def test_band_limit_corruption_raises_aac_decode_error(self, adts_frames): + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(_bitflip_first_quarter(adts_frames[2])) + + def test_false_sync_run_raises_aac_decode_error(self, adts_frames): + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(_overwrite_with_ff_run(adts_frames[3])) + + def test_header_only_payload_raises_aac_decode_error(self, adts_frames): + """A truncated header-only fragment (< 7 bytes) is rejected, not + silently skipped: the receiver's streak must see it, because a client + sending header fragments mid-stream is the desync shape behind the + prod ``Error decoding AAC frame header.`` lines.""" + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(adts_frames[1][:6]) + + def test_stereo_corrupt_frame_raises_aac_decode_error(self, adts_frames_stereo): + """Corruption detection must not depend on channel count: a scrambled + stereo payload raises the same typed failure through the stereo + resampler path.""" + decoder = AACDecoder(uid='u', session_id='s', channels=2) + with pytest.raises(AACDecodeError): + decoder.decode(_corrupt_payload(adts_frames_stereo[1])) + + def test_error_message_carries_ffmpeg_detail_and_cause(self, adts_frames): + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError) as excinfo: + decoder.decode(_corrupt_payload(adts_frames[1])) + assert str(excinfo.value) + assert isinstance(excinfo.value.__cause__, av.AVError) + + def test_clean_frames_decode_to_pcm(self, adts_frames): + decoder = AACDecoder(uid='u', session_id='s') + pcm = decoder.decode(adts_frames[1]) + assert pcm + # mono s16 PCM: even byte count, at least one 1024-sample frame + assert len(pcm) % 2 == 0 + assert len(pcm) >= 2048 + + def test_persistent_context_decodes_a_whole_stream(self, adts_frames): + """The codec context is persistent by design — every frame of a real + stream must decode through the one context (regression guard for the + decode() rewrite touching context lifecycle).""" + decoder = AACDecoder(uid='u', session_id='s') + total = 0 + for frame in adts_frames[1:]: + pcm = decoder.decode(frame) + assert len(pcm) % 2 == 0 + total += len(pcm) + assert total >= 2048 * (len(adts_frames) - 1) + + def test_decoder_recovers_after_a_corrupt_frame(self, adts_frames): + decoder = AACDecoder(uid='u', session_id='s') + first = decoder.decode(adts_frames[1]) + with pytest.raises(AACDecodeError): + decoder.decode(_corrupt_payload(adts_frames[2])) + after = decoder.decode(adts_frames[3]) + assert first and after + + def test_empty_payload_still_returns_empty_bytes(self): + decoder = AACDecoder(uid='u', session_id='s') + assert decoder.decode(b'') == b'' + + +# --------------------------------------------------------------------------- +# Constructor contract: resampler configuration actually used +# --------------------------------------------------------------------------- + + +class TestDecoderResamplerContract: + + def test_stereo_frames_decode_to_two_channel_pcm(self, adts_frames_stereo): + """channels=2 must configure a stereo resampler: decoded PCM stays + two-channel (double the mono byte count for the same samples).""" + decoder = AACDecoder(uid='u', session_id='s', channels=2) + mono = AACDecoder(uid='u', session_id='s', channels=1) + pcm_stereo = decoder.decode(adts_frames_stereo[1]) + pcm_mono_of_stereo = mono.decode(adts_frames_stereo[1]) + assert pcm_stereo and pcm_mono_of_stereo + assert len(pcm_stereo) == 2 * len(pcm_mono_of_stereo) + + def test_44100hz_input_resamples_to_16k(self, adts_frames_44k): + """sample_rate=16000 must drive real resampling: 44.1 kHz input comes + back as 16 kHz PCM. Expected samples derive from the frames the ADTS + parser actually found minus the encoder-priming frame (index 0); + measured output lands within 5% of that.""" + decoder = AACDecoder(uid='u', session_id='s', sample_rate=16000) + decoded_frames = adts_frames_44k[1:] + total_bytes = 0 + for frame in decoded_frames: + total_bytes += len(decoder.decode(frame)) + total_samples = total_bytes // 2 + expected = len(decoded_frames) * 1024 * 16000 / 44100 + assert abs(total_samples - expected) < expected * 0.05 + + +# --------------------------------------------------------------------------- +# Native duplicate suppression: no context-free ERROR:libav.aac lines +# --------------------------------------------------------------------------- + + +class TestNativeDuplicateSuppressed: + + def test_corrupt_decode_emits_no_libav_aac_error_log(self, adts_frames): + decoder = AACDecoder(uid='u', session_id='s') + records, remove = _capture_native_logs() + try: + with pytest.raises(AACDecodeError): + decoder.decode(_corrupt_payload(adts_frames[1])) + finally: + remove() + assert not [r for r in records if r.levelno >= logging.ERROR] + + def test_unsuppressed_the_same_corruption_does_emit_native_log(self, adts_frames): + """Negative control: with the suppression filter removed, the same + corrupt frame DOES produce the native libav.aac ERROR — proving the + suppression test is not vacuous.""" + corrupt = _corrupt_payload(adts_frames[1]) + libav_logger = logging.getLogger('libav.aac') + filters_before = list(libav_logger.filters) + libav_logger.filters = [ + f for f in libav_logger.filters if not isinstance(f, NativeDuplicateSuppressionFilter) + ] # negative control + records, remove = _capture_native_logs() + try: + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(corrupt) + finally: + remove() + libav_logger.filters = filters_before + assert records, 'expected the native libav.aac ERROR without suppression' + + def test_suppression_scoped_to_decode_window_only(self, adts_frames): + """Outside our own decode call the libav.aac logger must stay live — + other av users (transcode endpoints, speech profiles) still get their + native error logs. The error raised inside decode must not suppress a + later, unrelated native error.""" + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(_corrupt_payload(adts_frames[1])) + + records, remove = _capture_native_logs() + try: + logging.getLogger('libav.aac').error('channel element 9.9 is not allocated') + finally: + remove() + assert len(records) == 1 + + def test_suppression_window_is_thread_local(self): + """One session mid-decode must not mute a concurrent thread's native + error: the flag is thread-local by contract.""" + import utils.aac as aac_module + + other_thread_records: list = [] + ready = threading.Event() + release = threading.Event() + + def other_thread_decode(): + records, remove = _capture_native_logs() + try: + # this thread's flag is unset, so its native log must flow + # even while the main thread holds its own window open. + ready.set() + release.wait(timeout=5) + logging.getLogger('libav.aac').error('channel element 7.7 is not allocated') + finally: + remove() + other_thread_records.extend(records) + + main_flag_active = False + + def hold_window(): + nonlocal main_flag_active + aac_module.aac_decode_in_progress.active = True + main_flag_active = getattr(aac_module.aac_decode_in_progress, 'active', False) + ready.wait(timeout=5) + release.set() + aac_module.aac_decode_in_progress.active = False + + t_hold = threading.Thread(target=hold_window) + t_other = threading.Thread(target=other_thread_decode) + t_other.start() + t_hold.start() + t_hold.join(timeout=5) + t_other.join(timeout=5) + + assert main_flag_active, 'main thread window opened' + assert other_thread_records, 'concurrent thread native log must not be muted' + assert not getattr(aac_module.aac_decode_in_progress, 'active', False) + + def test_flag_cleared_when_decode_raises(self, adts_frames): + """The finally-clause contract: even the raising path must leave the + window closed (a stuck-open flag would mute all future native logs + in this thread).""" + import utils.aac as aac_module + + decoder = AACDecoder(uid='u', session_id='s') + with pytest.raises(AACDecodeError): + decoder.decode(_corrupt_payload(adts_frames[1])) + assert not getattr(aac_module.aac_decode_in_progress, 'active', False) + + def test_suppression_filter_directly_reflects_flag(self): + """Unit contract of the filter: flag unset → record passes; flag set + in this thread → record dropped.""" + import utils.aac as aac_module + + record = logging.LogRecord('libav.aac', logging.ERROR, __file__, 1, 'msg', None, None) + filt = NativeDuplicateSuppressionFilter() + aac_module.aac_decode_in_progress.active = True + try: + assert filt.filter(record) is False + finally: + aac_module.aac_decode_in_progress.active = False + assert filt.filter(record) is True + + +# --------------------------------------------------------------------------- +# Real receiver loop: streak, warning shape, one-shot silent-mic fallback +# --------------------------------------------------------------------------- + + +@pytest.fixture +def anyio_backend(): + return 'asyncio' + + +class _FramesWebSocket: + def __init__(self, frames): + self.frames = iter(frames) + + async def receive(self): + return next(self.frames) + + +def _host(websocket, channels=1): + return SimpleNamespace( + request=SimpleNamespace(websocket=websocket, uid='uid-1', codec='aac', sample_rate=16000, channels=channels), + 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, channels=1): + websocket = _FramesWebSocket(list(frames) + [{'type': 'websocket.disconnect', 'code': 1000}]) + receiver = ListenReceiver(_host(websocket, channels), [], {}) + receiver.aac_decoder = AACDecoder(uid='uid-1', session_id='sid-1', channels=channels) + return receiver + + +@pytest.fixture +def recorded_fallbacks(monkeypatch): + calls = [] + monkeypatch.setattr(receiver_module, 'record_fallback', lambda **kwargs: calls.append(kwargs)) + return calls + + +class TestReceiverIntegration: + + @pytest.mark.anyio + async def test_corrupt_aac_frame_logs_codec_message_and_advances_streak( + self, adts_frames, caplog, recorded_fallbacks + ): + corrupt = _corrupt_payload(adts_frames[1]) + receiver = _receiver([{'bytes': corrupt}]) + with caplog.at_level(logging.WARNING, logger=receiver_module.__name__): + await receiver.receive_data() + + (message,) = [r.getMessage() for r in caplog.records if 'decode failed' in r.getMessage()] + assert 'codec=aac' in message + assert 'type=AACDecodeError' in message + assert f'bytes={len(corrupt)}' in message + assert 'streak=1' in message + + @pytest.mark.anyio + async def test_aac_stream_undecodable_reports_silent_mic_once(self, recorded_fallbacks): + frame = bytes(60) + frames = [{'bytes': frame}] * (DECODE_FAILURE_STREAK_ALERT + 3) + receiver = _receiver(frames) + + await receiver.receive_data() + + assert receiver.decode_failure_streak == DECODE_FAILURE_STREAK_ALERT + 3 + assert recorded_fallbacks == [ + { + 'component': 'silent_mic', + 'from_mode': 'aac', + 'to_mode': 'none', + 'reason': 'capability_mismatch', + 'outcome': 'exhausted', + } + ] + + @pytest.mark.anyio + async def test_recovery_after_corrupt_frames_resets_streak_no_fallback(self, adts_frames, recorded_fallbacks): + corrupt_then_clean = [{'bytes': _corrupt_payload(adts_frames[1])}] * 3 + [{'bytes': adts_frames[2]}] + receiver = _receiver(corrupt_then_clean) + + await receiver.receive_data() + + assert receiver.decode_failure_streak == 0 + assert recorded_fallbacks == [] + + @pytest.mark.anyio + async def test_clean_aac_stream_produces_no_decode_failure_logs(self, adts_frames, caplog, recorded_fallbacks): + frames = [{'bytes': f} for f in adts_frames[1:6]] + receiver = _receiver(frames) + + with caplog.at_level(logging.WARNING, receiver_module.__name__): + await receiver.receive_data() + + assert not [r for r in caplog.records if 'decode failed' in r.getMessage()] + assert recorded_fallbacks == [] + + @pytest.mark.anyio + async def test_corrupt_and_clean_interleaved_resets_streak_each_recovery( + self, adts_frames, caplog, recorded_fallbacks + ): + """The prod-relevant shape: a flaky client stream with corrupt frames + scattered among good ones. Each good frame resets the streak, so the + session never reaches the silent-mic threshold it must not reach — + but every corrupt frame is still individually reported.""" + stream = [] + expected_failures = 0 + for i in range(1, 8): + if i % 2 == 0: + stream.append({'bytes': _corrupt_payload(adts_frames[i])}) + expected_failures += 1 + else: + stream.append({'bytes': adts_frames[i]}) + receiver = _receiver(stream) + + with caplog.at_level(logging.WARNING, logger=receiver_module.__name__): + await receiver.receive_data() + + failures = [r for r in caplog.records if 'decode failed' in r.getMessage()] + assert len(failures) == expected_failures + assert receiver.decode_failure_streak == 0 + assert recorded_fallbacks == [] + + @pytest.mark.anyio + async def test_decoder_recovers_after_burst_of_corrupt_frames(self, adts_frames, recorded_fallbacks): + """A burst of corruption must not poison the persistent codec + context: the first clean frame after a run of failures still decodes, + and the streak resets — the flaky-network recovery shape.""" + frames = [{'bytes': _corrupt_payload(adts_frames[1])}] * 4 + [{'bytes': adts_frames[3]}] + receiver = _receiver(frames) + + await receiver.receive_data() + + assert receiver.decode_failure_streak == 0 + assert recorded_fallbacks == [] + + @pytest.mark.anyio + async def test_second_consecutive_corrupt_frame_advances_streak(self, adts_frames, caplog): + """The streak is per-frame evidence, cumulative across consecutive + failures: the second corrupt frame logs streak=2 with the same + codec/type shape (the per-frame report the prod feed lacked).""" + corrupt = _corrupt_payload(adts_frames[1]) + receiver = _receiver([{'bytes': corrupt}, {'bytes': corrupt}]) + + with caplog.at_level(logging.WARNING, logger=receiver_module.__name__): + await receiver.receive_data() + + failures = [r.getMessage() for r in caplog.records if 'decode failed' in r.getMessage()] + assert len(failures) == 2 + assert 'streak=1' in failures[0] + assert 'streak=2' in failures[1] + + @pytest.mark.anyio + async def test_initialize_decoders_builds_real_aac_decoder(self): + """The production wiring path: initialize_decoders must construct the + real AACDecoder for codec=aac with the request's uid, sample rate, + and channel count (not leave the slot None, which would AttributeError + into the streak path for every frame).""" + websocket = _FramesWebSocket([{'type': 'websocket.disconnect', 'code': 1000}]) + host = _host(websocket) + host.session_id = 'sid-real' + receiver = ListenReceiver(host, [], {}) + + receiver.initialize_decoders() + + assert isinstance(receiver.aac_decoder, AACDecoder) + assert receiver.aac_decoder.uid == 'uid-1' + assert receiver.aac_decoder.session_id == 'sid-real' + assert receiver.aac_decoder.resampler.rate == 16000 diff --git a/backend/tests/unit/test_chat_file_gateway_stub_import_contract.py b/backend/tests/unit/test_chat_file_gateway_stub_import_contract.py new file mode 100644 index 00000000000..d9f4b7baf56 --- /dev/null +++ b/backend/tests/unit/test_chat_file_gateway_stub_import_contract.py @@ -0,0 +1,126 @@ +"""Regression test: the chat_file suite must import the real module it loads. + +Production/CI evidence (2026-08-31): e6b545c1b8 (file-chat gateway-lane +degrade) added ``is_gateway_model_not_found`` to the +``utils.llm.gateway_client`` import block in ``utils/other/chat_file.py``, +but the local ``gateway_client`` stub installed by +``test_chat_file_upload_unsupported.py`` did not grow the attribute. That +suite loads the REAL ``chat_file`` module against its stub, so the module +failed at import time and every test in the file errored at setup — +deterministically, on ``main``, while the push pipeline (which does not run +backend unit tests) stayed green. Any future import added to ``chat_file`` +re-creates the same outage the same way. + +The contract under test: the stub surface a test harness installs must +satisfy the real module it loads — pinned here by importing the real +``chat_file`` against a stub built from the SAME attribute recipe the +chat-file suite uses, so a missing attribute surfaces here first (named, +actionable) instead of as seven setup ERRORS in the sibling suite. + +Failure-Class: FC-ship-before-required-route — the sibling instance of the +same class as e6b545c1b8/#12444 (a client — here the test suite — that +hard-imports a name its serving environment must provide before it loads); +instance fix only, guard surface = this behavioral import test. +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from types import ModuleType +from unittest.mock import MagicMock + +os.environ.setdefault( + "ENCRYPTION_SECRET", + "omi_ZwB2ZNqB2HHpMK6wStk7sTpavJiPTFg7gXUHnc4tFABPU6pZ2c2DKgehtfgu4RZv", +) + +import tests.unit._chat_router_test_harness as harness # noqa: E402 + +BACKEND_DIR = Path(__file__).resolve().parents[2] + + +def _chat_file_gateway_stub() -> ModuleType: + """The exact attribute recipe test_chat_file_upload_unsupported installs.""" + gateway_client = ModuleType("utils.llm.gateway_client") + gateway_client.should_route_features_through_gateway = MagicMock(return_value=False) + gateway_client.CHAT_AGENT_ROUTE_DIRECT = "direct" + gateway_client.CHAT_AGENT_ROUTE_GATEWAY = "gateway" + gateway_client.get_chat_agent_route = MagicMock(return_value="direct") + gateway_client.file_chat_auto_lane_id = MagicMock(return_value="omi:auto:file-chat-vision") + gateway_client.file_chat_feature_header = MagicMock(return_value={}) + gateway_client.get_file_chat_gateway_async_client = MagicMock() + gateway_client.get_file_chat_gateway_sync_client = MagicMock() + gateway_client.is_gateway_model_not_found = MagicMock(return_value=False) + return gateway_client + + +def _install_stack(stub: ModuleType) -> None: + """Install the chat-file suite's package + stub stack (with restore).""" + harness.install_package("models", BACKEND_DIR / "models") + harness.install_package("database", BACKEND_DIR / "database") + harness.install_package("utils", BACKEND_DIR / "utils") + harness.install_package("utils.other", BACKEND_DIR / "utils" / "other") + harness.install_package("utils.llm", BACKEND_DIR / "utils" / "llm") + + harness.wire_common_stubs(harness.install_module) + harness.install_module("models.app") + + harness.install_module("utils.llm.gateway_client", stub) + + +def test_chat_file_imports_against_the_suite_gateway_stub(): + """The real chat_file module must load against the suite's stub recipe. + + Drives the production seam (import of ``utils/other/chat_file.py``) + through the same ``load_real_module`` the broken suite uses, with the + same ``sys.modules`` snapshot/restore discipline. Fails at the exact + import if the stub falls behind the module's import block. + """ + saved = {k: v for k, v in sys.modules.items()} + try: + _install_stack(_chat_file_gateway_stub()) + module = harness.load_real_module( + "utils.other.chat_file", BACKEND_DIR / "utils" / "other" / "chat_file.py" + ) + assert module is not None + # The module's real public surface: the typed errors the router maps + # to HTTP codes, present since the suite's founding. + assert hasattr(module, "UnsupportedChatFileError") + assert hasattr(module, "StaleChatFileError") + assert hasattr(module, "ProviderRejectedChatFileError") + finally: + harness.cleanup(saved) + + +def test_a_stale_stub_fails_on_the_exact_missing_attribute(): + """If chat_file grows a gateway import the stub lacks, the failure names it. + + Builds the PRE-FIX stub recipe (no ``is_gateway_model_not_found``) and + asserts the ImportError carries the missing attribute name, so the + recurring failure mode of this class is a one-line diagnosis instead of + seven opaque setup errors. If this ever raises AssertionError instead, + the module's import block changed and the recipe above must follow. + """ + saved = {k: v for k, v in sys.modules.items()} + try: + stale = _chat_file_gateway_stub() + del stale.is_gateway_model_not_found + _install_stack(stale) + try: + harness.load_real_module( + "utils.other.chat_file", BACKEND_DIR / "utils" / "other" / "chat_file.py" + ) + except ImportError as e: + assert "is_gateway_model_not_found" in str(e), ( + "import must fail on the exact missing attribute so the stub " + "fix is actionable, got: %s" % e + ) + else: + raise AssertionError( + "stale stub unexpectedly satisfied chat_file imports — the " + "guard recipe no longer matches the module's import block" + ) + finally: + harness.cleanup(saved) diff --git a/backend/tests/unit/test_embeddings_route_absent_fallback_telemetry.py b/backend/tests/unit/test_embeddings_route_absent_fallback_telemetry.py new file mode 100644 index 00000000000..266034d5e8c --- /dev/null +++ b/backend/tests/unit/test_embeddings_route_absent_fallback_telemetry.py @@ -0,0 +1,266 @@ +"""Regression test: an embeddings deploy-skew degrade must be countable. + +Production evidence (2026-08-30/31, Loop S sensor + GCP logs): the prod LLM +gateway (deployed 2026-08-20) predates the ``/v1/embeddings`` route +(2026-08-28, #12337), so every embeddings call 404s and the proxy degrades to +the direct path (#12444). That degrade branch changes provider and loses the +gateway ledger row, but it never called ``record_fallback`` — violating the +repo-wide contract in ``docs/agents/fallback-telemetry.md`` and +``backend/AGENTS.md`` rule 10 ("a branch that changes mode MUST call +``record_fallback``; do not invent per-domain counters"). The sibling +gateway-lane degrade in file chat (#12449, +``_record_gateway_file_chat_fallback``) already records; this file pins the +same contract for the embeddings surface. + +The distinction under test is deliberately two-sided: + +1. Every route-absent degrade — all four OpenAI embedding methods plus + ``gemini_embed_query`` — increments ``omi_fallback_total`` with the + surface's lane labels, so operators can see how much embeddings traffic + and ledger spend is bypassing the gateway while the skew lasts. +2. The narrative ERROR log stays once per process (the skew holds until the + gateway is redeployed; the gateway's own access log counts the 404s), and + a gateway that *owns* the route (typed ``model_not_found``) still raises + without any fallback telemetry, so lane misconfiguration is never masked + as deploy skew. + +Failure-Class: FC-ship-before-required-route — the violated contract is the +same one #12444 declared (client route shipped before the serving gateway), +but this instance is the observability half: a mode-changing degrade that +leaves no shared-telemetry trail. Instance fix within the existing class, +guard surface = these behavioral tests on the real proxy seam. +""" + +from __future__ import annotations + +import logging +import os +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +os.environ.setdefault('OPENAI_API_KEY', '***') +os.environ.setdefault('ENCRYPTION_SECRET', 'omi_ZwB2ZNqB2HHpMK6wStk7sTpavJiPTFg7gXUHnc4tFABPU6pZ2c2DKgehtfgu4RZv') + +import utils.llm.clients as clients # noqa: E402 +from utils.llm.gateway_client import LLM_GATEWAY_FEATURE_MODE_ENV_VAR # noqa: E402 +from utils.observability import fallback as fallback_mod # noqa: E402 + + +class _FakeCounterChild: + def __init__(self, parent, labels): + self.parent = parent + self.labels = labels + + def inc(self, amount: float = 1.0): + self.parent.increments.append((self.labels, amount)) + + +class _FakeCounter: + def __init__(self): + self.increments: list[tuple[dict[str, str], float]] = [] + + def labels(self, **labels): + return _FakeCounterChild(self, labels) + + +def _gateway_mode(monkeypatch): + monkeypatch.setenv(LLM_GATEWAY_FEATURE_MODE_ENV_VAR, 'gateway') + monkeypatch.setenv('OMI_ENV_STAGE', 'dev') + monkeypatch.delenv('K_SERVICE', raising=False) + monkeypatch.delenv('KUBERNETES_SERVICE_HOST', raising=False) + + +def _route_absent_error() -> httpx.HTTPStatusError: + """A 404 shaped like Starlette's answer for a path the app never registered.""" + request = httpx.Request('POST', 'http://gateway/v1/embeddings') + response = httpx.Response(404, json={'detail': 'Not Found'}, request=request) + return httpx.HTTPStatusError('Client error 404', request=request, response=response) + + +def _model_not_found_error() -> httpx.HTTPStatusError: + """A 404 the gateway itself emits: it owns the route, it rejected the lane.""" + request = httpx.Request('POST', 'http://gateway/v1/embeddings') + response = httpx.Response( + 404, + json={'error': {'message': 'unknown model', 'type': 'api_error', 'code': 'model_not_found'}}, + request=request, + ) + return httpx.HTTPStatusError('Client error 404', request=request, response=response) + + +@pytest.fixture +def fallback_counter(monkeypatch): + counter = _FakeCounter() + monkeypatch.setattr(fallback_mod, 'OMI_FALLBACK_TOTAL', counter) + return counter + + +@pytest.fixture(autouse=True) +def _reset_route_absent_warning(monkeypatch): + monkeypatch.setattr(clients, '_gateway_embeddings_route_absent_warned', False, raising=False) + + +def test_embed_query_degrade_records_fallback_telemetry(monkeypatch, fallback_counter): + _gateway_mode(monkeypatch) + direct = MagicMock() + direct.embed_query.return_value = [0.7, 0.8] + + with patch.object( + clients, 'invoke_openai_embeddings_gateway', MagicMock(side_effect=_route_absent_error()) + ), patch.object(clients, 'get_byok_key', MagicMock(return_value=None)), patch.object( + clients._OpenAIEmbeddingsProxy, '_resolve', MagicMock(return_value=direct) + ): + vector = clients.embeddings.embed_query('q') + + assert vector == [0.7, 0.8] + assert fallback_counter.increments == [ + ( + { + 'component': 'llm_gateway', + 'from_mode': 'gateway_embeddings', + 'to_mode': 'direct_embeddings', + 'reason': 'capability_mismatch', + 'outcome': 'degraded', + }, + 1.0, + ) + ] + + +def test_embed_documents_degrade_records_fallback_telemetry(monkeypatch, fallback_counter): + _gateway_mode(monkeypatch) + direct = MagicMock() + direct.embed_documents.return_value = [[0.1], [0.2]] + + with patch.object( + clients, 'invoke_openai_embeddings_gateway', MagicMock(side_effect=_route_absent_error()) + ), patch.object(clients, 'get_byok_key', MagicMock(return_value=None)), patch.object( + clients._OpenAIEmbeddingsProxy, '_resolve', MagicMock(return_value=direct) + ): + vectors = clients.embeddings.embed_documents(['a', 'b']) + + assert vectors == [[0.1], [0.2]] + assert fallback_counter.increments[0][0]['from_mode'] == 'gateway_embeddings' + assert fallback_counter.increments[0][0]['to_mode'] == 'direct_embeddings' + assert fallback_counter.increments[0][0]['outcome'] == 'degraded' + + +@pytest.mark.asyncio +async def test_aembed_query_degrade_records_fallback_telemetry(monkeypatch, fallback_counter): + _gateway_mode(monkeypatch) + direct = MagicMock() + direct.aembed_query = AsyncMock(return_value=[0.3, 0.4]) + + with patch.object( + clients, 'ainvoke_openai_embeddings_gateway', AsyncMock(side_effect=_route_absent_error()) + ), patch.object(clients, 'get_byok_key', MagicMock(return_value=None)), patch.object( + clients._OpenAIEmbeddingsProxy, '_resolve', MagicMock(return_value=direct) + ): + vector = await clients.embeddings.aembed_query('q') + + assert vector == [0.3, 0.4] + assert fallback_counter.increments[0][0]['component'] == 'llm_gateway' + assert fallback_counter.increments[0][0]['reason'] == 'capability_mismatch' + + +def test_every_degrade_counts_even_after_the_process_log_fired(monkeypatch, fallback_counter, caplog): + """The metric must count every degrade; only the narrative log is once-per-process. + + In prod the skew holds for hours and the proxy degrades thousands of times + an hour. A once-per-process counter would hide that volume — the exact + blind spot ``omi_fallback_total`` exists to surface. + """ + _gateway_mode(monkeypatch) + direct = MagicMock() + direct.embed_query.return_value = [0.1] + + def gateway_refuses(texts, **_kwargs): + raise _route_absent_error() + + with patch.object( + clients, 'invoke_openai_embeddings_gateway', MagicMock(side_effect=gateway_refuses) + ), patch.object(clients, 'get_byok_key', MagicMock(return_value=None)), patch.object( + clients._OpenAIEmbeddingsProxy, '_resolve', MagicMock(return_value=direct) + ): + with caplog.at_level(logging.ERROR, logger=clients.logger.name): + for _ in range(3): + assert clients.embeddings.embed_query('q') == [0.1] + + error_logs = [r for r in caplog.records if 'serves no /v1/embeddings route' in r.message] + assert len(error_logs) == 1, 'narrative log stays once per process' + assert len(fallback_counter.increments) == 3, 'metric fires per degrade' + + +def test_gemini_embed_query_degrade_records_fallback_telemetry(monkeypatch, fallback_counter): + _gateway_mode(monkeypatch) + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={'embedding': {'values': [0.6]}}) + + with patch.object(clients, 'get_byok_key', MagicMock(return_value=None)), patch.object( + clients, 'invoke_gemini_embedding_gateway', MagicMock(side_effect=_route_absent_error()) + ), patch.object( + clients.httpx, + 'post', + MagicMock( + side_effect=lambda url, **kwargs: httpx.Client(transport=httpx.MockTransport(handler)).post(url, **kwargs) + ), + ): + values = clients.gemini_embed_query('screen activity') + + assert values == [0.6] + assert fallback_counter.increments[0][0]['from_mode'] == 'gateway_embeddings' + assert fallback_counter.increments[0][0]['to_mode'] == 'direct_embeddings' + + +def test_gateway_owned_rejection_records_no_fallback(monkeypatch, fallback_counter): + """A typed model_not_found must raise without telemetry: masking lane drift as a + counted degrade would corrupt the deploy-skew signal these labels carry.""" + _gateway_mode(monkeypatch) + + with patch.object( + clients, 'invoke_openai_embeddings_gateway', MagicMock(side_effect=_model_not_found_error()) + ), patch.object(clients, 'get_byok_key', MagicMock(return_value=None)): + with pytest.raises(httpx.HTTPStatusError): + clients.embeddings.embed_query('q') + + assert fallback_counter.increments == [] + + +def test_healthy_gateway_path_records_no_fallback(monkeypatch, fallback_counter): + """The ledger lane staying up is the happy path — no mode change, no telemetry.""" + _gateway_mode(monkeypatch) + + with patch.object( + clients, 'invoke_openai_embeddings_gateway', MagicMock(return_value=[[0.5, 0.6]]) + ), patch.object(clients, 'get_byok_key', MagicMock(return_value=None)): + vector = clients.embeddings.embed_query('query') + + assert vector == [0.5, 0.6] + assert fallback_counter.increments == [] + + +def test_byok_key_failure_keeps_its_existing_fallback_labels(monkeypatch, fallback_counter): + """The BYOK→Omi-key degrade predates this change; it must keep firing exactly one + event per failure (not zero, and not the route-absent labels).""" + _gateway_mode(monkeypatch) + calls: list[dict] = [] + + def gateway_call(texts, *, byok_api_key=None): + calls.append({'texts': texts, 'byok': byok_api_key}) + if len(calls) == 1: + raise httpx.HTTPStatusError( + 'Client error 401', request=MagicMock(), response=MagicMock(status_code=401) + ) + return [[0.9]] + + with patch.object(clients, 'invoke_openai_embeddings_gateway', MagicMock(side_effect=gateway_call)), patch.object( + clients, 'get_byok_key', MagicMock(return_value='sk-user') + ): + vector = clients.embeddings.embed_query('q') + + assert vector == [0.9] + # No route-absent degrade happened: the gateway owned and answered the call. + assert all(inc[0]['from_mode'] != 'gateway_embeddings' for inc in fallback_counter.increments) diff --git a/backend/tests/unit/test_serve_death_feeds_provider_circuit.py b/backend/tests/unit/test_serve_death_feeds_provider_circuit.py new file mode 100644 index 00000000000..afec92b7387 --- /dev/null +++ b/backend/tests/unit/test_serve_death_feeds_provider_circuit.py @@ -0,0 +1,308 @@ +"""Regression: a serve-time provider death must feed the selection circuit. + +Production loop sensor (backend-listen, 30-min window 2026-08-31T05:30Z) +recorded 62 occurrences of: + + ERROR:utils.stt.streaming:Modulate streaming error: Internal server error + +Modulate accepted every WebSocket upgrade, served audio, and then died +mid-session with an internal error. Session teardown handled the dead client +correctly (stt_failed + 1011), but provider *selection* kept choosing +Modulate for every reconnecting client for the whole incident, because +``connect_stt_socket_with_fallback`` only records connect-time outcomes: + +- the dying provider still accepts connects, so each reconnect calls + ``record_success`` and RESETS the failure counter; +- the serve-time death never reached the circuit at all. + +So the counter provably never reaches ``failure_threshold`` under reconnect +load (connect -> die -> reconnect -> die), and the fleet hammers the dead +provider for the duration of the outage instead of skipping to the healthy +fallback configured right behind it (#11752 built that chain; this is the +feedback leg it never had). + +These tests drive the real ``terminate_live_stt_session`` funnel (not the +individual send paths — all four funnel there) and the real +``ProviderCircuitBreaker`` state machine at the real singleton seam +(``streaming._modulate_circuit`` etc.), so both the classification boundary +and the state transition are exercised for real. + +Failure-Class: new — the violated contract is the circuit's learning +boundary: a health signal that gates selection must be fed by every terminal +observation of the thing whose health it claims to track. Serve-time death +was terminal, observed, and invisible to it. +""" + +from dataclasses import dataclass +from types import SimpleNamespace +from typing import Any, List +from unittest.mock import patch + +import pytest + +from utils.stt import live_failure, streaming +from utils.stt.live_failure import terminate_live_stt_session +from utils.stt.outcomes import TranscriptionFailure, TranscriptionOutcome +from utils.stt.provider_resilience import ProviderCircuitBreaker +from utils.stt.streaming import STTService, open_provider_selection_circuit + + +@dataclass +class _Session: + active: bool = True + close_code: int = 1001 + stt_terminal_failure: bool = False + live_transcription_attempt: Any = None + client_live_transcription_attempt: Any = None + + +class _ClientSocket: + def __init__(self) -> None: + self.closed = False + + async def send_json(self, _data: Any) -> None: + pass + + async def close(self, code: int = 1000, reason: str | None = None) -> None: + self.closed = True + + +def _failure(provider: str) -> TranscriptionFailure: + return TranscriptionFailure(TranscriptionOutcome.UPSTREAM_ERROR, provider=provider, retryable=True) + + +def _connect_returning(socket: Any): + async def _connect() -> Any: + return socket + + return _connect + + +def _connect_raising(error: BaseException): + async def _connect() -> Any: + raise error + + return _connect + + +@pytest.fixture(autouse=True) +def _quiet_metrics(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(live_failure, 'record_live_stt_failure', lambda **_labels: None) + monkeypatch.setattr(streaming, 'record_fallback', lambda **_kwargs: None) + + +@pytest.fixture(autouse=True) +def _fresh_real_circuits(monkeypatch: pytest.MonkeyPatch) -> None: + """Swap fresh real breakers into the singleton seam, one per provider. + + The production code paths address ``streaming.__circuit`` + singletons. Fresh instances keep this file's assertions about circuit + STATE deterministic regardless of test order inside the file, while still + exercising the real ProviderCircuitBreaker transitions (open, cooldown, + half-open probe) rather than a mock's canned answers. + """ + for service in STTService: + monkeypatch.setattr( + streaming, + f'_{service.value}_circuit', + ProviderCircuitBreaker(failure_threshold=3, cooldown_seconds=30.0), + ) + + +async def _terminate(provider: str, reason: str) -> _ClientSocket: + websocket = _ClientSocket() + await terminate_live_stt_session( + websocket, _Session(), failure=_failure(provider), reason=reason, platform='ios' + ) + return websocket + + +# --- the incident shape: serve-time death must open the circuit --------------- + + +@pytest.mark.asyncio +async def test_serve_time_death_opens_the_serving_providers_circuit() -> None: + await _terminate('modulate', 'connection_lost') + + assert streaming._modulate_circuit.state == 'open' + # The incident's other providers are untouched: the death is evidence + # about the provider that was serving, not about its fallbacks. + assert streaming._deepgram_circuit.state == 'closed' + assert streaming._parakeet_circuit.state == 'closed' + + +@pytest.mark.asyncio +async def test_send_phase_death_also_feeds_the_circuit() -> None: + """A provider that died during send is the same terminal evidence.""" + await _terminate('parakeet', 'send_failed') + + assert streaming._parakeet_circuit.state == 'open' + + +@pytest.mark.asyncio +async def test_reconnect_skips_the_provider_that_died_serving() -> None: + """End to end at the selection seam: after a serve-time death the next + connect must walk straight to the healthy fallback instead of retrying + the provider that just died. + """ + await _terminate('modulate', 'connection_lost') + assert streaming._modulate_circuit.state == 'open' + + healthy_fallback = SimpleNamespace(is_connection_dead=False, death_reason=None) + with patch('utils.stt.provider_resilience.STT_FALLBACK_LIVENESS_GRACE_SECONDS', 0.05): + socket, service = await streaming.connect_stt_socket_with_fallback( + primary_service=STTService.modulate, + connect_primary=_connect_raising( + AssertionError('selection must not reconnect to the provider that died serving') + ), + connect_deepgram=_connect_returning(healthy_fallback), + ) + + assert socket is healthy_fallback + assert service == STTService.deepgram + + +# --- the boundary: what must NOT open the circuit ----------------------------- + + +@pytest.mark.asyncio +async def test_initialization_failure_does_not_open_the_circuit() -> None: + """Connect-phase failures already flow through the threshold in the selection helper. + + Recording them here too would double-count one failure and bypass the + threshold entirely for config errors at startup. + """ + await _terminate('modulate', 'initialization_failed') + + assert streaming._modulate_circuit.state == 'closed' + + +@pytest.mark.asyncio +async def test_unavailable_socket_does_not_open_the_circuit() -> None: + """``socket_unavailable`` is local state (no socket exists), not provider behavior.""" + await _terminate('modulate', 'socket_unavailable') + + assert streaming._modulate_circuit.state == 'closed' + + +@pytest.mark.asyncio +async def test_unbounded_reason_does_not_open_the_circuit() -> None: + """``_bounded_reason`` maps anything unknown to connection_lost — pinned so + the mapping cannot silently start tripping circuits for new reasons.""" + websocket = _ClientSocket() + await terminate_live_stt_session( + websocket, + _Session(), + failure=_failure('modulate'), + reason='something_new', + platform='ios', + ) + + assert websocket.closed is True + assert streaming._modulate_circuit.state == 'open' # bounded to connection_lost + + +@pytest.mark.asyncio +async def test_idempotent_terminal_runs_the_circuit_leg_once() -> None: + websocket = _ClientSocket() + session = _Session() + + await terminate_live_stt_session( + websocket, session, failure=_failure('modulate'), reason='connection_lost', platform='ios' + ) + # A second observer (teardown, another channel) sees stt_terminal_failure + # already latched and must not run the terminal path again. + await terminate_live_stt_session( + websocket, session, failure=_failure('modulate'), reason='connection_lost', platform='ios' + ) + + assert streaming._modulate_circuit.state == 'open' + # Still inside the 30s cooldown: the second call did not re-open with a + # later timestamp (which would extend the outage window). + assert streaming._modulate_circuit.allow_request() is False + + +# --- unknown provider names must not break the terminal path ------------------ + + +@pytest.mark.asyncio +async def test_unknown_provider_name_terminates_normally() -> None: + """The terminal close is the contract; circuit bookkeeping is best-effort. + + ``provider`` can be None or a future name STTService does not know; the + session must still terminate exactly as before. + """ + websocket = await _terminate('not-a-provider', 'connection_lost') + + assert websocket.closed is True + + +def test_open_provider_selection_circuit_rejects_unknown_names() -> None: + assert open_provider_selection_circuit(None, reason='connection_lost') is False + assert open_provider_selection_circuit('not-a-provider', reason='connection_lost') is False + + +def test_open_provider_selection_circuit_opens_the_named_provider() -> None: + assert open_provider_selection_circuit('soniox', reason='send_failed') is True + assert streaming._soniox_circuit.state == 'open' + + +# --- the state machine: recovery still exists --------------------------------- + + +def test_circuit_recovers_through_the_half_open_probe() -> None: + """Opening on serve-death must not brick the provider forever. + + After the cooldown the breaker offers exactly one probe; a successful + probe closes the circuit again. This pins the recovery half of the new + transition against future "safety" additions. + """ + now: List[float] = [0.0] + circuit = ProviderCircuitBreaker(failure_threshold=3, cooldown_seconds=30.0, clock=lambda: now[0]) + + circuit.record_serve_failure() + assert circuit.state == 'open' + assert circuit.allow_request() is False # inside cooldown + + now[0] = 31.0 + assert circuit.state == 'open' + assert circuit.allow_request() is True # the single half-open probe + assert circuit.allow_request() is False # and only one + + circuit.record_success() + assert circuit.state == 'closed' + assert circuit.allow_request() is True + + +# --- ProviderCircuitBreaker.record_serve_failure unit contract ---------------- + + +def test_record_serve_failure_opens_immediately_regardless_of_counter() -> None: + """The whole point: serve deaths cannot accumulate through record_failure. + + With threshold 3, two connect failures leave the counter at 2. A + serve-time death must open the circuit NOW — waiting for a third event + that the reconnect/reset cycle provably never delivers is the bug. + """ + circuit = ProviderCircuitBreaker(failure_threshold=3, cooldown_seconds=30.0) + circuit.record_failure() + circuit.record_failure() + assert circuit.state == 'closed' + + circuit.record_serve_failure() + assert circuit.state == 'open' + + +def test_record_failure_counter_is_reset_by_connect_success_between_deaths() -> None: + """Documents the mechanism the incident exposed (no behavior change here). + + Threshold 3, but under reconnect load each serve death is followed by the + next session's successful CONNECT before that session dies too: the + counter oscillates 1 -> 0 -> 1 and never opens. This is why serve deaths + need their own transition instead of reusing record_failure. + """ + circuit = ProviderCircuitBreaker(failure_threshold=3, cooldown_seconds=30.0) + for _ in range(10): + circuit.record_failure() # a session died; the client reconnects... + circuit.record_success() # ...and the dying provider ACCEPTS the connect + assert circuit.state == 'closed' diff --git a/backend/tests/unit/test_soniox_typed_rejections.py b/backend/tests/unit/test_soniox_typed_rejections.py new file mode 100644 index 00000000000..708cfb53644 --- /dev/null +++ b/backend/tests/unit/test_soniox_typed_rejections.py @@ -0,0 +1,450 @@ +"""Soniox in-stream rejections must keep their type, severity, and fleet effect. + +Production evidence (backend-listen, GCP 2026-08-30/31, Loop S sensor): +``ERROR:utils.stt.soniox:Soniox streaming error:`` was the #4 error signature, +~52 events per 30-minute window, three distinct provider shapes sharing one +free-text line: 400 ``invalid_request "No audio received"`` (×42, the VAD gate +starving the socket), 402 ``organization_balance_exhausted`` (×7, the account +cannot serve ANY stream), 413 ``max_duration_reached`` (×3, documented +rotation). The socket laundered all three into one death reason, the terminal +path collapsed them to ``connection_lost``, the 402's ``exhausted`` text +matched neither 'limit' nor 'quota' so fallback telemetry filed it as +``provider_5xx``, and a balance-exhausted provider kept accepting connects — +so mid-session failover quietly moved every session to the next provider and +selection never learned, handing each NEW session straight back to the +provider refusing to serve it. + +Failure-Class: FC-typed-failure-collapsed-to-generic — instance fix in the +class canonized by #11487, in the live-STT boundary instead of the proactive +lane. The typed rejection the provider already produced must survive to every +consumer: the log severity (only a 402 is a provider fault worth paging on), +the bounded terminal-failure reason vocabulary, the fallback-reason +classification, the provider label, and — for the one shape that is fleet +evidence (402) — the process-local selection circuit, opened at the failover +seam that would otherwise swallow it. + +These tests drive the REAL ``SafeSonioxSocket`` receive loop (synthetic +provider frames), the REAL failover method on ``ListenReceiver``, the REAL +death monitor, and the REAL terminal/send paths from ``utils.stt.live_failure`` +— only the process-global circuit opener is patched, at its lazy-import seam. +""" + +import asyncio +import json +import logging +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from routers.listen import receiver as receiver_mod +from routers.listen.receiver import ListenReceiver +from utils.stt import live_failure +from utils.stt.live_failure import ( + LIVE_STT_FAILURE_CLOSE_CODE, + live_stt_terminal_reason, + note_typed_provider_death, + send_live_stt_audio, + terminate_live_stt_session, +) +from utils.stt.outcomes import TranscriptionFailure, TranscriptionOutcome, bounded_provider +from utils.stt.soniox import ( + SONIOX_DEATH_ACCOUNT_STATE, + SONIOX_DEATH_IDLE_TIMEOUT, + SONIOX_DEATH_ROTATION, + SafeSonioxSocket, + soniox_death_reason, +) +from utils.stt.streaming import STTService, _fallback_failure_reason +from utils.stt.vad_gate import GatedSTTSocket + + +class FakeWebSocket: + """Provider WebSocket yielding a scripted inbound frame list.""" + + def __init__(self, inbound): + self._inbound = list(inbound) + self.sent = [] + + async def send(self, data): + self.sent.append(data) + + async def close(self): + pass + + def __aiter__(self): + async def gen(): + for msg in self._inbound: + yield json.dumps(msg) + + return gen() + + +def _drive_socket(frames, *, wait: float = 0.05): + """Run a real SafeSonioxSocket over the scripted frames and return it.""" + + def run(): + captured = [] + + async def main(): + ws = FakeWebSocket(frames) + sock = SafeSonioxSocket(ws, captured.append, asyncio.get_running_loop()) + await asyncio.sleep(wait) + return sock + + return asyncio.run(main()) + + return run() + + +def _frame(code, error_type, message): + return {'error_code': code, 'error_type': error_type, 'error_message': message} + + +# --------------------------------------------------------------------------- +# soniox_death_reason: bound the provider's own typed frame, degrading safely. +# --------------------------------------------------------------------------- + + +def test_a_402_balance_exhausted_frame_classifies_as_account_state(): + assert soniox_death_reason(402, 'organization_balance_exhausted') == SONIOX_DEATH_ACCOUNT_STATE + + +def test_a_400_no_audio_frame_classifies_as_idle_timeout(): + assert soniox_death_reason(400, 'invalid_request') == SONIOX_DEATH_IDLE_TIMEOUT + + +def test_a_413_max_duration_frame_classifies_as_documented_rotation(): + assert soniox_death_reason(413, 'max_duration_reached') == SONIOX_DEATH_ROTATION + + +def test_unknown_shapes_degrade_to_connection_lost(): + """A new provider error shape must not grow metric cardinality per message.""" + assert soniox_death_reason(500, 'internal_server_error') == 'connection_lost' + assert soniox_death_reason('not-a-code', 'whatever') == 'connection_lost' + assert soniox_death_reason(None, None) == 'connection_lost' + + +def test_the_error_type_match_tolerates_provider_casing(): + assert soniox_death_reason(402, 'Organization_Balance_Exhausted') == SONIOX_DEATH_ACCOUNT_STATE + + +# --------------------------------------------------------------------------- +# Severity: only an account-state refusal is a provider fault worth an ERROR. +# --------------------------------------------------------------------------- + + +def test_a_402_frame_logs_at_error_as_a_provider_fault(caplog): + with caplog.at_level(logging.WARNING, logger='utils.stt.soniox'): + sock = _drive_socket([_frame(402, 'organization_balance_exhausted', 'Organization balance exhausted.')]) + assert sock.is_connection_dead + errors = [r for r in caplog.records if r.levelno == logging.ERROR and 'Soniox streaming error:' in r.message] + assert errors, 'a 402 account-state refusal must stay an ERROR: it is the outage signal' + + +def test_an_idle_timeout_frame_logs_at_warning_not_error(caplog): + with caplog.at_level(logging.WARNING, logger='utils.stt.soniox'): + _drive_socket([_frame(400, 'invalid_request', 'No audio received')]) + assert [r.levelname for r in caplog.records if 'Soniox' in r.message] == ['WARNING'] + + +def test_a_rotation_frame_logs_at_warning_not_error(caplog): + with caplog.at_level(logging.WARNING, logger='utils.stt.soniox'): + _drive_socket([_frame(413, 'max_duration_reached', 'Maximum duration reached')]) + assert [r.levelname for r in caplog.records if 'Soniox' in r.message] == ['WARNING'] + + +def test_the_raw_provider_frame_stays_on_the_death_latch(): + """The typed reason bounds telemetry; the raw text stays on the latch for logs.""" + sock = _drive_socket([_frame(402, 'organization_balance_exhausted', 'Organization balance exhausted.')]) + assert '402' in (sock.death_reason or '') + assert 'organization_balance_exhausted' in (sock.death_reason or '') + assert sock.typed_death_reason == SONIOX_DEATH_ACCOUNT_STATE + + +def test_a_live_socket_reports_no_typed_reason(): + sock = _drive_socket( + [{'tokens': [{'text': 'hi', 'is_final': True, 'speaker': 1, 'start_ms': 0, 'duration_ms': 100}]}] + ) + assert not sock.is_connection_dead + assert sock.typed_death_reason is None + + +# --------------------------------------------------------------------------- +# The VAD gate must not erase the typed rejection of the socket it wraps. +# --------------------------------------------------------------------------- + + +def test_the_vad_gate_proxies_the_typed_death_reason(): + dead = _drive_socket([_frame(402, 'organization_balance_exhausted', 'Organization balance exhausted.')]) + gated = GatedSTTSocket(dead) + assert gated.typed_death_reason == SONIOX_DEATH_ACCOUNT_STATE + assert gated.is_connection_dead + + +def test_the_gate_reports_none_for_an_untyped_wrapped_socket(): + class UntypedSocket: + is_connection_dead = False + death_reason = None + + gated = GatedSTTSocket(UntypedSocket()) + assert gated.typed_death_reason is None + + +# --------------------------------------------------------------------------- +# live_stt_terminal_reason: every terminal funnel reports the provider's type. +# --------------------------------------------------------------------------- + + +def test_a_terminal_funnel_prefers_the_typed_rejection_over_its_vantage_point(): + sock = SimpleNamespace(typed_death_reason='soniox_rotation') + assert live_stt_terminal_reason(sock, 'connection_lost') == 'soniox_rotation' + + +def test_an_unknown_typed_value_is_bounded_to_the_fallback(): + sock = SimpleNamespace(typed_death_reason='not-a-known-reason') + assert live_stt_terminal_reason(sock, 'send_failed') == 'send_failed' + + +def test_a_socket_without_a_latch_falls_back_cleanly(): + assert live_stt_terminal_reason(object(), 'connection_lost') == 'connection_lost' + + +# --------------------------------------------------------------------------- +# Fleet evidence: a 402 refusal must reach the selection circuit. +# --------------------------------------------------------------------------- + + +def _circuit_recorder(): + calls = [] + + def fake_opener(provider, *, reason): + calls.append((provider, reason)) + return True + + return calls, fake_opener + + +def test_a_typed_account_state_death_opens_the_selection_circuit(): + calls, opener = _circuit_recorder() + sock = SimpleNamespace(typed_death_reason=SONIOX_DEATH_ACCOUNT_STATE) + with patch('utils.stt.streaming.open_provider_selection_circuit', side_effect=opener): + assert note_typed_provider_death(sock, 'soniox') is True + assert calls == [('soniox', 'soniox_account_state')] + + +def test_session_scoped_typed_deaths_do_not_bench_the_provider(): + """Idle-timeout and rotation are evidence about one session, not the provider.""" + calls, opener = _circuit_recorder() + with patch('utils.stt.streaming.open_provider_selection_circuit', side_effect=opener): + assert ( + note_typed_provider_death(SimpleNamespace(typed_death_reason=SONIOX_DEATH_IDLE_TIMEOUT), 'soniox') is False + ) + assert note_typed_provider_death(SimpleNamespace(typed_death_reason=SONIOX_DEATH_ROTATION), 'soniox') is False + assert note_typed_provider_death(SimpleNamespace(typed_death_reason=None), 'soniox') is False + assert calls == [] + + +@pytest.mark.asyncio +async def test_terminate_with_account_state_opens_the_circuit_and_reports_the_typed_reason(monkeypatch): + websocket = AsyncMock() + session = SimpleNamespace(active=True, stt_terminal_failure=False, close_code=1001, live_transcription_attempt=None) + failure = TranscriptionFailure(TranscriptionOutcome.UPSTREAM_ERROR, provider='soniox', retryable=True) + recorded = [] + monkeypatch.setattr(live_failure, 'record_live_stt_failure', lambda **labels: recorded.append(labels)) + calls, opener = _circuit_recorder() + + with patch('utils.stt.streaming.open_provider_selection_circuit', side_effect=opener): + sent = await terminate_live_stt_session( + websocket, session, failure=failure, reason='soniox_account_state', platform='ios' + ) + + assert sent is True + assert session.stt_terminal_failure is True + assert session.close_code == LIVE_STT_FAILURE_CLOSE_CODE + event = websocket.send_json.await_args.args[0] + assert event['reason'] == 'soniox_account_state' + assert event['provider'] == 'soniox' + assert recorded[0]['phase'] == 'connection' + assert calls == [('soniox', 'soniox_account_state')] + + +@pytest.mark.asyncio +async def test_terminate_with_a_session_scoped_reason_does_not_open_the_circuit(monkeypatch): + websocket = AsyncMock() + session = SimpleNamespace(active=True, stt_terminal_failure=False, close_code=1001) + monkeypatch.setattr(live_failure, 'record_live_stt_failure', lambda **labels: None) + calls, opener = _circuit_recorder() + + with patch('utils.stt.streaming.open_provider_selection_circuit', side_effect=opener): + await terminate_live_stt_session( + websocket, + session, + failure=live_failure.live_stt_upstream_failure('soniox'), + reason='soniox_idle_timeout', + platform='ios', + ) + + assert calls == [] + assert websocket.send_json.await_args.args[0]['reason'] == 'soniox_idle_timeout' + + +@pytest.mark.asyncio +async def test_the_audio_send_path_terminates_with_the_typed_reason(): + """The send path is often the first observer of death; it must not collapse the type.""" + + class DeadTypedSocket: + is_connection_dead = True + typed_death_reason = 'soniox_account_state' + + def send(self, _audio): + raise AssertionError('a dead socket must not be sent to') + + websocket = AsyncMock() + session = SimpleNamespace(active=True, stt_terminal_failure=False, close_code=1001) + with patch.object(live_failure, 'record_live_stt_failure'): + sent = await send_live_stt_audio( + websocket, + session, + stt_socket=DeadTypedSocket(), + audio=b'\x01\x00', + provider='soniox', + platform='ios', + ) + + assert sent is False + assert session.stt_terminal_failure is True + assert websocket.send_json.await_args.args[0]['reason'] == 'soniox_account_state' + + +# --------------------------------------------------------------------------- +# The failover seam: a session that SURVIVES a 402 still teaches selection. +# --------------------------------------------------------------------------- + + +class TypedFakeSocket: + def __init__(self, *, dead: bool, typed: str | None = None): + self._dead = dead + self.typed_death_reason = typed + self.finished = False + self.sent = [] + + @property + def is_connection_dead(self) -> bool: + return self._dead + + def send(self, audio: bytes) -> bool: + if self._dead: + return False + self.sent.append(audio) + return True + + def finish(self) -> None: + self.finished = True + + +def _receiver_with_dead_soniox(replacement): + host = MagicMock() + host.is_multi_channel = False + host.use_custom_stt = False + host.state.active = True + host.state.stt_terminal_failure = False + host.language = 'en' + host.multi_lang_enabled = True + host.stt_service = STTService.soniox + + receiver = ListenReceiver(host, [], {}) + receiver.stt_socket = TypedFakeSocket(dead=True, typed=SONIOX_DEATH_ACCOUNT_STATE) + receiver.vad_gate = None + receiver._stt_rebuild = (lambda _s: None, lambda _s: None, 16000) + receiver._create_stt_socket = AsyncMock(return_value=replacement) + return receiver + + +@pytest.mark.asyncio +async def test_failover_on_a_typed_402_death_opens_the_selection_circuit(): + """The session survives on the next provider; the circuit must still learn. + + This is the invisible-outage seam: without it, a balance-exhausted provider + accepts every connect, refuses every stream, and selection keeps choosing + it because each dying session's replacement connects successfully. + """ + healthy = TypedFakeSocket(dead=False) + receiver = _receiver_with_dead_soniox(healthy) + calls, opener = _circuit_recorder() + + with patch( + 'routers.listen.receiver.get_stt_service_for_language', + return_value=(STTService.deepgram, 'en', 'dg-nova-3'), + ), patch('utils.stt.streaming.open_provider_selection_circuit', side_effect=opener): + assert await receiver._failover_stt_socket() is True + + assert receiver.stt_socket is healthy + assert calls == [('soniox', 'soniox_account_state')] + + +@pytest.mark.asyncio +async def test_failover_on_an_untyped_death_leaves_the_circuit_alone(): + receiver = _receiver_with_dead_soniox(TypedFakeSocket(dead=False)) + receiver.stt_socket = TypedFakeSocket(dead=True, typed=None) + calls, opener = _circuit_recorder() + + with patch( + 'routers.listen.receiver.get_stt_service_for_language', + return_value=(STTService.deepgram, 'en', 'dg-nova-3'), + ), patch('utils.stt.streaming.open_provider_selection_circuit', side_effect=opener): + assert await receiver._failover_stt_socket() is True + + assert calls == [] + + +@pytest.mark.asyncio +async def test_the_death_monitor_terminalizes_with_the_typed_reason(): + """The monitor's vantage point is only 'the socket died'; the socket knows why.""" + + async def wait(_seconds): + return True # end the monitor loop after one poll + + host = SimpleNamespace( + state=SimpleNamespace(active=True, stt_terminal_failure=False), + request=SimpleNamespace(websocket=object()), + client_device_context=SimpleNamespace(platform='ios'), + stt_service=STTService.soniox, + wait=wait, + ) + monitor_self = SimpleNamespace( + host=host, + stt_socket=SimpleNamespace(is_connection_dead=True, typed_death_reason='soniox_rotation'), + ) + monitor_self._serving_provider = lambda: ListenReceiver._serving_provider(monitor_self) + + async def _no_failover(): + return False + + monitor_self._failover_stt_socket = _no_failover + + with patch.object(receiver_mod, 'terminate_live_stt_session', new=AsyncMock()) as terminate: + await ListenReceiver._monitor_stt_death(monitor_self) + + assert terminate.await_args.kwargs['reason'] == 'soniox_rotation' + + +# --------------------------------------------------------------------------- +# Provider vocabulary and fallback-reason classification. +# --------------------------------------------------------------------------- + + +def test_live_provider_tokens_survive_bounded_provider(): + """Terminal-failure metrics used to report provider='unknown' for every soniox death.""" + assert bounded_provider('soniox') == 'soniox' + assert bounded_provider('deepgram_cloud') == 'deepgram_cloud' + assert bounded_provider('user-supplied-provider') == 'unknown' + + +def test_soniox_402_text_classifies_as_quota_not_provider_5xx(): + """Fallback telemetry filed a balance-exhausted account as a server fault.""" + assert ( + _fallback_failure_reason(RuntimeError('402 organization_balance_exhausted: Organization balance exhausted.')) + == 'quota' + ) + assert _fallback_failure_reason(RuntimeError('Internal server error')) == 'provider_5xx' diff --git a/backend/tests/unit/test_update_conversation_delete_race.py b/backend/tests/unit/test_update_conversation_delete_race.py new file mode 100644 index 00000000000..346bedb38c8 --- /dev/null +++ b/backend/tests/unit/test_update_conversation_delete_race.py @@ -0,0 +1,448 @@ +"""Regression test: update_conversation must honor its gone-owner contract when the +delete lands between its existence read and its commit. + +Production loop sensor (pusher, latest 30-min windows) recorded a NEW error +signature immediately after the ``list_audio_chunks`` NameError class (fixed in +#12439) stopped firing: + + ERROR:routers.pusher:Error updating audio files: 404 No document to update: + projects/based-hardware/databases/(default)/documents/users// + conversations/ + +``database.conversations.update_conversation`` documents that it "Returns False +when the conversation no longer exists, so callers that keep producing work for +it (e.g. the pusher's private-cloud audio sync) can stop instead of writing +into a deleted owner", and the pusher's private-cloud flush builds its designed +gone-owner path on exactly that return value (stop syncing, add to +``deleted_conversations``, release the audio budget, record the drop +fallback). But the function implemented the check-then-act non-atomically: +``get()`` → ``update()`` with no transaction. A conversation deleted in between +(the same mid-session empty-generation delete #11860 defends against) makes +``update()`` raise ``google.api_core.exceptions.NotFound`` (404 No document to +update), which escapes to each caller's broad ``except Exception`` as an ERROR +log — so the designed gone-owner path never runs and the pusher session keeps +producing audio work for a conversation that no longer exists. + +The suite never noticed because every existing test exercises the two +sequential states (exists → update; missing → False) — nothing deletes the +document *between* the read and the commit, which is the only interleaving the +contract is about. + +These tests drive the real ``update_conversation`` through a controllable seam +(``conversations_db.db``), with a fake Firestore whose document can vanish +between ``get()`` and ``update()`` — the exact production race — and then drive +the two highest-volume affected callers (the pusher flush and the sync +finalizer) through their real code to prove they take their designed +gone-owner paths instead of logging ERROR, while genuine failures still log. + +Failure-Class: new — the violated contract is ``update_conversation``'s own +documented gone-owner return: a delete racing the write must be reported as +False, not raised as NotFound. Instance fix at the authoritative owner (the +database layer), where every caller benefits without per-call-site exception +handling — the pattern AGENTS.md prescribes ("don't add another call-site +exception when ownership is the real problem"). +""" + +import asyncio +import logging +import struct +from collections import deque +from copy import deepcopy +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from fastapi.websockets import WebSocketDisconnect +from google.api_core.exceptions import NotFound +from starlette.websockets import WebSocketState + +import database.conversations as conversations_db + +UID = 'uid' +CONVERSATION_ID = 'conversation' +CONVERSATION_PATH = ('users', UID, 'conversations', CONVERSATION_ID) + + +# --------------------------------------------------------------------------- +# Fake Firestore with a controllable mid-flight delete — the production race. +# --------------------------------------------------------------------------- + + +class _RaceDocumentRef: + """Document reference whose update() raises NotFound like real Firestore. + + Real Firestore ``update()`` on a missing document raises + ``google.api_core.exceptions.NotFound("404 No document to update: ...")`` — + the exact production message. ``get()`` may still see the snapshot if the + delete commits after the read, which is the race window being tested. + """ + + def __init__(self, store: '_RaceFirestore', path: tuple): + self._store = store + self._path = path + + def collection(self, name: str) -> '_CollectionRef': + return _CollectionRef(self._store, self._path + (name,)) + + def get(self): + documents = self._store.documents + if self._path not in documents: + return SimpleNamespace(exists=False, to_dict=lambda: None) + data = deepcopy(documents[self._path]) + return SimpleNamespace(exists=True, to_dict=lambda: data) + + def update(self, updates: dict): + documents = self._store.documents + if self._path in self._store.delete_on_get: + # The delete commits now: the document vanishes before this write + # lands, exactly like the production race. + documents.pop(self._path, None) + if self._path not in documents: + raise NotFound(f'404 No document to update: {"/".join(self._path)}') + documents[self._path].update(deepcopy(updates)) + + +class _CollectionRef: + def __init__(self, store: '_RaceFirestore', path: tuple): + self._store = store + self._path = path + + def document(self, document_id: str) -> '_RaceDocumentRef': + return _RaceDocumentRef(self._store, self._path + (document_id,)) + + +class _RaceFirestore: + """``conversations_db.db`` double that can delete mid-flight. + + ``delete_on_get`` holds the document paths that vanish between ``get()`` + and ``update()`` — the production race. The snapshot read still sees the + document; the commit raises NotFound with the production message. + """ + + def __init__(self, documents: dict = None): + self.documents = documents if documents is not None else {} + self.delete_on_get: set = set() + + def collection(self, name: str) -> '_CollectionRef': + return _CollectionRef(self, (name,)) + + +def _fake_store(monkeypatch, **conversation) -> _RaceFirestore: + store = _RaceFirestore({CONVERSATION_PATH: dict(conversation)}) + monkeypatch.setattr(conversations_db, 'db', store) + return store + + +# --------------------------------------------------------------------------- +# The contract: a delete between read and commit is False, not NotFound. +# --------------------------------------------------------------------------- + + +def test_update_raced_by_delete_returns_false_instead_of_raising(monkeypatch): + """Before the fix this raised NotFound('404 No document to update'). + + The document exists at the existence read and is gone at the commit — + the exact interleaving the gone-owner contract governs. + """ + store = _fake_store(monkeypatch, data_protection_level='standard', language='en') + store.delete_on_get.add(CONVERSATION_PATH) + + result = conversations_db.update_conversation(UID, CONVERSATION_ID, {'language': 'fr'}) + + assert result is False + + +def test_update_missing_document_returns_false(monkeypatch): + """Control: a document that never existed is the already-handled case.""" + store = _fake_store(monkeypatch) + del store.documents[CONVERSATION_PATH] + + assert conversations_db.update_conversation(UID, CONVERSATION_ID, {'language': 'fr'}) is False + + +def test_update_live_document_still_writes_and_returns_true(monkeypatch): + """Control: no race — the write lands exactly as before the fix.""" + store = _fake_store(monkeypatch, data_protection_level='standard', language='en') + + result = conversations_db.update_conversation(UID, CONVERSATION_ID, {'language': 'fr'}) + + assert result is True + assert store.documents[CONVERSATION_PATH]['language'] == 'fr' + assert store.documents[CONVERSATION_PATH]['data_protection_level'] == 'standard' + + +def test_update_uses_the_snapshot_data_protection_level(monkeypatch): + """The snapshot's data_protection_level still reaches the write prep. + + ``_prepare_conversation_for_write`` is driven for real; the enhanced level + from the snapshot must not block the write path. + """ + _fake_store(monkeypatch, data_protection_level='enhanced') + + assert conversations_db.update_conversation(UID, CONVERSATION_ID, {'language': 'fr'}) is True + + +def test_update_raced_by_delete_does_not_write_partial_data(monkeypatch): + """Nothing from update_data may land when the owner is gone.""" + store = _fake_store(monkeypatch, data_protection_level='standard', language='en') + store.delete_on_get.add(CONVERSATION_PATH) + + conversations_db.update_conversation(UID, CONVERSATION_ID, {'language': 'fr'}) + + assert store.documents == {} + + +def test_update_race_returns_false_for_every_caller_write_shape(monkeypatch): + """The contract holds for the plain field writes each caller makes — + the pusher flush (audio_files), the events/action_items wrappers + (structured.*), the calendar link (calendar_event), the deferral flag.""" + for payload in ( + {'audio_files': [{'path': 'audio.wav'}]}, + {'structured.events': []}, + {'structured.action_items': []}, + {'calendar_event': None}, + {'deferred': True}, + ): + store = _fake_store(monkeypatch, data_protection_level='standard') + store.delete_on_get.add(CONVERSATION_PATH) + assert conversations_db.update_conversation(UID, CONVERSATION_ID, dict(payload)) is False + + +def test_lifecycle_fields_are_still_rejected(monkeypatch): + """The lifecycle-owner guard precedes everything, race or not.""" + _fake_store(monkeypatch, data_protection_level='standard') + with pytest.raises(ValueError, match='lifecycle fields'): + conversations_db.update_conversation(UID, CONVERSATION_ID, {'status': 'processing'}) + + +def test_lifecycle_fields_rejected_even_when_raced_by_delete(monkeypatch): + """Guard ordering is load-bearing: it must not depend on Firestore state.""" + store = _fake_store(monkeypatch, data_protection_level='standard') + store.delete_on_get.add(CONVERSATION_PATH) + with pytest.raises(ValueError, match='lifecycle fields'): + conversations_db.update_conversation(UID, CONVERSATION_ID, {'discarded': True}) + + +def test_update_conversation_events_inherits_the_race_contract(monkeypatch): + """The events wrapper routes through update_conversation, so the raced + owner is a no-op instead of a NotFound escaping into its caller.""" + store = _fake_store(monkeypatch, data_protection_level='standard', structured={'events': []}) + store.delete_on_get.add(CONVERSATION_PATH) + + conversations_db.update_conversation_events(UID, CONVERSATION_ID, [{'type': 'app'}]) + + assert store.documents == {} + + +def test_update_conversation_action_items_inherits_the_race_contract(monkeypatch): + """The action_items wrapper inherits the same contract.""" + store = _fake_store(monkeypatch, data_protection_level='standard') + store.delete_on_get.add(CONVERSATION_PATH) + + conversations_db.update_conversation_action_items(UID, CONVERSATION_ID, [{'title': 't'}]) + + assert store.documents == {} + + +# --------------------------------------------------------------------------- +# Caller 1: the pusher's private-cloud flush (the signature's birthplace). +# --------------------------------------------------------------------------- + +SAMPLE_RATE = 8000 +CONVERSATION_FRAME = struct.pack(' dict: + return {'path': 'audio.wav'} + + +class SequencedWebSocket: + """Sends the second audio chunk only after the first flush was handled. + + Without this handshake the two chunks coalesce into one batch and the test + cannot observe whether the pusher keeps uploading after the owner died. + """ + + def __init__(self, first_flush_handled: asyncio.Event): + self.frames = deque([CONVERSATION_FRAME, AUDIO_FRAME]) + self.first_flush_handled = first_flush_handled + self.second_chunk_sent = False + self.client_state = WebSocketState.CONNECTED + self.close_code = None + + async def accept(self): + return None + + async def close(self, code=1000, reason=None): + self.close_code = code + self.client_state = WebSocketState.DISCONNECTED + + async def receive_bytes(self): + if self.frames: + return self.frames.popleft() + if not self.second_chunk_sent: + await self.first_flush_handled.wait() + self.second_chunk_sent = True + return AUDIO_FRAME + raise WebSocketDisconnect(1000) + + +@pytest.fixture +def raced_pusher_session(monkeypatch): + """Drive the pusher's private-cloud worker where update_conversation hits + the read-then-delete race (False) on the first flush.""" + import routers.pusher as pusher + import utils.pusher_protocol as pusher_protocol + + monkeypatch.setattr(pusher, 'get_audio_bytes_webhook_seconds', lambda uid: None) + monkeypatch.setattr(pusher, 'is_audio_bytes_app_enabled', lambda uid: False) + monkeypatch.setattr(pusher, 'is_audio_merge_dispatch_enabled', lambda: False) + monkeypatch.setattr(pusher.users_db, 'get_user_private_cloud_sync_enabled', lambda uid: True) + monkeypatch.setattr(pusher.users_db, 'get_data_protection_level', lambda uid: 'standard') + monkeypatch.setattr(pusher, 'PUSHER_ACTIVE_WS_CONNECTIONS', MagicMock()) + monkeypatch.setattr(pusher, 'PUSHER_PRIVATE_CLOUD_UPLOAD_DROPS', MagicMock()) + monkeypatch.setattr(pusher_protocol, 'PUSHER_QUEUE_DROPS', MagicMock()) + monkeypatch.setattr(pusher_protocol, 'PUSHER_QUEUE_DROPPED_BYTES', MagicMock()) + monkeypatch.setattr(pusher, 'PRIVATE_CLOUD_CHUNK_DURATION', 0.0) + monkeypatch.setattr(pusher, 'PRIVATE_CLOUD_BATCH_MAX_AGE', 0.0) + monkeypatch.setattr(pusher, 'PRIVATE_CLOUD_SYNC_PROCESS_INTERVAL', 0.01) + + first_flush_handled = asyncio.Event() + uploaded: list[str] = [] + fallbacks: list[dict] = [] + + def upload(chunks, uid, conversation_id, protection_level): + uploaded.append(conversation_id) + + def run(update_result: bool): + def update_conversation(uid, conversation_id, update_data): + first_flush_handled.set() + return update_result + + monkeypatch.setattr(pusher, 'upload_audio_chunks_batch', upload) + monkeypatch.setattr( + pusher.conversations_db, 'create_audio_files_from_chunks', lambda uid, conversation_id: [_AudioFile()] + ) + monkeypatch.setattr(pusher.conversations_db, 'update_conversation', update_conversation) + monkeypatch.setattr(pusher, 'record_fallback', lambda **kwargs: fallbacks.append(kwargs)) + return SequencedWebSocket(first_flush_handled) + + return {'run': run, 'uploaded': uploaded, 'fallbacks': fallbacks} + + +@pytest.mark.asyncio +async def test_pusher_flush_raced_by_delete_takes_the_gone_owner_path(raced_pusher_session, caplog): + """Before the fix, the race surfaced as the production ERROR signature. + + The pusher flush called update_conversation, the delete won, NotFound + escaped to the broad except, and the session logged + ``Error updating audio files: 404 No document to update`` — the designed + gone-owner path (stop syncing, drop fallback) never ran. update_conversation + returning False (the fix's contract) must drive that path silently. + """ + import routers.pusher as pusher_mod + + websocket = raced_pusher_session['run'](update_result=False) + + with caplog.at_level(logging.ERROR, logger='routers.pusher'): + await pusher_mod._websocket_util_trigger(websocket, UID, SAMPLE_RATE) + + # The first batch races the delete and is unavoidable; the second must not upload. + assert raced_pusher_session['uploaded'] == [CONVERSATION_ID] + assert raced_pusher_session['fallbacks'] == [ + { + 'component': 'pusher', + 'from_mode': 'private_cloud_sync', + 'to_mode': 'drop', + 'reason': 'policy', + 'outcome': 'exhausted', + 'log': pusher_mod.logger, + } + ] + # The race must no longer surface as an ERROR log. + assert not [r for r in caplog.records if 'Error updating audio files' in r.getMessage()] + + +@pytest.mark.asyncio +async def test_pusher_flush_still_errors_on_a_real_storage_failure(raced_pusher_session, caplog): + """Guard against over-broad suppression: a genuine (non-race) failure in + the flush must still be logged at ERROR, not silently dropped. + + The first flush completes normally (it arms the sequenced websocket); the + second flush's chunk listing raises. + """ + import routers.pusher as pusher_mod + + websocket = raced_pusher_session['run'](update_result=True) + + calls = {'n': 0} + + def flaky_create(uid, conversation_id): + calls['n'] += 1 + if calls['n'] == 1: + return [_AudioFile()] + raise RuntimeError('storage exploded') + + mp = pytest.MonkeyPatch() + mp.setattr(pusher_mod.conversations_db, 'create_audio_files_from_chunks', flaky_create) + try: + with caplog.at_level(logging.ERROR, logger='routers.pusher'): + await pusher_mod._websocket_util_trigger(websocket, UID, SAMPLE_RATE) + assert [r for r in caplog.records if 'Error updating audio files' in r.getMessage()], ( + 'a genuine flush failure must still be logged at ERROR' + ) + finally: + mp.undo() + + +# --------------------------------------------------------------------------- +# Caller 2: the sync pipeline's offline finalizer. +# --------------------------------------------------------------------------- + + +def test_sync_finalizer_raced_by_delete_is_not_an_error(monkeypatch, caplog): + """``_finalize_sync_audio_files`` must not log outcome=failed for the race. + + The finalizer persists audio_files per conversation with a broad except + that logs ``event=sync_audio_finalize outcome=failed`` — before the fix the + gone-owner race landed in that ERROR. After the fix the False return makes + the persist a no-op (the conversation is gone, there is nothing to persist + onto) and the ERROR signature disappears. + """ + from utils.sync import pipeline as sync_pipeline + + def update_conversation(uid, conversation_id, update_data): + return False # the race outcome after the fix + + monkeypatch.setattr( + sync_pipeline.conversations_db, 'create_audio_files_from_chunks', lambda uid, cid: [_AudioFile()] + ) + monkeypatch.setattr(sync_pipeline.conversations_db, 'update_conversation', update_conversation) + monkeypatch.setattr(sync_pipeline, 'precache_conversation_audio', lambda *a, **k: None) + monkeypatch.setattr(sync_pipeline, 'is_audio_merge_dispatch_enabled', lambda: False) + + response = {'new_memories': {CONVERSATION_ID}, 'updated_memories': set()} + with caplog.at_level(logging.ERROR, logger='utils.sync.pipeline'): + sync_pipeline._finalize_sync_audio_files(UID, response) + + assert not [r for r in caplog.records if 'sync_audio_finalize' in r.getMessage()] + + +def test_sync_finalizer_still_errors_on_a_real_failure(monkeypatch, caplog): + """Over-broad suppression guard: a genuine finalizer failure still logs.""" + from utils.sync import pipeline as sync_pipeline + + def boom(uid, cid): + raise RuntimeError('storage exploded') + + monkeypatch.setattr(sync_pipeline.conversations_db, 'create_audio_files_from_chunks', boom) + monkeypatch.setattr(sync_pipeline, 'precache_conversation_audio', lambda *a, **k: None) + monkeypatch.setattr(sync_pipeline, 'is_audio_merge_dispatch_enabled', lambda: False) + + response = {'new_memories': {CONVERSATION_ID}, 'updated_memories': set()} + with caplog.at_level(logging.ERROR, logger='utils.sync.pipeline'): + sync_pipeline._finalize_sync_audio_files(UID, response) + + assert [r for r in caplog.records if 'sync_audio_finalize' in r.getMessage()] diff --git a/backend/tests/unit/test_ws_auth_rejection_logging.py b/backend/tests/unit/test_ws_auth_rejection_logging.py new file mode 100644 index 00000000000..96e6512e647 --- /dev/null +++ b/backend/tests/unit/test_ws_auth_rejection_logging.py @@ -0,0 +1,404 @@ +"""WS auth rejection logging: severity must follow fault origin. + +Production evidence (backend-listen, GCP 2026-08-30/31, Loop S sensor): +``ERROR:utils.other.endpoints:WebSocket auth failed: code=4001 error=Token +expired, …`` and ``… error=Certificate for key id 6ac9047f… not found.`` were +the #2 and #3 error signatures, ×9–47 and ×1–34 per 30-min window, for 16 +consecutive hours. Root cause is client-side in both: the expired cohort +presents Firebase ID tokens hours past ``exp`` (samples show 3.7h/7.5h/18h +stale), and the certificate cohort presents tokens signed by a key id that is +absent from Google's *currently served* x509 set (verified directly against +https://www.googleapis.com/robot/v1/metadata/x509/securetoken@system.gserviceaccount.com +on 2026-08-31 — the kid is retired, so a refresh cannot resurrect it). The +server's rejection — and the close code it sends — is the protocol working. + +The defect was on the server side of the boundary: every expected, client- +caused rejection was logged at ERROR, making the stale-client reconnect +population indistinguishable from a serving outage in the error feed that +pages humans (the same feed in which a real outage — the Modulate STT 5xx +storm — had to be found). + +Failure-Class: FC-request-input-rejection-escapes-as-server-fault — instance +fix in the class canonized by #11853 ("a route owns the classification of its +own request input"). Here the violated contract is the WS auth boundary's: +a client-caused token rejection (InvalidIdTokenError — Firebase evaluated the +client's token and refused it) must be a warning, while a server fault +(CertificateFetchError — we could not even fetch Google's certs to evaluate +the token; any unexpected error) stays an error. Close codes and reasons are +unchanged — clients already receive the correct remediation hint; only the +severity classification is fixed. + +These tests drive the REAL ``_verify_ws_auth`` boundary and the REAL FastAPI +dependencies (``get_current_user_uid_ws_listen`` / ``get_current_user_uid_ws``) +through the sanctioned ``stub_modules`` isolation pattern used by +``test_ws_auth_handshake.py``, asserting on captured log records and the wire +close codes the client receives. +""" + +import importlib +import sys +import types +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import Depends, FastAPI, WebSocket +from fastapi.testclient import TestClient +from starlette.websockets import WebSocketDisconnect + +from testing.import_isolation import stub_modules + +BACKEND_DIR = Path(__file__).resolve().parents[2] + +# The production error samples these tests replay, verbatim shapes. +EXPIRED_MESSAGE = "Token expired, 1788101015 < 1788114388" +RETIRED_KEY_MESSAGE = ( + "Certificate for key id 6ac9047f6712fcd5cf67a3307941d9fa42283955 not found." +) + + +# Firebase auth exception classes. The hierarchy mirrors the real firebase_admin +# (_auth_utils): ExpiredIdTokenError and RevokedIdTokenError are SUBCLASSES of +# InvalidIdTokenError, and CertificateFetchError is a sibling. Defined at module +# scope so @patch decorators evaluated at class-definition time can reference +# them. The module-scoped autouse fixture installs these same objects onto the +# firebase_admin.auth stub, preserving isinstance identity (both the except +# clause and _get_ws_auth_close's isinstance chain) inside utils.other.endpoints. +class InvalidIdTokenError(Exception): + pass + + +class ExpiredIdTokenError(InvalidIdTokenError): + pass + + +class RevokedIdTokenError(InvalidIdTokenError): + pass + + +class CertificateFetchError(Exception): + pass + + +# Populated by the _ws_auth_rejection_logging_isolation module fixture; tests +# resolve them at call time (after the fixture has run). +_verify_ws_auth = None +get_current_user_uid_ws_listen = None +get_current_user_uid_ws = None +ENDPOINTS_LOGGER = "utils.other.endpoints" + + +def _build_fakes(): + """Build the namespace-package + firebase/database stub mapping for ``stub_modules``.""" + database_pkg = types.ModuleType("database") + database_pkg.__path__ = [str(BACKEND_DIR / "database")] + utils_pkg = types.ModuleType("utils") + utils_pkg.__path__ = [str(BACKEND_DIR / "utils")] + utils_other_pkg = types.ModuleType("utils.other") + utils_other_pkg.__path__ = [str(BACKEND_DIR / "utils" / "other")] + + firebase_admin_stub = types.ModuleType("firebase_admin") + firebase_auth_stub = types.ModuleType("firebase_admin.auth") + firebase_admin_stub.auth = firebase_auth_stub + for err_cls in (CertificateFetchError, ExpiredIdTokenError, InvalidIdTokenError, RevokedIdTokenError): + setattr(firebase_auth_stub, err_cls.__name__, err_cls) + firebase_auth_stub.verify_id_token = MagicMock(side_effect=InvalidIdTokenError("Invalid token")) + firebase_auth_stub.get_user = MagicMock() + + database_client_stub = types.ModuleType("database._client") + database_client_stub.db = MagicMock() + database_client_stub.document_id_from_seed = MagicMock(return_value="doc-id") + + database_redis_stub = types.ModuleType("database.redis_db") + database_redis_stub.check_rate_limit = MagicMock(return_value=True) + database_redis_stub.try_acquire_listen_lock = MagicMock(return_value=True) + database_redis_stub.try_acquire_user_platform_write_lock = MagicMock(return_value=True) + + users_stub = types.ModuleType("database.users") + users_stub.record_user_platform = MagicMock() + users_stub.record_client_device = MagicMock() + users_stub.get_user_deletion_wipe_status = MagicMock(return_value=None) + + fakes = { + "database": database_pkg, + "utils": utils_pkg, + "utils.other": utils_other_pkg, + "firebase_admin": firebase_admin_stub, + "firebase_admin.auth": firebase_auth_stub, + "database._client": database_client_stub, + "database.redis_db": database_redis_stub, + "database.users": users_stub, + # Pop polluted/prior copies so endpoints re-execs against these fakes; + # stub_modules restores/purges them on teardown (same rationale as + # test_ws_auth_handshake.py). + "utils.executors": None, + "utils.other.endpoints": None, + } + return fakes + + +@pytest.fixture(scope="module", autouse=True) +def _ws_auth_rejection_logging_isolation(): + """Install the stubs and exec utils.other.endpoints against them.""" + with stub_modules(_build_fakes()): + endpoints = importlib.import_module("utils.other.endpoints") + mod = sys.modules[__name__] + mod._verify_ws_auth = endpoints._verify_ws_auth + mod.get_current_user_uid_ws_listen = endpoints.get_current_user_uid_ws_listen + mod.get_current_user_uid_ws = endpoints.get_current_user_uid_ws + yield + + +class _SeverityAssertions(unittest.TestCase): + """Shared assertions on captured log records.""" + + def _assert_client_rejection_warns(self, captured, expected_code): + errors = [r for r in captured.records if r.levelname == "ERROR"] + self.assertEqual( + len(errors), 0, + f"client-caused rejection must not log at ERROR, got: {[r.getMessage() for r in errors]}", + ) + self.assertEqual(len(captured.records), 1, f"expected exactly one log record, got {len(captured.records)}") + record = captured.records[0] + self.assertEqual(record.levelname, "WARNING") + self.assertIn(f"code={expected_code}", record.getMessage()) + + def _assert_server_fault_errors(self, captured, expected_code): + self.assertTrue( + any(r.levelname == "ERROR" for r in captured.records), + f"server fault must log at ERROR, got: {[(r.levelname, r.getMessage()) for r in captured.records]}", + ) + self.assertIn(f"code={expected_code}", captured.records[0].getMessage()) + + +class TestRejectionSeverityAtTheBoundary(_SeverityAssertions): + """Direct calls to the real _verify_ws_auth: severity follows fault origin.""" + + def test_expired_token_rejection_logs_warning_not_error(self): + """The ×47/30m 'Token expired' signature is a client fault -> WARNING.""" + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=InvalidIdTokenError(EXPIRED_MESSAGE)): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(Exception): + _verify_ws_auth("Bearer expired-token") + self._assert_client_rejection_warns(captured, expected_code=4001) + self.assertIn(EXPIRED_MESSAGE, captured.records[0].getMessage()) + + def test_retired_signing_key_rejection_logs_warning_not_error(self): + """The ×34/30m 'Certificate for key id … not found' signature is a client + presenting a token signed by a key Google retired -> WARNING.""" + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=InvalidIdTokenError(RETIRED_KEY_MESSAGE)): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(Exception): + _verify_ws_auth("Bearer retired-key-token") + self._assert_client_rejection_warns(captured, expected_code=4001) + self.assertIn("6ac9047f6712fcd5cf67a3307941d9fa42283955", captured.records[0].getMessage()) + + def test_typed_expired_token_error_logs_warning(self): + """ExpiredIdTokenError (typed, not message-derived) is client-caused -> WARNING.""" + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=ExpiredIdTokenError(EXPIRED_MESSAGE)): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(Exception): + _verify_ws_auth("Bearer typed-expired-token") + self._assert_client_rejection_warns(captured, expected_code=4001) + + def test_revoked_token_rejection_logs_warning(self): + """Revoked token: the client must re-login (4004) — still a client fault.""" + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=RevokedIdTokenError("Token revoked")): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(Exception): + _verify_ws_auth("Bearer revoked-token") + self._assert_client_rejection_warns(captured, expected_code=4004) + + def test_generic_invalid_token_rejection_logs_warning(self): + """Any other InvalidIdTokenError (audience, malformed…) is client-caused.""" + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=InvalidIdTokenError("Firebase ID token has incorrect \"aud\" claim")): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(Exception): + _verify_ws_auth("Bearer wrong-audience-token") + self._assert_client_rejection_warns(captured, expected_code=1008) + + def test_certificate_fetch_failure_stays_error(self): + """CertificateFetchError = the SERVER could not fetch Google's certs to + evaluate the token — a genuine server fault, must stay at ERROR.""" + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=CertificateFetchError( + "Could not fetch certificates", RuntimeError("network unavailable"))): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(Exception): + _verify_ws_auth("Bearer any-token") + self._assert_server_fault_errors(captured, expected_code=4001) + + def test_message_derived_certificate_rejection_is_warning_not_error(self): + """An InvalidIdTokenError whose *message* mentions a certificate (retired + key id — client-caused) must be classified differently from a typed + CertificateFetchError (server-caused), even though both close with 4001.""" + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=InvalidIdTokenError("Could not verify token: certificate problems")): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(Exception): + _verify_ws_auth("Bearer message-cert-token") + errors = [r for r in captured.records if r.levelname == "ERROR"] + self.assertEqual(errors, [], "message-derived certificate rejection is client-caused, not a server fault") + + +class TestSeverityThroughListenDep(_SeverityAssertions): + """The /v4/listen dependency path: severity fixed, wire contract unchanged.""" + + def setUp(self): + self.app = FastAPI() + + @self.app.websocket("/ws-listen") + async def ws_listen(websocket: WebSocket, uid: str = Depends(get_current_user_uid_ws_listen)): + await websocket.accept() + await websocket.send_json({"uid": uid}) + await websocket.close() + + self.client = TestClient(self.app) + + def test_expired_token_close_4001_and_warning(self): + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=InvalidIdTokenError(EXPIRED_MESSAGE)): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(WebSocketDisconnect) as ctx: + with self.client.websocket_connect( + "/ws-listen", headers={"Authorization": "Bearer expired_token"} + ): + self.fail("Expected WebSocket to be closed by server") + self.assertEqual(ctx.exception.code, 4001, "client-visible close code must stay 4001 (refresh token)") + self._assert_client_rejection_warns(captured, expected_code=4001) + + def test_retired_key_close_4001_and_warning(self): + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=InvalidIdTokenError(RETIRED_KEY_MESSAGE)): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(WebSocketDisconnect) as ctx: + with self.client.websocket_connect( + "/ws-listen", headers={"Authorization": "Bearer stale_key_token"} + ): + self.fail("Expected WebSocket to be closed by server") + self.assertEqual(ctx.exception.code, 4001, "client-visible close code must stay 4001 (refresh token)") + self._assert_client_rejection_warns(captured, expected_code=4001) + + def test_revoked_token_close_4004_and_warning(self): + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=RevokedIdTokenError("Token revoked")): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(WebSocketDisconnect) as ctx: + with self.client.websocket_connect( + "/ws-listen", headers={"Authorization": "Bearer revoked_token"} + ): + self.fail("Expected WebSocket to be closed by server") + self.assertEqual(ctx.exception.code, 4004, "client-visible close code must stay 4004 (re-login)") + self._assert_client_rejection_warns(captured, expected_code=4004) + + def test_certificate_fetch_failure_close_4001_and_error(self): + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=CertificateFetchError( + "Could not fetch certificates", RuntimeError("network unavailable"))): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(WebSocketDisconnect) as ctx: + with self.client.websocket_connect( + "/ws-listen", headers={"Authorization": "Bearer cert_fetch_token"} + ): + self.fail("Expected WebSocket to be closed by server") + self.assertEqual(ctx.exception.code, 4001, "client-visible close code must stay 4001") + self._assert_server_fault_errors(captured, expected_code=4001) + + def test_unexpected_verify_error_stays_error_close_1008(self): + """The generic handler (unexpected exception) is a server fault: untouched + by the reclassification — still ERROR, still close 1008.""" + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=RuntimeError("unexpected error")): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(WebSocketDisconnect) as ctx: + with self.client.websocket_connect( + "/ws-listen", headers={"Authorization": "Bearer token"} + ): + self.fail("Expected connection to fail") + self.assertEqual(ctx.exception.code, 1008) + errors = [r for r in captured.records if r.levelname == "ERROR"] + self.assertTrue(errors, "unexpected errors must keep logging at ERROR") + + +class TestSeverityThroughRateLimitedDep(_SeverityAssertions): + """The rate-limited WS dependency funnels through the same boundary.""" + + def setUp(self): + self.app = FastAPI() + + @self.app.websocket("/ws-ratelimited") + async def ws_ratelimited(websocket: WebSocket, uid: str = Depends(get_current_user_uid_ws)): + await websocket.accept() + await websocket.send_json({"uid": uid}) + await websocket.close() + + self.client = TestClient(self.app) + + def test_expired_token_close_4001_and_warning(self): + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=InvalidIdTokenError(EXPIRED_MESSAGE)): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(WebSocketDisconnect) as ctx: + with self.client.websocket_connect( + "/ws-ratelimited", headers={"Authorization": "Bearer expired_token"} + ): + self.fail("Expected WebSocket to be closed by server") + self.assertEqual(ctx.exception.code, 4001) + self._assert_client_rejection_warns(captured, expected_code=4001) + + def test_certificate_fetch_failure_stays_error(self): + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=CertificateFetchError( + "Could not fetch certificates", RuntimeError("network unavailable"))): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + with self.assertRaises(WebSocketDisconnect) as ctx: + with self.client.websocket_connect( + "/ws-ratelimited", headers={"Authorization": "Bearer cert_fetch_token"} + ): + self.fail("Expected WebSocket to be closed by server") + self.assertEqual(ctx.exception.code, 4001) + self._assert_server_fault_errors(captured, expected_code=4001) + + +class TestErrorFeedPollutionContract(_SeverityAssertions): + """The incident restated as a contract: a burst of client-caused rejections + must leave the ERROR feed empty, while server faults land in it.""" + + def test_client_rejection_burst_leaves_error_feed_empty(self): + rejections = [ + InvalidIdTokenError(EXPIRED_MESSAGE), + InvalidIdTokenError(RETIRED_KEY_MESSAGE), + RevokedIdTokenError("Token revoked"), + ] + with patch.object(sys.modules["utils.other.endpoints"], "verify_token", + side_effect=rejections): + with self.assertLogs(ENDPOINTS_LOGGER, level="WARNING") as captured: + for _ in rejections: + with self.assertRaises(Exception): + _verify_ws_auth("Bearer stale-token") + self.assertEqual( + len(captured.records), len(rejections), + f"each rejection logs exactly once, got {len(captured.records)}", + ) + errors = [r for r in captured.records if r.levelname == "ERROR"] + self.assertEqual( + errors, [], + f"the stale-client population must not pollute the ERROR feed: {[r.getMessage() for r in errors]}", + ) + + def test_missing_auth_header_short_circuits_before_rejection_logging(self): + """No Authorization header -> close 1008 without any rejection log: the + classifier owns token-evaluation rejections only, not missing-input ones.""" + with self.assertNoLogs(ENDPOINTS_LOGGER, level="WARNING"): + with self.assertRaises(Exception): + _verify_ws_auth(None) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/utils/aac.py b/backend/utils/aac.py index cbc5ca20e75..ea93a9b177a 100644 --- a/backend/utils/aac.py +++ b/backend/utils/aac.py @@ -1,4 +1,5 @@ import logging +import threading from typing import Any, List import av @@ -10,6 +11,45 @@ av.logging.set_level(av.logging.ERROR) # type: ignore[reportAttributeAccessIssue,reportUnknownMemberType] # PyAV exposes logging dynamically +class AACDecodeError(Exception): + """A frame the AAC decoder rejected. + + The listen receiver's decode-failure contract (``_record_decode_failure``: + per-frame warning with the codec's own message, streak counter, and the + one-shot ``silent_mic`` fallback at ``DECODE_FAILURE_STREAK_ALERT``) hangs + off the decoder *raising*. ``opuslib`` raises ``OpusError``; before this + class existed, ``AACDecoder.decode`` caught ``av.AVError`` and returned + ``b''``, so a fully undecodable AAC stream recorded a whole session with + no transcript, no ring buffer, no fallback metric — and the only trace was + FFmpeg's context-free ``ERROR:libav.aac:Channel element … is not + allocated`` native line per frame (~130/30min in prod, Loop S sensor). + The message is the FFmpeg error verbatim (``Invalid data found when + processing input data``, ``Input buffer exhausted…``), the same + codec-owns-the-detail convention the opus path uses. + """ + + +# FFmpeg re-reports a rejected frame on the ``libav.aac`` Python logger while +# ``AACDecoder.decode`` is still on the stack. That native line is a duplicate +# of the failure this module turns into ``AACDecodeError`` — which the receiver +# reports with codec, type, payload size, streak, and the silent-mic metric — +# so it is dropped for the duration of our own decode call only. Outside this +# window the logger is untouched: other ``av`` users' native errors still flow. +# The flag is thread-local: one session's decode must not mute the native log +# of a concurrent session in the same process. +aac_decode_in_progress = threading.local() + + +class NativeDuplicateSuppressionFilter(logging.Filter): + """Drops FFmpeg's re-report of a frame our decode already rejected.""" + + def filter(self, record: logging.LogRecord) -> bool: + return not getattr(aac_decode_in_progress, 'active', False) + + +logging.getLogger('libav.aac').addFilter(NativeDuplicateSuppressionFilter()) + + class AACDecoder: def __init__(self, uid: str = '', session_id: str = '', sample_rate: int = 16000, channels: int = 1): @@ -17,7 +57,7 @@ def __init__(self, uid: str = '', session_id: str = '', sample_rate: int = 16000 self.session_id = session_id # Initialize codec context immediately - self.codec_context: Any = av.CodecContext.create('aac', 'r') # type: ignore[reportAttributeAccessIssue,reportUnknownMemberType] # PyAV CodecContext exposed dynamically + self.codec_context: Any = av.CodecContext.create('aac', 'r') # type: ignore[reportAttributeAccessIssue,reportUnknownMemberType] # PyAV exposes logging dynamically # Initialize resampler immediately target_layout = 'mono' if channels == 1 else 'stereo' @@ -28,21 +68,27 @@ def __init__(self, uid: str = '', session_id: str = '', sample_rate: int = 16000 ) def decode(self, aac_data: bytes) -> bytes: - """Decode AAC frame using persistent codec context. + """Decode one AAC frame with ADTS header into resampled PCM bytes. - Args: - aac_data: Complete AAC frame with ADTS header + Raises AACDecodeError when FFmpeg rejects the frame (corrupt, truncated, + or mid-stream desync — the shapes behind the prod ``libav.aac`` error + signatures). Callers that keep the socket alive on a dropped frame must + report the failure (the listen receiver does, via its decode-failure + streak); swallowing it here is what made corrupt AAC streams silent. - Returns: - PCM data as bytes + Returns b'' for benign no-output input: an empty payload, or a packet + the decoder accepted without emitting frames yet (encoder priming). """ if not aac_data: return b'' try: - # Create packet and decode - packet = av.Packet(aac_data) # type: ignore[reportArgumentType] # PyAV Packet accepts bytes at runtime - frames: List[Any] = self.codec_context.decode(packet) + aac_decode_in_progress.active = True + try: + packet = av.Packet(aac_data) # type: ignore[reportArgumentType] # PyAV Packet accepts bytes at runtime + frames: List[Any] = self.codec_context.decode(packet) + finally: + aac_decode_in_progress.active = False if not frames: return b'' @@ -59,9 +105,7 @@ def decode(self, aac_data: bytes) -> bytes: return b''.join(pcm_chunks) - except (EOFError, av.AVError): # type: ignore[reportAttributeAccessIssue,reportUnknownMemberType] # PyAV exposes AVError dynamically - # Expected for incomplete frames, return empty - return b'' - except Exception as e: - logger.error(f"[AAC] Decode error: {e} {self.uid} {self.session_id}") - return b'' + except (EOFError, av.AVError) as error: # type: ignore[reportAttributeAccessIssue,reportUnknownMemberType] # PyAV exposes AVError dynamically + raise AACDecodeError( + str(error) + ) from error # pyright: ignore[reportUnknownArgumentType] # av.AVError is partially unknown diff --git a/backend/utils/llm/clients.py b/backend/utils/llm/clients.py index b7704b06d7a..6f71f167125 100644 --- a/backend/utils/llm/clients.py +++ b/backend/utils/llm/clients.py @@ -244,14 +244,26 @@ def get_direct_anthropic_client(*, byok_api_key: str | None = None) -> anthropic def _warn_gateway_embeddings_route_absent(operation: str) -> None: - """Report gateway/backend deploy skew once per process. - - Once per process, not per call: this condition holds until the gateway is - redeployed, and the callers behind it run thousands of embeddings an hour. - The gateway's own access log keeps counting the 404s, so nothing is lost by - not repeating ourselves here. + """Report gateway/backend deploy skew: one metric per degrade, one log per process. + + The fallback telemetry fires on every degrade (``backend/AGENTS.md`` rule + 10 / ``docs/agents/fallback-telemetry.md``: a branch that changes mode MUST + call ``record_fallback``), because ``omi_fallback_total`` is how operators + see how much embeddings traffic and ledger spend is bypassing the gateway + while the skew lasts. The narrative ERROR log stays once per process: the + condition holds until the gateway is redeployed, the callers behind it run + thousands of embeddings an hour, and the gateway's own access log keeps + counting the 404s, so nothing is lost by not repeating ourselves there. """ global _gateway_embeddings_route_absent_warned + record_fallback( + component='llm_gateway', + from_mode='gateway_embeddings', + to_mode='direct_embeddings', + reason='capability_mismatch', + outcome='degraded', + log=logger, + ) if _gateway_embeddings_route_absent_warned: return _gateway_embeddings_route_absent_warned = True diff --git a/backend/utils/other/endpoints.py b/backend/utils/other/endpoints.py index 3a0c31e6144..1b5235902c4 100644 --- a/backend/utils/other/endpoints.py +++ b/backend/utils/other/endpoints.py @@ -304,7 +304,7 @@ def _verify_ws_auth(authorization: str) -> str: return verify_token(token) except (InvalidIdTokenError, CertificateFetchError) as e: close_code, reason = _get_ws_auth_close(e) - logger.error("WebSocket auth failed: code=%s error=%s", close_code, e) + _log_ws_auth_rejection(close_code, e) raise WebSocketException(code=close_code, reason=reason) except WebSocketException: raise @@ -313,6 +313,34 @@ def _verify_ws_auth(authorization: str) -> str: raise WebSocketException(code=1008, reason="Auth error") +def _log_ws_auth_rejection(close_code: int, error: Exception) -> None: + """Log a token rejection at the severity its fault origin deserves. + + InvalidIdTokenError means Firebase *evaluated* the client-supplied token + and rejected it for a client-side reason — expired, signed by a key Google + retired, wrong audience, malformed. The close frame (4001/4004/1008) + already tells that client what to do; the rejection is the protocol + working, not a server failure. Logging each attempt at ERROR turned the + stale-client reconnect population into a top-3 production error + signature (backend-listen, GCP 2026-08-30/31: ``Token expired`` up to + ×47/30m and ``Certificate for key id … not found`` ×34/30m for a single + retired key id), burying real serving faults in the same feed. + + CertificateFetchError is the other fault domain: the server could not + fetch Google's public certificates, so it could not even evaluate the + token. That is a server fault and stays at ERROR. + + Failure-Class: FC-request-input-rejection-escapes-as-server-fault — a + route owns the classification of its own request input; a client-caused + token rejection must not be indistinguishable from a serving outage in + error metrics. Close codes are unchanged; only severity is classified. + """ + if isinstance(error, CertificateFetchError): + logger.error("WebSocket auth failed: code=%s error=%s", close_code, error) + else: + logger.warning("WebSocket auth rejected: code=%s error=%s", close_code, error) + + def _get_ws_auth_close(error: Exception) -> 'tuple[int, str]': if isinstance(error, RevokedIdTokenError): return WS_AUTH_CODE_RELOGIN_REQUIRED, "Token revoked; re-login required" diff --git a/backend/utils/stt/live_failure.py b/backend/utils/stt/live_failure.py index d1b7601fef7..e8748b93e8f 100644 --- a/backend/utils/stt/live_failure.py +++ b/backend/utils/stt/live_failure.py @@ -32,6 +32,11 @@ 'connection_lost', 'send_failed', 'socket_unavailable', + # Typed in-stream provider rejections (utils.stt.soniox): a provider + # that accepted the upgrade and then answered an error frame. + 'soniox_account_state', + 'soniox_idle_timeout', + 'soniox_rotation', } ) _FAILURE_PHASE_BY_REASON = { @@ -39,7 +44,32 @@ 'connection_lost': 'connection', 'socket_unavailable': 'connection', 'send_failed': 'send', + # A typed in-stream rejection is the provider closing a connection it had + # accepted. The bounded phase vocabulary has no 'serve' bucket, and 'send' + # would claim our send failed, so 'connection' is the truthful bucket. + 'soniox_account_state': 'connection', + 'soniox_idle_timeout': 'connection', + 'soniox_rotation': 'connection', } +_CIRCUIT_OPENING_REASONS = frozenset( + { + # 402 organization_balance_exhausted: the provider still ACCEPTS the + # WebSocket upgrade but refuses to serve ANY stream, so the + # connect-time failure counter provably never accumulates under + # reconnect load (each dying session's replacement connects fine and + # calls record_success). Same mechanism record_serve_failure exists + # for. The other typed shapes are session-scoped — an idle-timeout is + # this session's VAD pattern and a 413 rotation serves fine on a fresh + # connection — so they must not bench the provider for everyone. + 'soniox_account_state', + } +) + +# Terminal reasons that are evidence about the *provider* while it was serving +# audio. ``initialization_failed`` happens at connect time, where the selection +# helper's threshold logic already sees it, and ``socket_unavailable`` is local +# state (no socket exists), not provider behavior. +_SERVE_FAILURE_REASONS = frozenset({'connection_lost', 'send_failed'}) class LiveSTTSession(Protocol): @@ -91,6 +121,70 @@ def live_stt_socket_is_dead(stt_socket: Any) -> bool: return True +def live_stt_terminal_reason(stt_socket: Any, fallback: str) -> str: + """Prefer a socket's typed provider rejection over the observing path's fallback. + + Every observer of a dead socket (death monitor, send path) knows only its + own vantage point ('connection_lost', 'send_failed'); the socket knows why + the provider actually refused to serve. Providers that answer typed + in-stream error frames (Soniox) latch that reason at the frame; this keeps + it bounded and lets every terminal funnel report it instead of collapsing + a named provider rejection back to generic connection loss. + """ + + try: + typed = getattr(stt_socket, 'typed_death_reason', None) + except Exception: + return fallback + return typed if typed in _KNOWN_FAILURE_REASONS else fallback + + +def note_typed_provider_death(stt_socket: Any, provider: str | None) -> bool: + """Open the selection circuit when a socket died by provider-level rejection. + + A 402 ``organization_balance_exhausted`` stream is served by NO session + while the provider keeps accepting connects, so the mid-session failover + path moves each dying session to the next provider and the session + SURVIVES — which is exactly why the death would otherwise stay invisible + to selection: the surviving session never runs the terminal path that + feeds the circuit, and the next new session is handed right back to the + provider that refuses to serve it. Observing the typed rejection at the + failover seam gives selection the same one-cooldown skip it already gets + from a serve-time death. Session-scoped reasons (idle timeout, rotation) + are deliberately ignored here: they are evidence about one session, not + the provider. + """ + + try: + typed = getattr(stt_socket, 'typed_death_reason', None) + except Exception: + return False + if typed not in _CIRCUIT_OPENING_REASONS: + return False + _open_serving_provider_circuit(typed, provider) + return True + + +def _open_serving_provider_circuit(bounded_reason: str, provider: str | None) -> None: + """Open the process-local selection circuit of the provider that died serving. + + Deliberately cheap and fail-open: the terminal close of the client session + must never be delayed or failed by circuit bookkeeping. Imported lazily + because ``utils.stt.streaming`` imports the socket implementations this + module classifies, so a module-level import would be circular. + """ + try: + from utils.stt.streaming import open_provider_selection_circuit + + open_provider_selection_circuit(provider, reason=bounded_reason) + except Exception as error: # noqa: BLE001 — telemetry-adjacent bookkeeping must not fail the terminal path + logger.warning( + 'Unable to open selection circuit after serve-time death provider=%s error_type=%s', + bounded_provider(provider), + type(error).__name__, + ) + + async def terminate_live_stt_session( websocket: LiveSTTClientSocket, session: LiveSTTSession, @@ -112,6 +206,18 @@ async def terminate_live_stt_session( session.stt_terminal_failure = True session.close_code = LIVE_STT_FAILURE_CLOSE_CODE bounded_reason = _bounded_reason(reason) + if bounded_reason in _SERVE_FAILURE_REASONS or bounded_reason in _CIRCUIT_OPENING_REASONS: + # A provider that died while serving audio is terminal evidence for + # this session, but selection only learns from connect-time outcomes: + # the next reconnect's successful *connect* would call + # ``record_success`` and reset the counter, so serve-time deaths never + # reach the threshold. Open the serving provider's circuit here so + # reconnecting clients skip straight to a healthy fallback for the + # cooldown, instead of being handed back to the provider that just + # died on them (connect -> die -> reconnect -> die under an outage). + # A typed account-state rejection (402) is the same evidence with a + # name: the provider accepts connects and refuses every stream. + _open_serving_provider_circuit(bounded_reason, failure.provider) try: record_live_stt_failure( provider=failure.provider, @@ -216,7 +322,7 @@ async def _recoverable_failure(reason: str) -> None: return False if live_stt_socket_is_dead(stt_socket): - await _recoverable_failure('connection_lost') + await _recoverable_failure(live_stt_terminal_reason(stt_socket, 'connection_lost')) return False try: diff --git a/backend/utils/stt/outcomes.py b/backend/utils/stt/outcomes.py index 400371421c5..8d25cbe3e09 100644 --- a/backend/utils/stt/outcomes.py +++ b/backend/utils/stt/outcomes.py @@ -12,6 +12,12 @@ PrerecordedSTTService.DEEPGRAM, PrerecordedSTTService.MODULATE, PrerecordedSTTService.PARAKEET, + # Live-path provider tokens (config.stt_provider_policy). The live socket + # and failover paths label failures with these; omitting them laundered + # every soniox and deepgram_cloud terminal failure to provider='unknown' + # in omi_live_stt_terminal_failures_total. + 'soniox', + 'deepgram_cloud', } _PUBLIC_FAILURES: dict[TranscriptionOutcome, tuple[int, str, str]] = { diff --git a/backend/utils/stt/provider_resilience.py b/backend/utils/stt/provider_resilience.py index d81a1a93f2a..93bdde893c5 100644 --- a/backend/utils/stt/provider_resilience.py +++ b/backend/utils/stt/provider_resilience.py @@ -113,6 +113,25 @@ def record_failure(self) -> None: self._state = 'open' self._opened_at = self._clock() + def record_serve_failure(self) -> None: + """Open the circuit after a provider died while serving a session. + + Serve-time deaths cannot flow through ``record_failure``: a provider + that accepted the upgrade and served audio before dying passes the + connect-time checks, and the next session's successful *connect* calls + ``record_success``, resetting the failure counter. Under the reconnect + load a mid-session outage produces (connect -> die -> reconnect -> + die), the counter never reaches ``failure_threshold``, so selection + keeps choosing the provider that stopped serving. A serve-time death + is terminal evidence for the session it killed, so it opens the + circuit for the full cooldown; the half-open probe afterwards restores + the provider as soon as one probe serves again. + """ + with self._lock: + self._probe_in_flight = False + self._state = 'open' + self._opened_at = self._clock() + def record_rejection(self, reason: str) -> None: if reason in EXPECTED_REJECTIONS: self.record_success() diff --git a/backend/utils/stt/soniox.py b/backend/utils/stt/soniox.py index 4c35a19c05c..9d4c6283db3 100644 --- a/backend/utils/stt/soniox.py +++ b/backend/utils/stt/soniox.py @@ -23,11 +23,47 @@ SONIOX_SERVICE_NAME: Final = 'soniox' SONIOX_WS_URL: Final = os.getenv('SONIOX_WS_URL', 'wss://stt-rt.soniox.com/transcribe-websocket') SONIOX_MODEL: Final = os.getenv('SONIOX_MODEL', 'stt-rt-v5') -# Soniox closes any socket that receives neither audio nor a keepalive for 20s. +# Soniox closes any socket that receives neither audio nor keepalive for 20s. # VAD gating routinely holds audio back for longer than that, so idle sockets die # as 408 request_timeout unless we fill the gap ourselves. SONIOX_KEEPALIVE_SECONDS: Final = 10.0 +# Typed in-stream rejection reasons, mapped off the provider's own error frame +# (``error_code`` + ``error_type``). Prod 2026-08-30/31 (backend-listen): +# 400 invalid_request "No audio received" (~42/30m), 402 +# organization_balance_exhausted (~7/30m), 413 max_duration_reached (~3/30m) +# all surfaced as one free-text ERROR signature, indistinguishable in metrics +# and in the terminal-failure reason vocabulary. +SONIOX_DEATH_IDLE_TIMEOUT: Final = 'soniox_idle_timeout' +SONIOX_DEATH_ACCOUNT_STATE: Final = 'soniox_account_state' +SONIOX_DEATH_ROTATION: Final = 'soniox_rotation' + + +def soniox_death_reason(error_code: Any, error_type: Any) -> str: + """Bound a Soniox in-stream error frame to a typed death reason. + + The raw provider text stays on the death latch for logs; this mapping is + what the bounded terminal-failure vocabulary consumes, so a new provider + error shape degrades to ``connection_lost`` rather than growing a new + metric cardinality per message. + """ + error = str(error_type or '').strip().lower() + if error == 'organization_balance_exhausted': + return SONIOX_DEATH_ACCOUNT_STATE + try: + code = int(error_code) + except (TypeError, ValueError): + return 'connection_lost' + if code == 400: + # "No audio received": the idle watchdog fired. The socket's keepalive + # covers the no-client-audio case; this shape arrives when VAD gating + # withheld real audio for the whole window. + return SONIOX_DEATH_IDLE_TIMEOUT + if code == 413: + # Documented rotation: open a new WebSocket. The failover path does. + return SONIOX_DEATH_ROTATION + return 'connection_lost' + class SafeSonioxSocket(STTSocket): """Streaming socket for Soniox real-time. @@ -53,6 +89,9 @@ def __init__( self._dead = False self._closed = False self._death_reason: Optional[str] = None + # Typed, bounded death reason (e.g. SONIOX_DEATH_ACCOUNT_STATE) for the + # terminal-failure vocabulary; None until the socket dies. + self._typed_death_reason: Optional[str] = None self._lock = threading.Lock() self._send_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=2000) self._done_event = asyncio.Event() @@ -70,11 +109,17 @@ def is_connection_dead(self) -> bool: def death_reason(self) -> Optional[str]: return self._death_reason - def _mark_dead(self, reason: str) -> None: + @property + def typed_death_reason(self) -> Optional[str]: + """Bounded reason for the terminal-failure vocabulary (None = untyped).""" + return self._typed_death_reason + + def _mark_dead(self, reason: str, typed_reason: Optional[str] = None) -> None: with self._lock: if not self._dead: self._dead = True self._death_reason = reason + self._typed_death_reason = typed_reason def send(self, data: bytes) -> bool: with self._lock: @@ -166,9 +211,20 @@ async def _recv_loop(self) -> None: continue if msg.get('error_code'): err = f"{msg.get('error_code')} {msg.get('error_type', '')} {msg.get('error_message', '')}".strip() - logger.error(f'Soniox streaming error: {err}') + typed = soniox_death_reason(msg.get('error_code'), msg.get('error_type')) + if typed == SONIOX_DEATH_ACCOUNT_STATE: + # The provider evaluated the account and refused to + # serve: a provider fault, and the dominant signal an + # on-call needs during a balance outage. + logger.error(f'Soniox streaming error: {err}') + else: + # Idle-timeout and documented rotation are the + # protocol answering how the session was used, not a + # provider fault; failing to discriminate kept this the + # top backend-listen error signature with no signal. + logger.warning('Soniox stream closed: %s', err) self._done_event.set() - self._mark_dead(f'soniox error: {err}') + self._mark_dead(f'soniox error: {err}', typed_reason=typed) break tokens: List[Any] = msg.get('tokens') or [] diff --git a/backend/utils/stt/streaming.py b/backend/utils/stt/streaming.py index 88808bbbaea..acc8d3a8978 100644 --- a/backend/utils/stt/streaming.py +++ b/backend/utils/stt/streaming.py @@ -111,13 +111,36 @@ def _circuit_for_primary(primary_service: STTService) -> ProviderCircuitBreaker: raise ValueError(f'connection fallback is not defined for a {primary_service.value} primary') +def open_provider_selection_circuit(provider: str | None, *, reason: str) -> bool: + """Open a provider's process-local selection circuit after a serve-time death. + + Selection normally learns from connect-time outcomes alone, so a provider + that accepts the upgrade and dies while serving audio is invisible to it: + the next reconnect's successful connect resets the failure counter. The + live-session terminal path calls this so reconnecting clients skip the + provider that just died for one cooldown window. Returns whether a known + provider's circuit was opened; unknown provider names are tolerated + (same shapes metrics accept) and simply report ``False``. + """ + if not provider: + return False + try: + service = STTService(provider) + except ValueError: + return False + circuit = _circuit_for_primary(service) + logger.warning('Opening %s selection circuit after serve-time death reason=%s', provider, reason) + circuit.record_serve_failure() + return True + + def _fallback_failure_reason(error: BaseException) -> str: """Classify why a fallback provider could not serve, for the next leg's telemetry.""" if isinstance(error, (asyncio.TimeoutError, TimeoutError)): return 'timeout' detail = str(error).lower() - if 'limit' in detail or 'quota' in detail: - return 'quota' + if 'limit' in detail or 'quota' in detail or 'exhausted' in detail or 'balance' in detail: + return 'quota' # incl. Soniox 402 'organization_balance_exhausted' return 'provider_5xx' diff --git a/backend/utils/stt/vad_gate.py b/backend/utils/stt/vad_gate.py index 0094b3b1661..1d384a4190b 100644 --- a/backend/utils/stt/vad_gate.py +++ b/backend/utils/stt/vad_gate.py @@ -639,6 +639,11 @@ def is_connection_dead(self) -> bool: def death_reason(self) -> Optional[str]: return self._conn.death_reason + @property + def typed_death_reason(self) -> Optional[str]: + """Proxy the wrapped socket's typed rejection (None when untyped).""" + return getattr(self._conn, 'typed_death_reason', None) + def _counted(self, audio: bytes) -> bytes: """Count frames that are not whole 16-bit samples, as the provider receives them.