feat(assistant): 组合式连续对话在 LLM 和播音残留窗口内打断

backend/src/timeflow/intelligence/composed/agent.py

@@ -63,6 +63,9 @@
 # delete that commits the first few schedules then hits the cap). The earlier tool calls
 # have already committed, so the user must be told the rest were not processed.
 _TOOL_ROUND_LIMIT_MESSAGE = "一次操作太多了，请拆成几次再试。"
+# After the server finishes sending TTS, the phone may still be playing it. Keep a
+# barge-in window of at least this long so a quick "取消" is not treated as a new turn.
+_MIN_PLAYABLE_SECONDS = 0.4
 
 
 def _client_location_from_stream(stream: AudioStreamInfo) -> ClientLocation | None:
@@ -240,12 +243,13 @@ async def interrupt(self, session_id: str, reason: str) -> None:
             session.generation += 1
             turn = session.active_turn
             session.active_turn = None
-            audio_id = session.active_audio_id
-            audio_stream = session.active_audio_stream
-            session.active_audio_id = None
-            session.active_audio_stream = None
-        if session.voice_mode == "continuous" and audio_id is not None and audio_stream is not None:
-            await self._result_sink.deliver_canceled(AudioCanceled(audio_id), audio_stream)
+        canceled = False
+        if session.voice_mode == "continuous":
+            canceled = await self._cancel_playable_reply(session)
+        if not canceled:
+            async with session.lock:
+                session.active_audio_id = None
+                session.active_audio_stream = None
         self._telemetry.set_session_stage(session_id, "waiting_user")
         await self._cancel(turn)
 
