diff --git a/backend/src/timeflow/intelligence/composed/agent.py b/backend/src/timeflow/intelligence/composed/agent.py index 47aba65a..06b94a56 100644 --- a/backend/src/timeflow/intelligence/composed/agent.py +++ b/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 diff --git a/backend/src/timeflow/intelligence/composed/session.py b/backend/src/timeflow/intelligence/composed/session.py index c660d9c6..05f08fbd 100644 --- a/backend/src/timeflow/intelligence/composed/session.py +++ b/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) diff --git a/backend/src/timeflow/intelligence/conversation/agent.py b/backend/src/timeflow/intelligence/conversation/agent.py index 69052e3f..06eed97e 100644 --- a/backend/src/timeflow/intelligence/conversation/agent.py +++ b/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 diff --git a/backend/src/timeflow/intelligence/conversation/schedule_tools.py b/backend/src/timeflow/intelligence/conversation/schedule_tools.py index 4988bdfa..4904044b 100644 --- a/backend/src/timeflow/intelligence/conversation/schedule_tools.py +++ b/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, diff --git a/backend/tests/intelligence/composed/test_composed_agent.py b/backend/tests/intelligence/composed/test_composed_agent.py index 794c48b2..ef980097 100644 --- a/backend/tests/intelligence/composed/test_composed_agent.py +++ b/backend/tests/intelligence/composed/test_composed_agent.py @@ -25,6 +25,8 @@ LlmStreamCompleted, TextDelta, ToolCallDelta, + ToolDefinition, + ToolResultMessage, ) from timeflow.intelligence.conversation.tools import ToolRegistry, request_user_input_definition from timeflow.intelligence.ports import ( @@ -35,6 +37,7 @@ Transcript, ) from timeflow.intelligence.speech.tts import SpeechSegment, TtsAudioChunk, TtsCompleted +from timeflow.intelligence.telemetry import NoOpVoiceTelemetry @dataclass(frozen=True, slots=True) @@ -144,6 +147,14 @@ async def deliver_session_end(self, stream: Any) -> None: self.calls.append(("session_end", stream.session_id)) +class InterruptTelemetry(NoOpVoiceTelemetry): + def __init__(self) -> None: + self.interrupts: list[str] = [] + + def record_interrupt(self, session_id: str) -> None: + self.interrupts.append(session_id) + + async def chunks(payload: bytes = b"a" * 3200) -> AsyncIterator[bytes]: yield payload @@ -941,6 +952,162 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_continuous_speech_started_cancels_after_tts_send_while_playback_remains() -> None: + """Barge-in after the server finished sending still cancels audible leftover audio.""" + + class LongTts: + def stream(self, segments: AsyncIterable[SpeechSegment]) -> AsyncIterator[Any]: + async def events() -> AsyncIterator[Any]: + async for _segment in segments: + # 24 kHz 16-bit mono: 48000 bytes is one second of playback. + yield TtsAudioChunk(b"a" * 48_000) + yield TtsCompleted(1) + + return events() + + class BlockingSink(RecordingSink): + def __init__(self) -> None: + super().__init__() + self.audio_ended = asyncio.Event() + + async def deliver_audio( + self, + value: AudioReply, + chunks: AsyncIterator[bytes], + stream: Any, + ) -> None: + self.calls.append(("audio_start", value)) + async for chunk in chunks: + self.calls.append(("audio", chunk)) + self.calls.append(("audio_end", value.audio_id)) + self.audio_ended.set() + + class LateBargeAsr: + def __init__(self, audio_ended: asyncio.Event) -> None: + self._audio_ended = audio_ended + + def stream(self, audio: AsyncIterable[bytes]) -> AsyncIterator[Any]: + async def events() -> AsyncIterator[Any]: + async for _ in audio: + pass + yield SpeechStarted() + yield TranscriptCompleted("第一句") + await self._audio_ended.wait() + # Let _deliver_speech clear active_audio so barge-in must use leftover + # playable_until, not the in-flight send path. + await asyncio.sleep(0.01) + yield SpeechStarted() + yield TranscriptCompleted("第二句") + + return events() + + async def scenario() -> None: + sink = BlockingSink() + agent = ComposedVoiceAgent( + LateBargeAsr(sink.audio_ended), + lambda account_id, observer, client_location: Agent( + FakeLlm( + [ + [TextDelta("第一句回复。"), completed()], + [TextDelta("第二句回复。"), completed()], + ] + ), + ToolRegistry([]), + ), + LongTts(), + sink, + ) + + await agent.handle_audio(chunks(), Stream(voice_mode="continuous")) + + kinds = [kind for kind, _ in sink.calls] + assert "audio_canceled" in kinds + assert sum(1 for kind, _ in sink.calls if kind == "transcript") == 2 + + asyncio.run(scenario()) + + +def test_continuous_speech_after_playback_window_is_a_new_turn( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Once leftover playback has expired, the next utterance is a new command, not a barge-in.""" + + import timeflow.intelligence.composed.agent as composed_agent + + monkeypatch.setattr(composed_agent, "_MIN_PLAYABLE_SECONDS", 0.0) + + class TinyTts: + def stream(self, segments: AsyncIterable[SpeechSegment]) -> AsyncIterator[Any]: + async def events() -> AsyncIterator[Any]: + async for _segment in segments: + yield TtsAudioChunk(b"aa") + yield TtsCompleted(1) + + return events() + + class EndedSink(RecordingSink): + def __init__(self) -> None: + super().__init__() + self.audio_ended = asyncio.Event() + + async def deliver_audio( + self, + value: AudioReply, + chunks: AsyncIterator[bytes], + stream: Any, + ) -> None: + self.calls.append(("audio_start", value)) + async for chunk in chunks: + self.calls.append(("audio", chunk)) + self.calls.append(("audio_end", value.audio_id)) + self.audio_ended.set() + + class AfterPlaybackAsr: + def __init__(self, audio_ended: asyncio.Event) -> None: + self._audio_ended = audio_ended + + def stream(self, audio: AsyncIterable[bytes]) -> AsyncIterator[Any]: + async def events() -> AsyncIterator[Any]: + async for _ in audio: + pass + yield TranscriptCompleted("第一句") + await self._audio_ended.wait() + await asyncio.sleep(0.02) + yield SpeechStarted() + yield TranscriptCompleted("取消") + + return events() + + async def scenario() -> None: + sink = EndedSink() + telemetry = InterruptTelemetry() + agent = ComposedVoiceAgent( + AfterPlaybackAsr(sink.audio_ended), + lambda account_id, observer, client_location: Agent( + FakeLlm( + [ + [TextDelta("第一句回复。"), completed()], + [TextDelta("好的，已取消。"), completed()], + [TextDelta("好的。"), completed()], + ] + ), + ToolRegistry([]), + ), + TinyTts(), + sink, + telemetry=telemetry, + ) + + await agent.handle_audio(chunks(), Stream(voice_mode="continuous")) + + transcripts = [value.text for kind, value in sink.calls if kind == "transcript"] + assert transcripts == ["第一句", "取消"] + assert all(kind != "audio_canceled" for kind, _ in sink.calls) + assert telemetry.interrupts == [] + + asyncio.run(scenario()) + + def test_continuous_end_conversation_waits_for_farewell_audio() -> None: """Session end follows farewell text, audio, and voice.tts.end.""" @@ -1383,53 +1550,162 @@ async def scenario() -> None: asyncio.run(scenario()) -def test_continuous_speech_started_before_tts_is_ignored() -> None: - """A speech_started while the LLM is still generating (no active TTS) is ignored.""" +def test_continuous_speech_started_during_llm_interrupts_the_turn() -> None: + """Speech during LLM cancels that turn so a later '取消' is not waiting on it.""" llm_started = asyncio.Event() - release_llm = asyncio.Event() - class GatedLlm: + class GatedThenFastLlm: + def __init__(self) -> None: + self.calls = 0 + def stream(self, messages: Any, tools: Any) -> AsyncIterator[Any]: del tools + self.calls += 1 + call = self.calls async def events() -> AsyncIterator[Any]: - llm_started.set() - yield TextDelta("慢速回复") - await release_llm.wait() + if call == 1: + llm_started.set() + await asyncio.Event().wait() + yield TextDelta("慢速回复") + yield completed() + return + yield TextDelta("好的。") yield completed() return events() - class EarlySpeechAsr: - def __init__(self, llm_started: asyncio.Event) -> None: - self._llm_started = llm_started + class CancelDuringLlmAsr: + def stream(self, audio: AsyncIterable[bytes]) -> AsyncIterator[Any]: + async def events() -> AsyncIterator[Any]: + async for _ in audio: + pass + yield TranscriptCompleted("创建日程") + await llm_started.wait() + yield SpeechStarted() + yield TranscriptCompleted("取消") + + return events() + + async def scenario() -> None: + sink = RecordingSink() + telemetry = InterruptTelemetry() + agent = ComposedVoiceAgent( + CancelDuringLlmAsr(), + lambda account_id, observer, client_location: Agent( + GatedThenFastLlm(), ToolRegistry([]) + ), + FakeTts(), + sink, + telemetry=telemetry, + ) + await asyncio.wait_for( + agent.handle_audio(chunks(), Stream(voice_mode="continuous")), + timeout=2, + ) + + transcripts = [value.text for kind, value in sink.calls if kind == "transcript"] + assert transcripts == ["创建日程", "取消"] + assert telemetry.interrupts == ["session_1"] + assert all(kind != "audio_canceled" for kind, _ in sink.calls) + + asyncio.run(scenario()) + + +def test_continuous_speech_started_during_tool_keeps_committed_result() -> None: + """Barge-in during a shielded tool cannot un-commit; the next turn sees the tool result.""" + + tool_started = asyncio.Event() + cancel_seen = asyncio.Event() + release_tool = asyncio.Event() + result = '{"status":"ok","id":"schedule_1"}' + + @dataclass + class ShieldedCreateTool: + definition: Any = field( + default_factory=lambda: ToolDefinition( + "schedule_create", "创建日程", {"type": "object"} + ) + ) + started: asyncio.Event = field(default_factory=lambda: tool_started) + cancel_seen: asyncio.Event = field(default_factory=lambda: cancel_seen) + release: asyncio.Event = field(default_factory=lambda: release_tool) + committed: bool = False + + async def execute(self, arguments: Any) -> str: + del arguments + work = asyncio.create_task(self._commit()) + cancelled = False + try: + payload = await asyncio.shield(work) + except asyncio.CancelledError: + cancelled = True + self.cancel_seen.set() + payload = await work + self.committed = True + if cancelled: + error = asyncio.CancelledError() + error.result = payload # type: ignore[attr-defined] + raise error + return payload + + async def _commit(self) -> str: + self.started.set() + await self.release.wait() + return result + class CancelDuringToolAsr: def stream(self, audio: AsyncIterable[bytes]) -> AsyncIterator[Any]: async def events() -> AsyncIterator[Any]: async for _ in audio: pass - yield TranscriptCompleted("你好") - await self._llm_started.wait() + yield TranscriptCompleted("创建日程") + await tool_started.wait() yield SpeechStarted() + await cancel_seen.wait() + release_tool.set() + yield TranscriptCompleted("取消") return events() async def scenario() -> None: sink = RecordingSink() + llm = FakeLlm( + [ + [ + ToolCallDelta(0, "call_1", "schedule_create", '{"title":"开会"}'), + LlmStreamCompleted("tool_calls", None), + ], + [TextDelta("好的。"), completed()], + ] + ) + tool = ShieldedCreateTool() + telemetry = InterruptTelemetry() agent = ComposedVoiceAgent( - EarlySpeechAsr(llm_started), - lambda account_id, observer, client_location: Agent(GatedLlm(), ToolRegistry([])), + CancelDuringToolAsr(), + lambda account_id, observer, client_location: Agent(llm, ToolRegistry([tool])), FakeTts(), sink, + telemetry=telemetry, + ) + await asyncio.wait_for( + agent.handle_audio(chunks(), Stream(voice_mode="continuous")), + timeout=2, ) - task = asyncio.create_task(agent.handle_audio(chunks(), Stream(voice_mode="continuous"))) - await asyncio.wait_for(llm_started.wait(), 2) - await asyncio.sleep(0.01) - release_llm.set() - await task + transcripts = [value.text for kind, value in sink.calls if kind == "transcript"] + assert transcripts == ["创建日程", "取消"] + assert tool.committed is True + assert any( + isinstance(message, ToolResultMessage) and message.content == result + for snapshot in llm.messages + for message in snapshot + ) + assert telemetry.interrupts == ["session_1"] assert all(kind != "audio_canceled" for kind, _ in sink.calls) + replies = [value.speech_text for kind, value in sink.calls if kind == "reply"] + assert all("已创建" not in text for text in replies) asyncio.run(scenario()) diff --git a/backend/tests/intelligence/conversation/test_agent.py b/backend/tests/intelligence/conversation/test_agent.py index 38c86363..09ca893b 100644 --- a/backend/tests/intelligence/conversation/test_agent.py +++ b/backend/tests/intelligence/conversation/test_agent.py @@ -737,6 +737,64 @@ async def generate() -> AsyncIterator[LlmEvent]: assert len(conversation.messages) == 2 +@pytest.mark.asyncio +async def test_cancellation_after_tool_commit_keeps_tool_messages() -> None: + """A shielded tool still commits; the result must stay in history when the turn is cut.""" + + started = asyncio.Event() + release = asyncio.Event() + result = '{"status":"ok","id":"schedule_1"}' + + class ShieldedTool: + definition = ToolDefinition("schedule_create", "创建日程", {"type": "object"}) + + async def execute(self, arguments: Mapping[str, object]) -> str: + del arguments + work = asyncio.create_task(self._commit()) + cancelled = False + try: + payload = await asyncio.shield(work) + except asyncio.CancelledError: + cancelled = True + payload = await work + if cancelled: + error = asyncio.CancelledError() + error.result = payload # type: ignore[attr-defined] + raise error + return payload + + async def _commit(self) -> str: + started.set() + await release.wait() + return result + + llm = FakeLlm( + [ + tool_events("schedule_create", '{"title":"开会"}'), + [TextDelta("已创建。"), completed()], + ] + ) + conversation = AgentConversation() + agent = Agent(llm, ToolRegistry([ShieldedTool()])) + + async def consume() -> None: + _ = [event async for event in agent.run_turn(conversation, "创建开会")] + + task = asyncio.create_task(consume()) + await started.wait() + task.cancel() + release.set() + with pytest.raises(asyncio.CancelledError): + await task + + assert any( + isinstance(message, ToolResultMessage) and message.content == result + for message in conversation.messages + ) + assert any(isinstance(message, AssistantToolCallMessage) for message in conversation.messages) + assert len(llm.requests) == 1 + + def test_agent_turn_context_system_message_includes_utc_and_local_times() -> None: context = AgentTurnContext(datetime(2026, 8, 20, 12, 0, tzinfo=UTC), "Asia/Shanghai") diff --git a/backend/tests/intelligence/conversation/test_schedule_tools.py b/backend/tests/intelligence/conversation/test_schedule_tools.py index 0989ffad..a4a46831 100644 --- a/backend/tests/intelligence/conversation/test_schedule_tools.py +++ b/backend/tests/intelligence/conversation/test_schedule_tools.py @@ -552,11 +552,14 @@ async def succeeded(self, operation: str, result: object) -> None: task.cancel() release.set() - with pytest.raises(asyncio.CancelledError): + with pytest.raises(asyncio.CancelledError) as raised: await task assert notified.is_set() assert [call for call, _, _ in service.calls] == ["create"] + payload = getattr(raised.value, "result", None) + assert isinstance(payload, str) + assert json.loads(payload)["status"] == "ok" @pytest.mark.asyncio diff --git a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts index ade4e149..13eae540 100644 --- a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts +++ b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts @@ -531,6 +531,18 @@ export class AssistantContinuousConversationService implements AssistantApplicat ) { return; } + // tts.end 之后 currentAudioId 已清空、phase 回到 listening、回复已不再流式： + // 残留窗口里迟到的 canceled 只停播放，不要把 listening 打成 interrupted。 + // 流式回复还没发 tts.start 时 currentAudioId 也是 null，但 replyStreaming + // 仍为 true，那种 barge-in 必须继续走 interrupted，否则波形和跨轮覆盖会回归。 + if ( + this.currentAudioId === null && + this.state.phase === 'listening' && + !this.replyStreaming + ) { + void this.stopPlaybackImmediately(); + return; + } this.canceledAudioId = message.audio_id || this.currentAudioId; this.currentAudioId = null; // composed 代理打断回复时只发 voice.tts.canceled、不会再补 done=true， diff --git a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts index a3068779..86f3d95a 100644 --- a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts +++ b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts @@ -1303,6 +1303,70 @@ describe('AssistantContinuousConversationService', () => { disposeService(service); }); + it('does not enter interrupted for a stale cancellation', async () => { + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + fake.emitMessage({ + audio_id: 'audio_002', + conversation_id: 'conv_001', + payload: { + format: 'pcm_s16le', + purpose: 'reply', + sample_rate_hz: 24000, + speech_text: '', + }, + type: 'voice.tts.start', + } as AssistantServerMessage); + fake.emitMessage({ + audio_id: 'audio_001', + conversation_id: 'conv_001', + type: 'voice.tts.canceled', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getState()).toEqual({ conversationId: 'conv_001', phase: 'speaking' }); + disposeService(service); + }); + + it('does not enter interrupted for a leftover cancellation after tts.end', async () => { + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + fake.emitMessage({ + audio_id: 'audio_001', + conversation_id: 'conv_001', + payload: { + format: 'pcm_s16le', + purpose: 'reply', + sample_rate_hz: 24000, + speech_text: '', + }, + type: 'voice.tts.start', + } as AssistantServerMessage); + fake.emitMessage({ + audio_id: 'audio_001', + conversation_id: 'conv_001', + type: 'voice.tts.end', + } as AssistantServerMessage); + await flushAsync(); + expect(service.getState()).toEqual({ conversationId: 'conv_001', phase: 'listening' }); + + fake.emitMessage({ + audio_id: 'audio_001', + conversation_id: 'conv_001', + type: 'voice.tts.canceled', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getState()).toEqual({ conversationId: 'conv_001', phase: 'listening' }); + disposeService(service); + }); + it('ignores the canceled stream end that arrives after interruption', async () => { const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection });