fix(voice): stop the recorder writing over audio still in the resampler

livekit-agents/livekit/agents/voice/recorder_io/recorder_io.py

@@ -35,6 +35,10 @@
 # beyond it keeps a drifting capture clock from sliding the channel
 RESYNC_TOLERANCE = 0.1
 
+# a resampler reports its input as arrived before it emits it, so the writer waits this long for
+# the samples it still holds rather than writing silence over their place
+MAX_RESAMPLER_LAG = 0.1
+
 
 @dataclass
 class _Captured:
@@ -109,6 +113,13 @@ def end_run(self) -> None:
         self._run_start = None
         self._run_samples = 0
 
+    @property
+    def placed_through(self) -> int | None:
+        """How far the open run reaches, or None when the resampler holds nothing back."""
+        if self._resampler is None or self._run_start is None:
+            return None
+        return round((self._run_start - self._t0) * self._sample_rate) + self._run_samples
+
     def _place(self, frames: list[rtc.AudioFrame]) -> None:
         if not frames:
             return
@@ -293,6 +304,16 @@ def _encode_thread(self) -> None:
                         continue
 
                     end = round((item.until - self._t0) * self._sample_rate)
+                    held = end - round(MAX_RESAMPLER_LAG * self._sample_rate)
+                    for track in tracks:
+                        if (through := track.placed_through) is None:
+                            continue
+                        if through >= held:
+                            end = min(end, through)
+                        else:
+                            # the source stopped delivering; place what it holds before
+                            # the cursor moves past the run it belongs to
+                            track.end_run()
                     if end <= cursor:
                         continue
 

tests/test_recorder_io.py

@@ -247,6 +247,49 @@ def test_ending_a_resampled_run_twice_is_idempotent() -> None:
     assert np.count_nonzero(block[48000:]) == pytest.approx(4800, abs=50)
 
 
+def test_the_writer_waits_for_samples_the_resampler_still_holds() -> None:
+    """A resampler reports its input as arrived before it emits it, so a cursor placed at
+    the arrival time would write silence over audio that is still on its way."""
+    track = _Track(sample_rate=48000, t0=0.0)
+    for i in range(10):  # 500ms of continuous 24kHz input, as room_io delivers it
+        track.push(i * 0.05, _loud(1200, sample_rate=24000))
+
+    arrived = round(0.5 * 48000)
+    assert track.placed_through is not None
+    assert track.placed_through < arrived  # the resampler is still holding the difference
+
+    block = track.take(0, track.placed_through)
+    track.push(0.5, _loud(1200, sample_rate=24000))
+    block = np.concatenate([block, track.take(track.placed_through, arrived)])
+    assert np.count_nonzero(block) == len(block)  # no gap where the held-back samples belong
+
+
+def test_ending_a_stalled_run_leaves_its_tail_where_the_cursor_can_reach_it() -> None:
+    """When a source stops, the writer ends its run before it moves past it, so the samples
+    the resampler still holds land at the cursor instead of behind it."""
+    track = _Track(sample_rate=48000, t0=0.0)
+    for i in range(10):  # 500ms of 24kHz input, then the source goes quiet
+        track.push(i * 0.05, _loud(1200, sample_rate=24000))
+
+    cursor = track.placed_through
+    assert cursor is not None
+    written = track.take(0, cursor)  # the writer has written everything placed so far
+    track.end_run()
+    tail = track.take(cursor, cursor + 48000)
+
+    # all 500ms survives: what was placed, plus the tail the resampler was holding
+    assert np.count_nonzero(written) + np.count_nonzero(tail) == pytest.approx(24000, abs=50)
+    assert track.dropped_samples == 0
+
+
+def test_a_track_holding_nothing_back_does_not_hold_the_writer() -> None:
+    track = _Track(sample_rate=48000, t0=0.0)
+    assert track.placed_through is None  # nothing pushed yet
+
+    track.push(0.0, _loud(4800, sample_rate=48000))  # recording rate, so no resampler
+    assert track.placed_through is None
+
+
 def test_a_stereo_source_is_mixed_down() -> None:
     track = _Track(sample_rate=1000, t0=0.0)
     track.push(0.0, _loud(100, channels=2))