fix(realtime): 连续对话重复回复、结束对话卡死与播放卡顿修复

backend/src/timeflow/gateway/websocket/agent_ports.py

@@ -76,6 +76,11 @@ def duration_ms(self) -> int:
         """How long the audio ran."""
         ...
 
+    @property
+    def turn_id(self) -> str | None:
+        """Which stretch of user speech this belongs to, when the producer knows."""
+        ...
+
 
 class ReplyTextProgress(Protocol):
     """How much of a reply's wording is known so far."""
@@ -95,6 +100,11 @@ def done(self) -> bool:
         """Whether this is the last update for this reply."""
         ...
 
+    @property
+    def turn_id(self) -> str | None:
+        """Which stretch of user speech this belongs to, when the producer knows."""
+        ...
+
 
 class DialogueQuestionInfo(Protocol):
     """A question the agent needs answered before it can act."""
@@ -124,6 +134,11 @@ def candidates(self) -> tuple[dict[str, Any], ...]:
         """Choices the user is being asked to pick between, when there are any."""
         ...
 
+    @property
+    def turn_id(self) -> str | None:
+        """Which stretch of user speech this belongs to, when the producer knows."""
+        ...
+
 
 class CommandOutcome(Protocol):
     """A command that was carried out, ready to be sent to the client."""

backend/src/timeflow/gateway/websocket/handlers/agent_result.py

@@ -76,6 +76,7 @@ async def deliver_transcript(
                 transcript=transcript.text,
                 language=transcript.language,
                 duration_ms=transcript.duration_ms,
+                turn_id=transcript.turn_id,
             ),
         )
         await self._send(stream.session_id, message.type, message.model_dump())
@@ -90,6 +91,7 @@ async def deliver_reply_text(self, reply: ReplyTextProgress, stream: StreamIdent
                 reply_id=reply.reply_id,
                 speech_text=reply.speech_text,
                 done=reply.done,
+                turn_id=reply.turn_id,
             ),
         )
         await self._send(stream.session_id, message.type, message.model_dump())
@@ -123,6 +125,7 @@ async def deliver_question(
                 speech_text=question.speech_text,
                 required_response=question.required_response,
                 candidates=list(question.candidates),
+                turn_id=question.turn_id,
             ),
         )
         await self._send(stream.session_id, message.type, message.model_dump())

backend/src/timeflow/gateway/websocket/handlers/voice_stream.py

@@ -114,6 +114,18 @@ async def handle_start(
         # between this check and the assignment below.
         active = self._active_streams.get(session.session_id)
         if active is not None:
+            # Logged, not silent: a client that never sends voice.stream.end for an
+            # active stream (found on push-to-talk's swipe-to-cancel gesture, which
+            # closed the shared connection's local listener without ending the stream)
+            # gets every future voice.stream.start on this session rejected here, with
+            # nothing else in this file's logs to explain why.
+            logger.warning(
+                "rejected voice.stream.start: session_id=%s already has an active "
+                "stream stream_id=%s -- the client must send voice.stream.end (or "
+                "disconnect) before opening a new one",
+                session.session_id,
+                active.context.stream_id,
+            )
             return self._error(request_id, "A stream is already active for this session", active)
 
         payload = message.payload
@@ -239,17 +251,35 @@ async def _drain_to_sink(self, stream: _ActiveStream) -> None:
                     "stream_id": stream.context.stream_id,
                 },
             )
-            self._retire_failed_stream(stream)
+        finally:
+            # Retired however consume() ended, not just when it raised: a sink may
+            # also give up by returning -- the realtime agent does exactly that when
+            # it cannot open a vendor session, and its pump returning early on a
+            # vendor error looks the same from here. Either way nothing drains the
+            # queue afterwards, so leaving the stream registered wedges the whole
+            # connection (see _retire_stream).
+            self._retire_stream(stream)
 
-    def _retire_failed_stream(self, stream: _ActiveStream) -> None:
-        """Drop a stream whose consumer has died, so the session keeps working.
+    def _retire_stream(self, stream: _ActiveStream) -> None:
+        """Drop a stream nobody is reading any more, so the session keeps working.
+
+        A no-op on the ordinary path, where voice.stream.end already took the stream
+        out before the sink finished with it.
 
         Removing it alone is not enough: the receive loop may be parked on a full
-        queue with nobody left to drain it, and would stay there forever. Emptying
-        the queue releases it, and the next frame is refused instead of enqueued.
+        queue with nobody left to drain it, and would stay there forever -- including
+        for the voice.stream.end that is the client's only way out. Emptying the queue
+        releases it, and the next frame is refused instead of enqueued.
         """
         session_id = stream.context.session.session_id
         if self._active_streams.get(session_id) is stream:
+            logger.warning(
+                "retiring an audio stream its consumer walked away from: "
+                "session_id=%s stream_id=%s -- further frames will be refused until "
+                "the client starts a new stream",
+                session_id,
+                stream.context.stream_id,
+            )
             self._active_streams.pop(session_id, None)
         while not stream.queue.empty():
             stream.queue.get_nowait()

backend/src/timeflow/gateway/websocket/messages/agent.py

@@ -13,6 +13,10 @@ class VoiceAsrCompletedPayload(BaseModel):
     transcript: str
     language: str
     duration_ms: int
+    # Which stretch of user speech this transcribes, so the client can pair it with the
+    # reply answering it instead of guessing from arrival order. Optional: only the
+    # realtime backend has a vendor id to report, and older clients ignore it.
+    turn_id: str | None = None
 
 
 class VoiceAsrCompleted(BaseModel):

backend/src/timeflow/gateway/websocket/messages/dialogue.py

@@ -15,6 +15,8 @@ class VoiceDialogueReplyPayload(BaseModel):
     reply_id: str
     speech_text: str
     done: bool = False
+    # Which stretch of user speech this answers; see VoiceAsrCompletedPayload.turn_id.
+    turn_id: str | None = None
 
 
 class VoiceDialogueReply(BaseModel):
@@ -34,6 +36,8 @@ class VoiceDialogueQuestionPayload(BaseModel):
     speech_text: str
     required_response: str | None = None
     candidates: list[dict[str, Any]] = []
+    # Which stretch of user speech this asks about; see VoiceAsrCompletedPayload.turn_id.
+    turn_id: str | None = None
 
 
 class VoiceDialogueQuestion(BaseModel):