fix(stepfun): reset consumed audio per reconnect attempt

A reconnect with nothing left to resend has no upload to wait for, so an
empty attempt must not report a missing transcription. Replayed frames set the
flag again, keeping the finalization wait when audio really is resent.

Author: robinzsx

livekit-plugins/livekit-plugins-stepfun/livekit/plugins/stepfun/stt.py

@@ -537,6 +537,9 @@ async def _run_impl(self) -> None:
         self._commits.clear()
         self._expected_finals = 0
         self._finals_received = 0
+        # 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
         self._input_finished_at = 0.0
         self._last_event_at = 0.0
         self._uncommitted_audio = False

tests/test_plugin_stepfun_stt.py

@@ -1334,6 +1334,39 @@ async def send_str(self, data: str) -> None:
         await stream.aclose()
 
 
+async def test_reconnect_with_nothing_to_resend_ends_cleanly() -> None:
+    """A finalized turn, then a drop, then end_input with no further audio."""
+
+    class Failing(_ReadyPeer):
+        async def send_str(self, data: str) -> None:
+            await super().send_str(data)
+            if self.sent[-1]["type"] == "input_audio_buffer.append":
+                for event in (
+                    {"type": "input_audio_buffer.speech_started", "item_id": "item-1"},
+                    {"type": "input_audio_buffer.speech_stopped", "item_id": "item-1"},
+                    {"type": _COMPLETED, "item_id": "item-1", "transcript": "hello"},
+                ):
+                    self.queue_server_event(event)
+                self.queue_error()
+
+    pool = _SequencedPool([Failing(), _ReadyPeer()])
+    s = stepfun.STT()
+    s._pool = pool  # type: ignore[assignment]
+    stream = s.stream(conn_options=APIConnectOptions(max_retry=1, retry_interval=0, timeout=1))
+    stream.push_frame(_frame())
+    try:
+        for _ in range(100):
+            if pool.handed_out == 2:
+                break
+            await asyncio.sleep(0.01)
+        assert pool.handed_out == 2
+        stream.end_input()  # nothing was sent on the new session
+        events = await asyncio.wait_for(_collect(stream), 3)
+        assert _texts(events, stt.SpeechEventType.FINAL_TRANSCRIPT) == ["hello"]
+    finally:
+        await stream.aclose()
+
+
 async def test_connection_loss_beyond_the_replay_cap_is_fatal() -> None:
     """Unbounded resend is not attempted; past the cap the loss is reported."""