@@ -38,11 +38,12 @@ the other.
   processed, so completion is best-effort: it succeeds only after a final was
   delivered, and otherwise reports that completion cannot be confirmed once a
   bounded interval elapses. A turn first reported later than the quiet period is
-  not recovered. `aclose()` cancels without waiting.
-  `aclose()` cancels the stream without draining it. A dropped connection
-  reconnects and resends the audio sent since the last final (up to 5 seconds);
-  beyond that, or with a turn still pending, the error is non-retryable because
-  a fresh session cannot recover the loss.
+  not recovered. `aclose()` cancels the stream without waiting or draining it.
+  A dropped connection
+  reconnects and resends the audio the last final left unfinalized (up to 5
+  seconds); beyond that, with a turn still pending, or when a final without a
+  reported end position follows newer audio, the error is non-retryable
+  because a fresh session cannot recover the loss.
 - `stepfun.TTS()` supports streaming and one-shot synthesis. Discover voice IDs
   using `await tts.list_voices()`. One-shot synthesis streams PCM over SSE, so
   audio starts arriving before the whole request is synthesized; inputs longer
 
@@ -4,6 +4,7 @@
 import json
 import time
 import weakref
+from collections import deque
 from dataclasses import dataclass, replace
 from typing import TYPE_CHECKING, Any, Literal
 
@@ -470,6 +471,24 @@ class _PendingCommit:
     acknowledged_at: float | None = None
 
 