@@ -332,8 +336,11 @@ async def _run_continuous(
         """Run one Agent turn per server-VAD final within a single long audio stream.
 
         A pump task consumes the ASR stream, forwarding finals to the main loop and
-        flagging the start of new speech; when speech starts while a turn's TTS is
-        playing, that turn is cancelled so the next final takes over immediately.
+        flagging the start of new speech. New speech while a turn is still running —
+        LLM, tool, or TTS — cancels that turn so the next final takes over immediately.
+        A tool that already started still commits; its result stays in the conversation
+        and TTS is skipped. After leftover playback has expired, the next final is a
+        new command.
         """
         events = self._asr.stream(chunks)
         completed_queue: asyncio.Queue[
@@ -369,6 +376,7 @@ async def pump() -> None:
                         if speech_started_at is None:
                             speech_started_at = self._monotonic()
                         self._telemetry.set_session_stage(stream.session_id, "asr")
+                        await self._cancel_playable_reply(session)
                         barge_in.set()
                     elif isinstance(event, SpeechStopped):
                         if speech_stopped_at is None:
@@ -421,13 +429,16 @@ async def pump() -> None:
                     status = await self._finish_continuous_turn(turn_task, stream)
                     self._log_timing(stream, status, timing, turn_span)
                 else:
+                    # Speech started while this turn is still running — including LLM or
+                    # tool execution, not only TTS. Skip leftover TTS; a tool that already
+                    # started still commits, and the next final sees that result.
+                    cut = await self._cancel_playable_reply(session)
                     async with session.lock:
-                        audio_id = session.active_audio_id
-                        audio_stream = session.active_audio_stream
-                    if audio_id is not None and audio_stream is not None:
-                        await self._result_sink.deliver_canceled(
-                            AudioCanceled(audio_id), audio_stream
-                        )
+                        cut = cut or session.playback_canceled
+                        session.playback_canceled = False
+                    if not turn_task.done():
+                        if not cut:
+                            self._telemetry.record_interrupt(stream.session_id)
                         turn_task.cancel()
                         await asyncio.gather(turn_task, return_exceptions=True)
                         self._log_timing(stream, "interrupted", timing, turn_span)
@@ -466,6 +477,31 @@ async def _finish_continuous_turn(
             )
             return "failed"
 
+    async def _cancel_playable_reply(self, session: ComposedSession) -> bool:
+        """Cancel a reply still being sent, or one the phone may still be playing."""
+        now = time.monotonic()
+        async with session.lock:
+            if session.active_audio_id is not None and session.active_audio_stream is not None:
+                audio_id = session.active_audio_id
+                stream = session.active_audio_stream
+            elif (
+                session.last_audio_id is not None
+                and session.last_audio_stream is not None
+                and now < session.playable_until
+            ):
+                audio_id = session.last_audio_id
+                stream = session.last_audio_stream
+            else:
+                return False
+            session.active_audio_id = None
+            session.active_audio_stream = None
+            session.last_audio_id = None
+            session.last_audio_stream = None
+            session.playable_until = 0.0
+            session.playback_canceled = True
+        await self._result_sink.deliver_canceled(AudioCanceled(audio_id), stream)
+        return True
+
     async def _act_on_transcript(
         self,
         session: ComposedSession,
@@ -750,6 +786,11 @@ async def timed_events() -> AsyncIterator[AgentEvent]:
                             return
                         session.active_audio_id = event.audio_id
                         session.active_audio_stream = stream
+                        session.last_audio_id = event.audio_id
+                        session.last_audio_stream = stream
+                        session.reply_audio_bytes = 0
+                        session.reply_sample_rate_hz = max(event.sample_rate_hz, 1)
+                        session.playback_canceled = False
                     delivery = asyncio.create_task(
                         self._result_sink.deliver_audio(reply, self._audio_chunks(queue), stream)
                     )
@@ -758,6 +799,8 @@ async def timed_events() -> AsyncIterator[AgentEvent]:
                         raise ValueError("Speech audio chunk arrived before start")
                     if delivery is not None and delivery.done():
                         delivery.result()
+                    async with session.lock:
+                        session.reply_audio_bytes += len(event.data)
                     await queue.put(event.data)
                 elif isinstance(event, SpeechAudioCompleted):
                     if queue is None or delivery is None:
@@ -766,6 +809,14 @@ async def timed_events() -> AsyncIterator[AgentEvent]:
                     await queue.put(None)
                     await delivery
                     async with session.lock:
+                        duration = session.reply_audio_bytes / max(
+                            session.reply_sample_rate_hz * 2, 1
+                        )
+                        # Remaining playback from *now*, matching realtime: generation
+                        # often finishes before the phone has sounded the last byte.
+                        session.playable_until = time.monotonic() + max(
+                            duration, _MIN_PLAYABLE_SECONDS
+                        )
                         if session.active_audio_id == event.audio_id:
                             session.active_audio_id = None
                             session.active_audio_stream = None

backend/src/timeflow/intelligence/composed/session.py

@@ -41,5 +41,11 @@ class ComposedSession:
     active_turn: TurnState | None = None
     active_audio_id: str | None = None
     active_audio_stream: StreamInfo | None = None
+    last_audio_id: str | None = None
+    last_audio_stream: StreamInfo | None = None
+    playable_until: float = 0.0
+    reply_audio_bytes: int = 0
+    reply_sample_rate_hz: int = 24_000
+    playback_canceled: bool = False
     lock: asyncio.Lock = field(default_factory=asyncio.Lock)
     turn_lock: asyncio.Lock = field(default_factory=asyncio.Lock)

backend/src/timeflow/intelligence/conversation/agent.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
 import time
@@ -407,6 +408,25 @@ async def _run_turn(
                 self._telemetry.set_session_stage(session_id, "tool")
             try:
                 result = await tool.execute(arguments)
+            except asyncio.CancelledError as exc:
+                tool_execution_ms += round((self._monotonic() - exec_started) * 1000, 1)
+                committed = getattr(exc, "result", None)
+                if isinstance(committed, str):
+                    # Schedule tools shield the business thread: cancel cannot un-commit.
+                    # Keep the tool result in history so the next utterance can react.
+                    tool_span.finish(status=tool_result_status(committed))
+                    conversation.messages.extend(
+                        [
+                            assistant_message,
+                            ToolResultMessage(
+                                tool_call_id=tool_call.call_id,
+                                content=committed,
+                            ),
+                        ]
+                    )
+                else:
+                    tool_span.finish(status="error", error_kind="cancelled")
+                raise
             except Exception as exc:
                 tool_span.finish(status="error", error_kind="exception")
                 raise AgentToolError(f"Agent tool execution failed: {tool_call.name}") from exc

backend/src/timeflow/intelligence/conversation/schedule_tools.py

@@ -97,6 +97,18 @@
 }
 
 
+class CommittedScheduleToolCancelled(asyncio.CancelledError):
+    """The business call finished, then the awaiting Agent turn was cancelled.
+
+    Schedule tools shield the worker thread so a barge-in cannot un-commit. The
+    JSON result is attached so Agent can keep the tool messages in history.
+    """
+
+    def __init__(self, result: str) -> None:
+        super().__init__()
+        self.result = result
+
+
 class ScheduleToolInputError(ValueError):
     """A schedule tool payload cannot be mapped to the business contract."""
 
@@ -140,10 +152,14 @@ async def execute(self, arguments: Mapping[str, object]) -> str:
             return _business_error_json(exc)
         except ScheduleToolInputError as exc:
             return _refusal_json(str(exc))
-        await self._notify_observer(result)
+        try:
+            await self._notify_observer(result)
+        except asyncio.CancelledError:
+            cancelled = True
+        payload = _result_json(result)
         if cancelled:
-            raise asyncio.CancelledError
-        return _result_json(result)
+            raise CommittedScheduleToolCancelled(payload)
+        return payload
 
     async def _notify_observer(
         self,