"""Regression test for #53 (head-clipping) + #54 (capture start latency) — 2026-05-18. NOTE (2026-05-18): the SPACE hold-to-talk path was removed entirely; the only voice-capture control is now the on-screen Push-to-talk button. This test no longer pins any page.tsx SPACE wiring. It still pins the two contracts that remain live and valuable: 1. The warm-stream PreRollRing + evaluateHoldGate exported from frontend/src/lib/useStreamingVoice.ts (a rolling ~PRE_ROLL_MS pre-roll buffer + a deliberate-hold gate that ignores sub-threshold taps). 2. The backend STT chunking path: the pre-roll head word survives transcription and long audio is chunked, not truncated — this guards the /api/transcribe behavior the Push-to-talk button still relies on. ORIGINAL ROOT CAUSE (kept for context) ----------------------------------------------------------------------------- A cold-started mic loses every word spoken between capture-request and recorder.start() (getUserMedia is 200-700ms). Real repro: user said "Sir. My age is 29 ..." -> transcribed "S A R. My age is 29 ..." (#53); the same cold-start is the multi-second start delay (#54). The warm-stream PreRollRing keeps the first word at the head of the blob. WHAT THIS TEST PINS ----------------------------------------------------------------------------- There is no JS test runner in this repo, so the two contracts are pinned at the layers a Python test can reach honestly: PART A (pure frontend logic, executed against the REAL shipped code): The exported PreRollRing + evaluateHoldGate from useStreamingVoice.ts are loaded under Node (TS types stripped; react/./api/./voice_resilience stubbed since the PURE exports don't use them) and asserted: A1 a pre-roll ring fed lead-in slices then drained returns the LEADING slice first -> the first word is at the head of the blob. A2 the ring retains >= PRE_ROLL_MS of audio (covers the worst-case page.tsx cold-start gap) and evicts only OLDER audio. A3 evaluateHoldGate: a >= threshold hold is deliberate (submit); a sub-threshold tap is NOT (gated, nothing submitted). A4 the integrated contract: engage-after-threshold seeds the capture with the pre-roll so leadIn[0] precedes live audio; a tap drains nothing and submits nothing. PART B (backend STT, the bug's observable failure surface): The fix prepends a short pre-roll HEAD to the blob. Prove backend/providers/sarvam_stt.py never silently drops/truncates that head — i.e. the FIRST word survives transcription, including when the full utterance is long enough to hit the silence-chunk splitter (the pre-fix 30s truncation dropped tail words; a regression that mishandles the prepended head would drop the FIRST word instead). Run: cd /Users/rohitsar/Developer/Insurance\\ Sales\\ Bot PYTHONPATH=$PWD .venv/bin/python -m pytest \ tests/test_ptt_preroll_warm_stream.py -v """ from __future__ import annotations import asyncio import json import os import shutil import subprocess import sys import textwrap import types from pathlib import Path import pytest os.environ.setdefault("SARVAM_API_KEY", "test-key-for-ptt-preroll") REPO = Path(__file__).resolve().parents[1] HOOK = REPO / "frontend" / "src" / "lib" / "useStreamingVoice.ts" # =========================================================================== # PART A — pure frontend logic, executed against the REAL shipped exports. # =========================================================================== # The harness registers an ESM resolve/load hook (see _build_node_loader) # that stubs `react`, `./api`, `./voice_resilience` — modules the React hook # body imports but which the PURE exports (PreRollRing / evaluateHoldGate) # never touch — and lets Node 24 strip the TS types on load. This exercises # the genuine SHIPPED code, not a copy. def _have_node() -> bool: return shutil.which("node") is not None def _build_node_loader(tmpdir: Path) -> Path: """Write an ESM loader that stubs react/./api/./voice_resilience and strips TS types so the REAL hook module's pure exports can be imported. """ loader = tmpdir / "loader.mjs" loader.write_text( textwrap.dedent( r""" import { readFileSync } from 'node:fs'; import { fileURLToPath } from 'node:url'; import vm from 'node:vm'; const STUBS = { react: ` export const useCallback = (f) => f; export const useEffect = () => {}; export const useRef = (v) => ({ current: v }); export const useState = (v) => [typeof v === 'function' ? v() : v, () => {}]; export default {}; `, api: `export const postTranscribe = async () => ({ text: '' });`, voice_resilience: ` export const retryPostTranscribe = async () => null; export const scaleSpeechZcrBand = () => ({ min: 20, max: 250 }); export class AdaptiveNoiseFloor { feed(){} currentThreshold(){ return 0.008; } } `, }; export async function resolve(specifier, context, nextResolve) { if (specifier === 'react') return { url: 'stub:react', shortCircuit: true }; if (specifier === './api' || specifier.endsWith('/api')) return { url: 'stub:api', shortCircuit: true }; if (specifier === './voice_resilience' || specifier.endsWith('/voice_resilience')) return { url: 'stub:voice_resilience', shortCircuit: true }; return nextResolve(specifier, context); } export async function load(url, context, nextLoad) { if (url.startsWith('stub:')) { const key = url.slice('stub:'.length); return { format: 'module', source: STUBS[key], shortCircuit: true }; } if (url.endsWith('.ts')) { const path = fileURLToPath(url); const src = readFileSync(path, 'utf8'); // Node 24 strips TS types natively for .ts; force module // format + hand it the raw source (type-strip happens in the // default ts transform path). return { format: 'module-typescript', source: src, shortCircuit: true }; } return nextLoad(url, context); } """ ).strip() + "\n" ) return loader def _run_node_contract(tmp_path: Path) -> dict: """Import the REAL PreRollRing + evaluateHoldGate under Node and run the four pure-logic contracts. Returns the parsed JSON verdict.""" loader = _build_node_loader(tmp_path) script = tmp_path / "contract.mjs" # Pull the real exports + the real PRE_ROLL_MS / HOLD_THRESHOLD_MS / # WARM_TIMESLICE_MS constants from the shipped hook module. script.write_text( textwrap.dedent( f""" const mod = await import({json.dumps(HOOK.as_uri())}); const {{ PreRollRing, evaluateHoldGate, PRE_ROLL_MS, HOLD_THRESHOLD_MS, WARM_TIMESLICE_MS, }} = mod; const out = {{}}; // --- A1/A2: pre-roll ring keeps the LEADING audio + >=PRE_ROLL_MS const ring = new PreRollRing(PRE_ROLL_MS); // Feed far more than the window so eviction is forced. Each slice // is a distinct 1-byte blob tagged by index so order is provable. const totalSlices = Math.ceil((PRE_ROLL_MS / WARM_TIMESLICE_MS) * 4); for (let i = 0; i < totalSlices; i++) {{ ring.push(new Blob([String.fromCharCode(65 + (i % 26))]), WARM_TIMESLICE_MS); }} out.retainedMs = ring.retainedDurationMs(); out.retainsAtLeastWindow = ring.retainedDurationMs() >= PRE_ROLL_MS; const drained = ring.drain(); out.drainedCount = drained.length; out.ringEmptyAfterDrain = ring.retainedDurationMs() === 0; // Now the CORE #53 contract: a head word spoken just before the // window boundary must still be at the HEAD of the drained blob. const r2 = new PreRollRing(PRE_ROLL_MS); // 'HEAD' = the first word's slice. Then enough silence-ish slices // to *almost* (but not quite, given the +1 slice slack) fill the // window, so HEAD must still be retained as slice 0. const headBlob = new Blob(['HEAD']); r2.push(headBlob, WARM_TIMESLICE_MS); const fill = Math.floor(PRE_ROLL_MS / WARM_TIMESLICE_MS) - 1; for (let i = 0; i < fill; i++) r2.push(new Blob(['x']), WARM_TIMESLICE_MS); const d2 = r2.drain(); out.headSurvives = d2.length > 0 && (await d2[0].text()) === 'HEAD'; // --- A3: deliberate-hold gate const tap = evaluateHoldGate(1000, 1000 + (HOLD_THRESHOLD_MS - 1), HOLD_THRESHOLD_MS); const hold = evaluateHoldGate(1000, 1000 + HOLD_THRESHOLD_MS, HOLD_THRESHOLD_MS); const longHold = evaluateHoldGate(1000, 1000 + 5000, HOLD_THRESHOLD_MS); out.tapGated = tap.deliberate === false; out.holdAtThresholdDeliberate = hold.deliberate === true; out.longHoldDeliberate = longHold.deliberate === true; out.tapHeldMs = tap.heldMs; // --- A4: integrated — engage seeds capture with pre-roll so the // leading word precedes the live audio; a tap submits nothing. const r3 = new PreRollRing(PRE_ROLL_MS); r3.push(new Blob(['FIRST_WORD']), WARM_TIMESLICE_MS); // spoken in cold-start gap // engage: drain pre-roll into capture, THEN append live slices const capture = [...r3.drain()]; capture.push(new Blob(['rest_of_sentence'])); out.captureLeadIsFirstWord = (await capture[0].text()) === 'FIRST_WORD'; out.captureIncludesLive = capture.length === 2; // tap path: gate says not deliberate -> nothing assembled const tapDecision = evaluateHoldGate(2000, 2050, HOLD_THRESHOLD_MS); out.tapAssemblesNothing = tapDecision.deliberate === false; out.constants = {{ PRE_ROLL_MS, HOLD_THRESHOLD_MS, WARM_TIMESLICE_MS }}; process.stdout.write(JSON.stringify(out)); """ ).strip() + "\n" ) proc = subprocess.run( [ "node", "--no-warnings", f"--experimental-loader={loader.as_uri()}", str(script), ], capture_output=True, text=True, cwd=str(REPO / "frontend"), timeout=60, ) if proc.returncode != 0: raise AssertionError( "Node pure-logic harness failed:\n" f"STDOUT:\n{proc.stdout}\n\nSTDERR:\n{proc.stderr}" ) return json.loads(proc.stdout.strip()) @pytest.mark.skipif(not _have_node(), reason="node not available") def test_preroll_and_hold_gate_pure_logic(tmp_path): """PART A — exercise the REAL exported PreRollRing + evaluateHoldGate.""" r = _run_node_contract(tmp_path) # A2 — the ring always retains at least the requested pre-roll window so # the worst-case page.tsx cold-start gap (getUserMedia 200-700ms + the # 400ms Live-teardown wait at page.tsx:994) is fully covered. assert r["retainsAtLeastWindow"] is True, r assert r["retainedMs"] >= r["constants"]["PRE_ROLL_MS"], r assert r["ringEmptyAfterDrain"] is True, r assert r["drainedCount"] >= 1, r # A1 — the CORE #53 fix: the first word's slice is at the HEAD of the # drained lead-in (it is NOT evicted while it is within the window). assert r["headSurvives"] is True, ( "pre-roll dropped the leading slice — first word would be clipped: " f"{r}" ) # A3 — deliberate-hold gate (#54): a sub-threshold tap is gated; a hold # exactly at the threshold (and longer) is deliberate. assert r["tapGated"] is True, r assert r["holdAtThresholdDeliberate"] is True, r assert r["longHoldDeliberate"] is True, r assert r["tapHeldMs"] < r["constants"]["HOLD_THRESHOLD_MS"], r # A4 — integrated: engage seeds the capture with the pre-roll so the # leading word precedes the live audio; a tap assembles/submits nothing. assert r["captureLeadIsFirstWord"] is True, ( "engaged capture does not start with the pre-roll lead-in — first " f"word lost: {r}" ) assert r["captureIncludesLive"] is True, r assert r["tapAssemblesNothing"] is True, r # The shipped constants must stay in the spec'd bands. assert r["constants"]["PRE_ROLL_MS"] >= 500, r # >= 500ms pre-roll assert 150 <= r["constants"]["HOLD_THRESHOLD_MS"] <= 250, r # deliberate assert r["constants"]["WARM_TIMESLICE_MS"] <= 250, r # fine-grained ring # =========================================================================== # PART B — backend STT: the prepended pre-roll HEAD must never be silently # dropped/truncated (the bug's observable failure surface). # =========================================================================== from backend.providers.sarvam_stt import STT_CHUNK_MS, SarvamSTT # noqa: E402 class FakeAudioSegment: """Minimal pydub.AudioSegment stub (same surface as test_stt_long_audio_chunking.py). One word per second; word_0 is the PRE-ROLL HEAD (the first word spoken in page.tsx's cold-start gap).""" def __init__(self, start_ms: int, end_ms: int): self.start_ms = start_ms self.end_ms = end_ms def __len__(self): return self.end_ms - self.start_ms def __getitem__(self, sl): if isinstance(sl, slice): lo = 0 if sl.start is None else sl.start hi = len(self) if sl.stop is None else sl.stop lo = max(0, min(lo, len(self))) hi = max(0, min(hi, len(self))) return FakeAudioSegment(self.start_ms + lo, self.start_ms + hi) raise TypeError("only slice indexing used by splitter") @property def dBFS(self): if self.end_ms - self.start_ms == 0: return float("-inf") # A real pause near the 25s ceiling so the splitter snaps there. if 24000 <= self.start_ms < 24400: return float("-inf") return -20.0 def set_frame_rate(self, _): return self def set_channels(self, _): return self def set_sample_width(self, _): return self def export(self, buf, format="wav"): # noqa: A002 buf.write(f"FAKEWAV:{self.start_ms}:{self.end_ms}".encode()) buf.seek(0) return buf SARVAM_HARD_LIMIT_MS = 30_000 class _FakeResp: def __init__(self, payload): self._payload = payload def raise_for_status(self): return None def json(self): return self._payload class _FakeClient: def __init__(self, *a, **k): pass async def __aenter__(self): return self async def __aexit__(self, *a): return False async def post(self, url, headers=None, files=None, data=None): raw = files["file"][1].read() text = raw.decode() assert text.startswith("FAKEWAV:"), text _, s, e = text.split(":") start_ms, end_ms = int(s), int(e) capped_end = min(end_ms, start_ms + SARVAM_HARD_LIMIT_MS) words = [f"word_{ms // 1000}" for ms in range(start_ms, capped_end, 1000)] return _FakeResp( { "transcript": " ".join(words), "language_code": "en-IN", "language_probability": 0.99, } ) @pytest.fixture def patch_pydub(monkeypatch): fake_pydub = types.ModuleType("pydub") class _Factory: @staticmethod def from_file(_bio, format=None): # noqa: A002 # 92s utterance — long enough to hit the silence splitter (>3 # Sarvam 25s windows). word_0 is the prepended PRE-ROLL HEAD. return FakeAudioSegment(0, 92_000) fake_pydub.AudioSegment = _Factory monkeypatch.setitem(sys.modules, "pydub", fake_pydub) yield @pytest.fixture def patch_httpx(monkeypatch): import backend.providers.sarvam_stt as mod monkeypatch.setattr(mod.httpx, "AsyncClient", _FakeClient) yield def test_preroll_head_word_survives_backend_stt(patch_pydub, patch_httpx): """The FIRST word (word_0 — the pre-roll head the warm-stream fix prepends) must appear in the transcript. A regression that mishandled a prepended head, or re-introduced single-shot truncation, would drop word_0 (head-clipping #53 at the STT layer).""" stt = SarvamSTT() audio = b"\x00" * 4096 # > 1 KB so the short-audio guard doesn't fire result = asyncio.run( stt.transcribe(audio_bytes=audio, audio_format="webm", language_code="en-IN") ) words = result.text.split() # CORE #53 CONTRACT at the STT boundary: the leading word is present and # is actually FIRST (not just somewhere in the middle). assert "word_0" in words, f"FIRST word dropped by STT — head clipped: {words[:5]}" deduped = [w for i, w in enumerate(words) if i == 0 or w != words[i - 1]] assert deduped[0] == "word_0", ( f"transcript does not START with the pre-roll head word: {deduped[:5]}" ) # And the WHOLE 92s utterance survives in order (no head drop, no tail # truncation) — the prepended pre-roll doesn't break chunking. expected = [f"word_{i}" for i in range(92)] assert deduped == expected, ( f"transcript garbled — deduped {len(deduped)} words " f"(first={deduped[:3]}, last={deduped[-3:]}), expected 92 in order" ) assert result.raw.get("chunked") is True assert result.raw.get("chunk_count", 0) >= 4 def test_short_preroll_only_blob_is_not_truncated(patch_pydub, patch_httpx, monkeypatch): """A short blob that is essentially pre-roll + a couple words (the common PTT case) must transcribe in ONE call with word_0 intact — proving the pre-roll head adds no truncation and no extra latency for short clips.""" short = FakeAudioSegment(0, 5_000) # ~5s: pre-roll + a short answer monkeypatch.setattr( sys.modules["pydub"].AudioSegment, "from_file", staticmethod(lambda *a, **k: short), ) stt = SarvamSTT() result = asyncio.run( stt.transcribe(audio_bytes=b"\x00" * 4096, audio_format="webm") ) words = result.text.split() assert words == [f"word_{i}" for i in range(5)], words assert words[0] == "word_0", f"pre-roll head missing on short clip: {words}" # Single-call path — no chunked marker, no added round-trips. assert result.raw.get("chunked") is not True