+@dataclass(eq=False)
+class _Boundary:
+    """What the server reported about where an utterance ended.
+
+    A stop carries the engine's end position when the gateway exposes one.
+    Without it, `sent_at_ms` still records how much audio had been uploaded
+    when the report arrived: anything sent afterwards is beyond what the
+    server had classified, so it may belong to a later utterance.
+    """
+
+    position_ms: float | None
+    #: Audio uploaded when this boundary was recorded.
+    sent_at_ms: float
+    #: The manual commit that established this boundary, when any. Its item
+    #: window bounds how long a final is still expected for it.
+    commit: _PendingCommit | None = None
+
+
 class RecognizeStream(stt.RecognizeStream):
     def __init__(
         self,
@@ -509,8 +528,15 @@ def __init__(
         # sent in this attempt; both restart from zero per connection.
         self._replay_base_ms = 0.0
         self._sent_ms = 0.0
-        # Last audio position the server reported as finalized, if any.
-        self._settled_ms: float | None = None
+        # Finalization reports awaiting the final that owns them, oldest
+        # first. The server finalizes utterances in order, so the oldest
+        # report belongs to the next completed transcript.
+        self._pending_boundaries: deque[_Boundary] = deque()
+        # Audio uploaded when the last finalization report arrived, kept
+        # after its boundary is consumed or pruned. A final that arrives
+        # once its own report is gone is still bounded by it: audio uploaded
+        # afterwards may belong to a later utterance.
+        self._last_report_ms: float | None = None
         self._replay_ambiguous = False
 
         self._audio_duration = 0.0
@@ -547,7 +573,8 @@ async def _run_impl(self) -> None:
         self._finals_received = 0
         self._replay_base_ms = 0.0
         self._sent_ms = 0.0
-        self._settled_ms = None
+        self._pending_boundaries.clear()
+        self._last_report_ms = None
         # Per attempt: a reconnect with nothing to resend has no upload to wait
         # for. Replayed or new frames set it again.
         self._consumed_audio = False
@@ -727,19 +754,100 @@ def _remember(self, data: rtc.AudioFrame | stt.RecognizeStream._FlushSentinel) -
         self._replay.append((data, seconds * 1000))
 
     def _settle(self, boundary_ms: float) -> None:
-        """Drop retained audio the server finalized up to `boundary_ms`."""
+        """Drop retained audio the server finalized up to `boundary_ms`.
+
+        A boundary is measured in milliseconds and can fall inside a retained
+        frame, so that frame keeps only its unfinalized suffix; resending the
+        whole frame would repeat speech the server already transcribed.
+        """
         kept: list[tuple[Any, float]] = []
-        position = self._replay_base_ms
+        base: float | None = None
+        cursor = self._replay_base_ms
+        trimmed = False
         for item, duration_ms in self._replay:
-            if position + duration_ms <= boundary_ms + 0.001:
-                position += duration_ms
-                self._replay_seconds = max(0.0, self._replay_seconds - duration_ms / 1000)
+            end = cursor + duration_ms
+            if end <= boundary_ms + 0.001:
+                cursor = end
                 continue
+            if not trimmed:
+                trimmed = True
+                if isinstance(item, rtc.AudioFrame) and cursor < boundary_ms:
+                    rate = item.sample_rate
+                    cut = min(
+                        int(round((boundary_ms - cursor) * rate / 1000)),
+                        item.samples_per_channel,
+                    )
+                    if cut > 0:
+                        stride = item.num_channels * 2
+                        item = rtc.AudioFrame(
+                            data=item.data.tobytes()[cut * stride :],
+                            sample_rate=rate,
+                            num_channels=item.num_channels,
+                            samples_per_channel=item.samples_per_channel - cut,
+                        )
+                        duration_ms -= cut * 1000 / rate
+                        cursor = boundary_ms
+            if base is None:
+                base = cursor
             kept.append((item, duration_ms))
-        self._replay_base_ms = position
+            cursor += duration_ms
         self._replay = kept
+        self._replay_base_ms = base if base is not None else cursor
+        self._replay_seconds = sum(duration for _, duration in kept) / 1000
         self._audio_since_final = self._replay_seconds
 
+    def _prune_boundaries(self, now: float) -> None:
+        """Drop finalization reports whose final is no longer expected.
+
+        A manual commit whose item window elapsed produced no final, so it
+        committed silence: its boundary must not be consumed by the final of
+        a later utterance. Speech-stopped reports carry no such bound, since
+        the gateway pairs every stop with a final.
+        """
+        while self._pending_boundaries:
+            commit = self._pending_boundaries[0].commit
+            if (
+                commit is None
+                or commit.acknowledged_at is None
+                or now < commit.acknowledged_at + _COMMIT_ITEM_WINDOW
+            ):
+                break
+            self._pending_boundaries.popleft()
+
+    def _settle_replay(self) -> None:
+        """Release the replay audio a completed transcript finalized.
+
+        Each report is consumed by the final that owns it, so a boundary
+        always describes the utterance that reported it. A report without a
+        position still bounds what the final can cover by how much audio had
+        been uploaded when it arrived: if more was sent afterwards, that
+        audio may belong to a later utterance, so it stays retained and a
+        retry is refused rather than allowed to discard it. A final whose own
+        report is already gone is bounded the same way by the last one.
+        """
+        self._prune_boundaries(time.monotonic())
+        boundary = self._pending_boundaries.popleft() if self._pending_boundaries else None
+
+        if boundary is not None and boundary.position_ms is not None:
+            self._replay_ambiguous = False
+            self._replay_overflow = False
+            self._settle(boundary.position_ms)
+            return
+
+        reported_ms = boundary.sent_at_ms if boundary is not None else self._last_report_ms
+        if reported_ms is not None and self._sent_ms > reported_ms:
+            # Audio arrived after the server finalized up to here, so the
+            # final cannot be shown to cover it.
+            self._replay_ambiguous = True
+            self._audio_since_final = self._replay_seconds
+            return
+
+        # Nothing indicates audio beyond this final, so it covers everything
+        # uploaded so far and there is nothing left to resend.
+        self._replay_ambiguous = False
+        self._replay_overflow = False
+        self._settle(self._sent_ms)
+
     async def _send_item(
         self,
         ws: _WebSocket,
@@ -935,19 +1043,25 @@ async def _maybe_commit(self, ws: _WebSocket) -> None:
             if pending.acknowledged_at is None
             or now < pending.acknowledged_at + _COMMIT_ITEM_WINDOW
         ]
+        self._prune_boundaries(now)
         commit = _PendingCommit()
         # Register before yielding: the receiver may process the ack while
         # send_str is suspended. Roll back only if still unacked.
         self._commits.append(commit)
+        # A commit finalizes everything sent so far, so its boundary is the
+        # position at send time. Recorded before the await for the same
+        # reason as the commit itself: the final can be processed while the
+        # send is still draining. A failed upload takes the whole attempt
+        # down with it, so a boundary left behind is never used.
+        self._last_report_ms = self._sent_ms
+        self._pending_boundaries.append(_Boundary(self._sent_ms, self._sent_ms, commit))
         try:
             await self._send_commit(ws)
         except BaseException:
             if commit.acknowledged_at is None:
                 self._commits.remove(commit)
             raise
         self._uncommitted_audio = False
-        # A commit finalizes everything sent so far.
-        self._settled_ms = self._sent_ms
 
     # -- downstream --------------------------------------------------------
 
@@ -1024,8 +1138,13 @@ def _handle_event(self, event: dict[str, Any]) -> None:
             self._speech_active = False
             # The gateway exposes the engine's end position as audio_start_ms.
             position = event.get("audio_end_ms", event.get("audio_start_ms"))
-            if isinstance(position, (int, float)):
-                self._settled_ms = max(self._settled_ms or 0.0, float(position))
+            self._last_report_ms = self._sent_ms
+            self._pending_boundaries.append(
+                _Boundary(
+                    float(position) if isinstance(position, (int, float)) else None,
+                    self._sent_ms,
+                )
+            )
             self._event_ch.send_nowait(
                 stt.SpeechEvent(
                     type=stt.SpeechEventType.END_OF_SPEECH,
@@ -1103,15 +1222,7 @@ def _handle_completed(self, event: dict[str, Any]) -> None:
         self._open_items.discard(item_id)
         self._expected_finals = max(0, self._expected_finals - 1)
         self._finals_received += 1
-        if self._settled_ms is None:
-            # No boundary for this final: later audio may be buffered, so a
-            # retry could discard it. Keep everything and refuse the retry.
-            self._replay_ambiguous = True
-            self._audio_since_final = self._replay_seconds
-        else:
-            self._replay_ambiguous = False
-            self._replay_overflow = False
-            self._settle(self._settled_ms)
+        self._settle_replay()
         # A final closes any unidentified report too: the server finalizes
         # utterances in order, and an id-less start belongs to this one.
         self._anonymous_open = False