diff --git a/backend/src/timeflow/gateway/websocket/agent_ports.py b/backend/src/timeflow/gateway/websocket/agent_ports.py index d44812c3..268ac4ac 100644 --- a/backend/src/timeflow/gateway/websocket/agent_ports.py +++ b/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.""" diff --git a/backend/src/timeflow/gateway/websocket/handlers/agent_result.py b/backend/src/timeflow/gateway/websocket/handlers/agent_result.py index 1785643f..a22c929c 100644 --- a/backend/src/timeflow/gateway/websocket/handlers/agent_result.py +++ b/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()) diff --git a/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py b/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py index ca8871ef..54727af2 100644 --- a/backend/src/timeflow/gateway/websocket/handlers/voice_stream.py +++ b/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() diff --git a/backend/src/timeflow/gateway/websocket/messages/agent.py b/backend/src/timeflow/gateway/websocket/messages/agent.py index bb2b1cc7..de36ae69 100644 --- a/backend/src/timeflow/gateway/websocket/messages/agent.py +++ b/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): diff --git a/backend/src/timeflow/gateway/websocket/messages/dialogue.py b/backend/src/timeflow/gateway/websocket/messages/dialogue.py index 310118df..3fbacf26 100644 --- a/backend/src/timeflow/gateway/websocket/messages/dialogue.py +++ b/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): diff --git a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py index f21e3ea5..7ff2edee 100644 --- a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py +++ b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py @@ -59,24 +59,26 @@ async def close(self) -> None: class Observer(Protocol): """Where this adapter reports what the model says; restated, never imported.""" - async def heard(self, text: str) -> None: - """The model reported what the user said.""" + async def heard(self, text: str, turn_id: str | None = None) -> None: + """The model reported what the user said, and which input item it transcribes.""" ... async def user_started_speaking(self) -> None: """The vendor detected user speech, including a barge-in.""" ... - async def spoke(self, text: str) -> None: - """The model reported the words it is saying.""" + async def spoke(self, text: str, turn_id: str | None = None) -> None: + """The model reported the words it is saying, and which input item they answer.""" ... async def audio(self, data: bytes) -> None: """One chunk of the model's own speech, decoded to raw bytes.""" ... - async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: - """The model asked for a tool to run.""" + async def tool_requested( + self, call_id: str, name: str, arguments: dict[str, Any], turn_id: str | None = None + ) -> None: + """The model asked for a tool to run, as part of answering one utterance.""" ... async def turn_completed(self) -> None: @@ -195,6 +197,35 @@ def __init__( self._speech_started_at: float | None = None self._response_started_at: float | None = None self._first_audio_at: float | None = None + # Diagnostic only: counts every response.create we send against every + # response.created the vendor reports back. + self._responses_sent = 0 + self._responses_created = 0 + # Continuous mode only: true once real user speech has started a reply we are + # legitimately waiting on, or a tool result has asked for a follow-up; false again + # the instant that reply settles. The vendor has been observed to start an extra + # response.created on its own -- no new speech_started, nothing asked for -- + # sometimes minutes after a reply already settled, re-answering the same query + # under a fresh reply_id. _pump_continuous cancels a response.created it sees + # while this is false, before any of its text or audio can reach the client. + self._expecting_response = False + # True from the moment cancel_response() sends response.cancel until a new + # response actually starts. A cancel can race a response that the vendor was + # already finishing on its own -- observed in production as an "error" event + # back, "Conversation has no active response" -- which _pump_continuous would + # otherwise treat as fatal and hang up the whole call over a no-op. Held open + # until response.created rather than cleared on the next event, because the + # vendor having finished that response is exactly what makes its response.done + # arrive in between: a one-event allowance is spent on that response.done and + # never reaches the error it was meant for. + self._cancel_pending = False + # The vendor's own id for the stretch of user speech the current reply answers. + # It rides on speech_started and input_audio_buffer.committed, and the vendor + # stamps the matching transcript with it too -- which is what makes pairing a + # reply with the words that prompted it exact instead of positional. The + # transcript routinely arrives after the reply it belongs to has already + # started, so arrival order cannot do this job. + self._input_item_id: str | None = None async def configure(self, instructions: str, tools: list[dict[str, Any]]) -> None: """Set the session up before any audio; turn_detection only takes effect here.""" @@ -237,6 +268,7 @@ async def finish_input(self) -> None: await self._send({"type": "input_audio_buffer.commit"}) await self._send({"type": "response.create"}) self._open_responses += 1 + self._responses_sent += 1 async def send_tool_result(self, call_id: str, output: str, *, respond: bool = True) -> None: """Write a tool's output back and mark whether it should be followed by a reply. @@ -262,6 +294,7 @@ async def send_tool_result(self, call_id: str, output: str, *, respond: bool = T ) if respond: self._followup_requested = True + self._expecting_response = True else: self._followup_suppressed = True @@ -281,6 +314,7 @@ async def cancel_response(self) -> None: self._followup_requested = False self._followup_suppressed = False await self._send({"type": "response.cancel"}) + self._cancel_pending = True async def close(self) -> None: """Close the underlying connection, ignoring an already-closed one.""" @@ -341,6 +375,14 @@ async def _pump_single_turn(self, observer: Observer) -> None: if kind == "conversation.item.input_audio_transcription.completed": await observer.heard(str(event.get("transcript", ""))) elif kind == "response.created": + self._responses_created += 1 + if self._responses_created > self._responses_sent: + logger.warning( + "realtime vendor started a response we never requested " + "(push_to_talk): sent=%s created=%s", + self._responses_sent, + self._responses_created, + ) self._response_started_at = self._clock() elif kind == "response.audio_transcript.delta": spoken += str(event.get("delta", "")) @@ -380,8 +422,10 @@ async def _pump_single_turn(self, observer: Observer) -> None: self._followup_requested = False self._followup_suppressed = False if wants_followup: + logger.info("realtime sending follow-up response.create after a tool result") await self._send({"type": "response.create"}) self._open_responses += 1 + self._responses_sent += 1 continue if self._open_responses <= 0: return @@ -398,6 +442,18 @@ async def _pump_continuous(self, observer: Observer) -> None: """ spoken = "" self._responding = False + # A held session serves this conversation's next stream too, so everything this + # loop reasons about has to start over here rather than carry the last stream's + # answers into a call that has heard nothing yet: an expectation left standing + # would wave through the first spontaneous response, and a playback estimate + # left standing would report the opening speech_started as a barge-in on a reply + # from the previous call. Within one stream `suppressed` shadows a stale estimate; + # across streams nothing does. A follow-up already asked for and not yet delivered + # is the one expectation that legitimately outlives the loop that created it. + self._expecting_response = self._followup_requested + self._playable_until = 0.0 + self._reply_bytes = 0 + self._cancel_pending = False # True from the moment we cancel a reply until the next one starts: the vendor # may still emit a few queued deltas for the cancelled reply before it catches up. suppressed = False @@ -408,7 +464,9 @@ async def _pump_continuous(self, observer: Observer) -> None: kind = event.get("type") if kind == "input_audio_buffer.speech_started": + self._input_item_id = _item_id(event) or self._input_item_id self._speech_started_at = self._clock() + self._expecting_response = True # A real barge-in even once generation has finished: the phone can still # be sounding out audio that was already fully sent (see _playable_until). await observer.user_started_speaking() @@ -417,22 +475,43 @@ async def _pump_continuous(self, observer: Observer) -> None: await self.cancel_response() await observer.interrupted() elif kind == "response.created": + self._responses_created += 1 + # A response is running again, so any cancel still awaiting its verdict + # is settled: whatever the vendor says from here on is about this one. + self._cancel_pending = False + if not self._expecting_response: + logger.warning( + "realtime vendor started a response with no user speech or " + "requested follow-up behind it -- cancelling it before its " + "content can reach the client" + ) + suppressed = True + self._responding = True + await self.cancel_response() + continue self._responding = True suppressed = False spoken = "" self._reply_bytes = 0 self._response_started_at = self._clock() self._first_audio_at = None + elif kind == "input_audio_buffer.committed": + # Sent immediately before the response.created that answers it, so this + # is the closest reading of "which utterance the next reply is for". + self._input_item_id = _item_id(event) or self._input_item_id elif kind == "conversation.item.input_audio_transcription.completed": - await observer.heard(str(event.get("transcript", ""))) + # Its own item_id, not the tracked one: a transcript that arrives late + # still says which utterance it belongs to, even if the user has since + # started another. + await observer.heard(str(event.get("transcript", "")), _item_id(event)) elif kind == "response.audio_transcript.delta" and not suppressed: spoken += str(event.get("delta", "")) - await observer.spoke(spoken) + await observer.spoke(spoken, self._input_item_id) elif kind == "response.audio_transcript.done" and not suppressed: final = str(event.get("transcript", "")) if final and final != spoken: spoken = final - await observer.spoke(spoken) + await observer.spoke(spoken, self._input_item_id) elif kind == "response.audio.delta" and not suppressed: decoded = _decode_audio(event.get("delta")) if decoded: @@ -445,7 +524,10 @@ async def _pump_continuous(self, observer: Observer) -> None: if requested is None: await observer.failed("realtime session sent an unusable tool call") return - await observer.tool_requested(**requested) + # Named here rather than left to spoke(): a tool call happens before any + # wording exists for this reply, so a question it raises would otherwise + # be stamped with the previous turn's id -- worse than carrying none. + await observer.tool_requested(**requested, turn_id=self._input_item_id) elif kind == "response.done": usage = _parse_usage(event) if usage is not None: @@ -470,16 +552,32 @@ async def _pump_continuous(self, observer: Observer) -> None: self._followup_requested = False self._followup_suppressed = False if wants_followup: + logger.info("realtime sending follow-up response.create after a tool result") await self._send({"type": "response.create"}) self._open_responses += 1 + self._responses_sent += 1 continue self._responding = False + # Not reset when suppressed: this response.done is the trailing tail of a + # reply a barge-in already cancelled, and that barge-in's own + # speech_started is what set this true -- the new reply it is about to + # start is exactly what we are still legitimately waiting on. + if not suppressed: + self._expecting_response = False # The bytes just sent still take this long to actually play out on the # phone; a barge-in landing before then is still cancelling something # audible, even though generation itself has already finished. self._playable_until = self._clock() + self._reply_bytes / _OUTPUT_BYTES_PER_SECOND await observer.turn_completed() elif kind == "error": + if self._cancel_pending and _is_benign_cancel_race(event): + self._cancel_pending = False + logger.info( + "realtime response.cancel raced a response the vendor had " + "already finished on its own -- nothing to actually cancel, " + "continuing the call" + ) + continue await observer.failed(_error_message(event)) return @@ -543,6 +641,12 @@ def _tool_request(event: dict[str, Any]) -> dict[str, Any] | None: return {"call_id": call_id, "name": name, "arguments": arguments} +def _item_id(event: dict[str, Any]) -> str | None: + """Lift the vendor's conversation item id out of an event, when it carries one.""" + item_id = event.get("item_id") + return item_id if isinstance(item_id, str) and item_id else None + + def _error_message(event: dict[str, Any]) -> str: """Extract a readable message from a vendor error event.""" error = event.get("error") @@ -553,6 +657,17 @@ def _error_message(event: dict[str, Any]) -> str: return "realtime session reported an error" +def _is_benign_cancel_race(event: dict[str, Any]) -> bool: + """Whether an error event is the vendor saying there was nothing to cancel. + + Sent back when our own response.cancel loses a race against the vendor finishing + that same response on its own a moment earlier -- expected, not a real failure, + and safe to ignore: we already treat that response as discarded either way at both + cancel_response() call sites, whether the vendor got to finish it or not. + """ + return "no active response" in _error_message(event).lower() + + def _parse_usage(event: dict[str, Any]) -> dict[str, Any] | None: """Flatten a response.done event's usage block, or None when it has none. diff --git a/backend/src/timeflow/intelligence/ports.py b/backend/src/timeflow/intelligence/ports.py index 149c18de..cc5a2b4e 100644 --- a/backend/src/timeflow/intelligence/ports.py +++ b/backend/src/timeflow/intelligence/ports.py @@ -66,6 +66,11 @@ class Transcript: text: str language: str duration_ms: int + # Which stretch of user speech this transcribes, when the producer can say. Carried + # so the client can pair it with the reply that answers it instead of guessing from + # arrival order -- the realtime vendor routinely sends a transcript after the reply + # it belongs to has already started. None from producers with no such id. + turn_id: str | None = None @dataclass(frozen=True, slots=True) @@ -75,6 +80,8 @@ class ReplyText: reply_id: str speech_text: str done: bool = False + # Which stretch of user speech this answers; see Transcript.turn_id. + turn_id: str | None = None @dataclass(frozen=True, slots=True) @@ -86,6 +93,8 @@ class DialogueQuestion: speech_text: str required_response: str | None = None candidates: tuple[dict[str, Any], ...] = () + # Which stretch of user speech this asks about; see Transcript.turn_id. + turn_id: str | None = None @dataclass(frozen=True, slots=True) diff --git a/backend/src/timeflow/intelligence/realtime/agent.py b/backend/src/timeflow/intelligence/realtime/agent.py index 2b275a49..d1c256a9 100644 --- a/backend/src/timeflow/intelligence/realtime/agent.py +++ b/backend/src/timeflow/intelligence/realtime/agent.py @@ -2,6 +2,7 @@ import asyncio import contextlib +import json import logging import time from collections.abc import AsyncIterator, Awaitable, Callable @@ -74,6 +75,15 @@ def new_question_id() -> str: return f"question_{uuid4().hex}" +# Same list the composed agent uses for the same fallback (conversation/agent.py). +_FAREWELL_MARKERS = ("再见", "拜拜", "先这样", "就这样") + + +def _is_farewell(text: str) -> bool: + """Return True when a reply's wording is a farewell that should end the session.""" + return any(marker in text for marker in _FAREWELL_MARKERS) + + def _client_location_from_stream(stream: StreamInfo) -> ClientLocation | None: """Build a validated client position from the stream's raw wire fields, or None. @@ -366,12 +376,19 @@ def __init__( # phone is playing -- the model finishes generating well before playback ends. self._last_audio_id: str | None = None self._reply_id: str | None = None + # The vendor's id for the utterance the current reply answers, set by spoke(). + # Not reset between replies: a question asked from a tool call happens before + # any wording has been spoken for that reply, and must still name its turn. + self._turn_id: str | None = None self._spoken = "" self._purpose = REPLY_PURPOSE self._input_bytes = 0 self._audio: asyncio.Queue[bytes | None] = asyncio.Queue() self._speaking: asyncio.Task[None] | None = None self._ends_conversation = False + # Whether a tool ran as part of the reply currently being settled. Read by + # _finish_reply()'s farewell fallback below, then reset for the next reply. + self._tool_called = False self.failure: str | None = None def note_input_chunk(self, chunk_bytes: int) -> None: @@ -386,30 +403,49 @@ async def user_started_speaking(self) -> None: """Occupy ASR while the vendor is hearing the user, including barge-ins.""" self._telemetry.set_session_stage(self._stream.session_id, "asr") - async def heard(self, text: str) -> None: + async def heard(self, text: str, turn_id: str | None = None) -> None: """Push what the user was heard to say.""" if not text: logger.info("realtime model returned an empty transcript") self._input_bytes = 0 return + # Logged in full, and interpolated rather than passed via extra= for the reason + # failed() states below. Without it a log shows a reply arriving with no + # transcript behind it and no way to tell whether the user said nothing, the + # vendor dropped the transcription event, or the reply belongs to some other + # turn entirely -- every one of those reads the same. Quoted so leading and + # trailing whitespace is visible. + logger.info("realtime heard the user say: %r", text) await self._result_sink.deliver_transcript( HeardSpeech( text=text, language=ASSUMED_LANGUAGE, duration_ms=self._input_bytes // _INPUT_BYTES_PER_MS, + turn_id=turn_id, ), self._stream, ) self._telemetry.set_session_stage(self._stream.session_id, "llm") self._input_bytes = 0 - async def spoke(self, text: str) -> None: + async def spoke(self, text: str, turn_id: str | None = None) -> None: """Push the reply's wording so far, and keep it for the audio's opening message.""" + # Kept for the closing done=true this reply gets in _finish_reply, and for any + # question it asks along the way: both have to name the same utterance the + # streaming updates did, or the client cannot tell they are one turn. + self._turn_id = turn_id if self._reply_id is None: self._reply_id = self._reply_id_factory() + logger.info( + "realtime turn started a new spoken reply: reply_id=%s account_id=%s " + "conversation_id=%s", + self._reply_id, + self._stream.account_id, + self._stream.conversation_id, + ) self._spoken = text await self._result_sink.deliver_reply_text( - ReplyText(reply_id=self._reply_id, speech_text=text), self._stream + ReplyText(reply_id=self._reply_id, speech_text=text, turn_id=turn_id), self._stream ) async def audio(self, data: bytes) -> None: @@ -420,12 +456,19 @@ async def audio(self, data: bytes) -> None: self._speaking = asyncio.create_task(self._speak()) await self._audio.put(data) - async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: + async def tool_requested( + self, call_id: str, name: str, arguments: dict[str, Any], turn_id: str | None = None + ) -> None: """Run the tool, tell the client what came of it, and let the model continue. The client needs the data to display and the model needs it to say anything true about it, so both are answered from the one call. """ + self._tool_called = True + # Before _ask() below can need it: a question raised here is the first thing this + # turn says, so nothing else has named the turn yet. + if turn_id is not None: + self._turn_id = turn_id if self._tools is None: logger.warning( "realtime model asked for a tool while none are registered", @@ -436,6 +479,16 @@ async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any occupy_tool = name not in {"end_conversation", "request_user_input"} if occupy_tool: self._telemetry.set_session_stage(self._stream.session_id, "tool") + # Ahead of the call, not folded into the "executed" line below: run() raises on a + # failing tool and can sit there on a slow one, and the arguments are exactly what + # is wanted in both cases -- a request line with no executed line after it says + # which call hung or blew up. Interpolated for the reason failed() states below; + # default=str keeps an unexpected value from turning a log line into a crash. + logger.info( + "realtime tool requested: tool=%s arguments=%s", + name, + json.dumps(arguments, ensure_ascii=False, default=str), + ) tool_started = self._stopwatch() try: result = await self._tools.run(name, arguments) @@ -487,6 +540,7 @@ async def _ask(self, question: dict[str, Any]) -> None: speech_text=str(question["speech_text"]), required_response=question["required_response"], candidates=question["candidates"], + turn_id=self._turn_id, ), self._stream, ) @@ -556,9 +610,27 @@ async def _finish_reply(self, *, canceled: bool) -> None: if self._spoken: assert self._reply_id is not None await self._result_sink.deliver_reply_text( - ReplyText(reply_id=self._reply_id, speech_text=self._spoken, done=True), + ReplyText( + reply_id=self._reply_id, + speech_text=self._spoken, + done=True, + turn_id=self._turn_id, + ), self._stream, ) + # The model sometimes says goodbye without remembering to call + # end_conversation. Composed guards the same gap by pattern-matching a + # no-tool-call reply; mirrored here rather than trusted to the prompt -- + # otherwise the call is left open, listening, after its own farewell. + # Scoped to a reply that called no tool at all, same as composed: a reply + # that also did something ("帮你删掉了,再见") should not risk a false hit. + if ( + not canceled + and not self._ends_conversation + and not self._tool_called + and _is_farewell(self._spoken) + ): + self._ends_conversation = True if self._speaking is not None: if canceled: assert self._audio_id is not None @@ -583,6 +655,7 @@ async def _finish_reply(self, *, canceled: bool) -> None: self._audio_id = None self._speaking = None self._purpose = REPLY_PURPOSE + self._tool_called = False if self._ends_conversation: # Only after the settling above -- any farewell this reply spoke has already # gone out in full, so telling the client to hang up here never cuts it short. diff --git a/backend/src/timeflow/intelligence/realtime/instructions.py b/backend/src/timeflow/intelligence/realtime/instructions.py index bd3b0096..145f831d 100644 --- a/backend/src/timeflow/intelligence/realtime/instructions.py +++ b/backend/src/timeflow/intelligence/realtime/instructions.py @@ -6,14 +6,7 @@ _WEEKDAYS = ("星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日") -# A/B flag for exp01's finding: the fixed per-response floor (this prompt) dwarfs what -# history growth adds, since a realtime session re-sends it on every single response. -# Flip to True to bill a real call against the compressed variant and compare with the -# conservative default. Once one variant is picked from real usage data, delete the -# other branch and this flag -- this is a measurement tool, not a permanent setting. -_AGGRESSIVE = False - -_ROLE_CONSERVATIVE = """你是 TimeFlow 的日程助手,帮用户用说话的方式管理日程和提醒。 +_ROLE = """你是 TimeFlow 的日程助手,帮用户用说话的方式管理日程和提醒。 语言与口吻 - 始终用中文回答,无论用户说什么语言。 @@ -33,25 +26,30 @@ - schedule_create 新建日程,schedule_update 修改,schedule_delete 删除。 - location_search 搜索真实地点。 - request_user_input 向用户提问。 -- end_conversation 结束这次语音对话,见工具说明里的触发词;想道别就先说再调用。 +- end_conversation 结束这次语音对话。用户说「结束对话」「先这样」「不用了」「退出语音 + 模式」「停止监听」这类明确想停下来的话时调用,想道别就先说再调用。 + +调用工具前不用开口——不管是查、建、改、删还是搜地点,都不要说「我查一下」「稍等」 +这类过渡话,等工具有结果了再一次性把话说完。 改动日程的规矩 - 地点型日程必须有地点,缺了就问(见下面「地点怎么定」);其余没提到的字段不用问, 按「没提到的字段怎么补」直接用默认值建。 - 改和删都要先用 schedule_query 找到那条日程,拿它的 id 和 revision 去调用,不要凭印象编 id。 - 这一步不用开口,不要说「我查一下」「稍等」这类话,查完直接接着往下做,把话留到最后 - 一次性说结果。 - 删除之前先确认一次,question_kind 用 confirmation,把要删的那条说清楚。 - 工具报 failed 就说没做成,说明原因。绝不要把没成功的说成已经办好了。 没提到的字段怎么补 - 没说标题,用「新建日程」。 -- 没说开始时间,用当前时间之后的下一个整点。 +- 连哪天都没说,才用当前时间之后的下一个整点起建。只要说了具体是哪天(今天、明天、 + 下周三、几月几号……),哪怕没说几点,也不要自己拿"下一个整点"套到那天上去猜—— + 调 request_user_input,question_kind 用 missing_field,required_response 填 + start_time,问清楚几点再建。 - 非全天日程没说结束时间,按开始时间往后算一小时。 - 全天日程只说了一天,就当整个自然日,不用问是不是只有这一天。 -- 没说要重复,就当不重复。 +- 没说要重复,就当不重复,不要反问要不要设成重复。 - 没提到地点,就当时间型日程处理,不要为了填 latitude/longitude 去调用 location_search - 或编一个地点——地点型日程的地点仍然必须问清楚。 + 或编一个地点——地点型日程的地点仍然必须问清楚,这条只管时间型日程不要凭空加地点。 - 没说提醒方式,默认建一个 reminder_strength 为 medium 的提醒:非全天日程用 reminder_type=before_start、reminder_offset_minutes=15(开始前 15 分钟);全天日程用 reminder_type=at_time、reminder_trigger_at 填当天上午 10:00(带时区偏移)。 @@ -61,9 +59,10 @@ location_search,不要凭印象编地址或经纬度。 - 只搜到一条且明确对得上用户说的地方,直接用。 - 搜到两条分不清是哪个,调用 request_user_input 让用户选,question_kind 用 - ambiguous_target,candidates 里放这两条的名称和地址备用,不要念经纬度。客户端不弹 - 候选卡片,用户只能靠听你说的话来选,所以 speech_text 必须把两条地点说清楚、报出 - 顺序,例如「第一个是万达广场银川路店,第二个是万达广场银川路辅路店,你要去哪个」。 + ambiguous_target。客户端不弹候选卡片,用户只能靠听你说的话来选,所以 speech_text + 必须把两条地点说清楚、报出顺序,例如「第一个是万达广场银川路店,第二个是万达广场 + 银川路辅路店,你要去哪个」,不能只说一句「找到两个类似的」就完事——那样用户没法选。 + 不要念经纬度,也不要把搜索结果原文抄进工具参数里。 - 用户回答「第一个」「第二个」,或者直接说出某条候选的名字,都算选中了对应那条, 接着用它的地址和坐标建日程,不用再跟用户确认一遍选的是哪个。 - 什么都没搜到,如实说没找到,不要编一个。 @@ -73,7 +72,11 @@ 什么时候提问 - 缺少必要信息时调用 request_user_input,question_kind 用 missing_field,required_response 写缺哪个字段。 -- 用户指代不明(「那个会」「上次那个」)时,**先用 schedule_query 查一遍**,再调用 request_user_input,question_kind 用 ambiguous_target,把查到的几条放进 candidates。不要空着 candidates 就问,客户端要靠它把选项列出来给用户点。 +- 用户指代不明(「那个会」「上次那个」)时,**先用 schedule_query 查一遍**,再调用 + request_user_input,question_kind 用 ambiguous_target。客户端不弹候选卡片,用户 + 只能靠听你说的话来选,所以 speech_text 必须把查到的几条说清楚、报出顺序,例如 + 「找到两个开会,第一个是今天下午三点,第二个是每周一早上十点,你要删哪个」。 + 只说一句「找到两个」用户没法选。不要把日程原文抄进工具参数里——说出来就够了。 - 一次只问一件事。缺日期又缺时间,先问日期。 - 调用 request_user_input 之后,把 speech_text 的原话说出来,让用户听见问题。除此之外不要多说。 - 能自己想明白的不要问。「明天」「下周三」这类你能算出来的,直接算,不要反问用户是哪天。 @@ -89,61 +92,6 @@ - 不要念日程的 id 或版本号。 """ -_ROLE_AGGRESSIVE = """你是 TimeFlow 的日程助手,帮用户用说话的方式管理日程和提醒。 - -语言与口吻 -- 始终用中文回答。像朋友说话,简短自然,一两句话说完,不客套、不复述原话。 - -输出格式 -- 只输出纯文本,不要 emoji/Markdown/列表符号/标题。 -- 数字、时间、地点直接说出来,如「明天下午三点,203」,不用「15:00」这种书面形式。 - -时间的理解 -- 相对时间说法(今天/明天/下周三/这周末等)按当前时间换算成具体日期。 -- 没说上下午的时间点按最近合理时间理解,白天「三点」通常指下午。不编用户没说的信息。 - -能做什么 -- schedule_query 查询;schedule_create/update/delete 增改删;location_search 搜地点; - request_user_input 提问;end_conversation 结束对话(触发词见工具说明),道别先说再调用。 - -改动日程的规矩 -- 地点型日程缺地点必须问(见「地点怎么定」);其余缺省字段直接按默认值处理,不用问。 -- 改/删前先用 schedule_query 定位,拿 id 和 revision,不要凭印象编。这一步不开口, - 查完直接往下做,话留到最后一次性说结果。 -- 删除前用 confirmation 问题确认一次,说清要删的是哪条。 -- 工具报 failed 就说没做成并说明原因,绝不说成已办好。 - -缺省字段默认值 -- 无标题→「新建日程」;无开始时间→当前时间后下一个整点;非全天无结束时间→开始后一小时; - 全天日程只说一天→当整个自然日;未说重复→不重复;未提地点→按时间型处理,不编地点。 -- 未说提醒→medium 强度:非全天用 before_start/提前15分钟;全天用 at_time/当天10:00(带时区)。 - -地点怎么定 -- 有具体地点(非「公司」「家」这类用户自明的说法)先调 location_search,不编地址/经纬度。 -- 一条明确匹配直接用;搜到多条用 request_user_input(ambiguous_target)让用户选, - candidates 放名称地址,speech_text 必须报清楚顺序和名字,例如「第一个是万达广场银川路 - 店,第二个是万达广场银川路辅路店,你要去哪个」,不念经纬度。 -- 用户答「第一个」/说出候选名字即视为选中,直接用其地址坐标建日程,不用二次确认。 -- 没搜到如实说没找到;provider_unavailable 就说位置搜索暂不可用,其余信息照常处理。 -- 选定候选后,location_search 返回的坐标原样抄给 create/update,不要重新估算。 - -什么时候提问 -- 缺必要信息用 missing_field,required_response 写缺的字段名。 -- 指代不明(「那个会」)先 schedule_query 查一遍,再用 ambiguous_target 把结果放进 - candidates 问,不能空着 candidates。 -- 一次只问一件事;自己能算出来的不要问。调用后把 speech_text 原话说出来,别多说。 - -用户回答之后 -- 记得上一轮问了什么;回答是补缺的那部分,接着做原来的事,不重新问、不复述。 - 补齐后还缺别的就接着问下一件。 - -查询之后怎么说 -- 先说条数,再逐条说时间/标题/地点,一条一句;没查到直接说没有,不建议改条件重试; - 不念 id 或版本号。 -""" - -_ROLE = _ROLE_AGGRESSIVE if _AGGRESSIVE else _ROLE_CONSERVATIVE - def build_instructions(timezone: str, now: Callable[[], datetime] | None = None) -> str: """Return the instructions with the current time in the client's zone stated at the top.""" diff --git a/backend/src/timeflow/intelligence/realtime/ports.py b/backend/src/timeflow/intelligence/realtime/ports.py index ea0c3012..961e2189 100644 --- a/backend/src/timeflow/intelligence/realtime/ports.py +++ b/backend/src/timeflow/intelligence/realtime/ports.py @@ -6,24 +6,26 @@ class TurnObserver(Protocol): """What a realtime session reports while a turn runs, in this layer's own terms.""" - async def heard(self, text: str) -> None: - """The model reported what the user said.""" + async def heard(self, text: str, turn_id: str | None = None) -> None: + """The model reported what the user said, and which utterance it transcribes.""" ... async def user_started_speaking(self) -> None: """The vendor detected user speech, including a barge-in.""" ... - async def spoke(self, text: str) -> None: - """The model reported the words it is saying.""" + async def spoke(self, text: str, turn_id: str | None = None) -> None: + """The model reported the words it is saying, and which utterance they answer.""" ... async def audio(self, data: bytes) -> None: """One chunk of the model's own speech, already decoded to raw bytes.""" ... - async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: - """The model asked for a tool to run before it continues.""" + async def tool_requested( + self, call_id: str, name: str, arguments: dict[str, Any], turn_id: str | None = None + ) -> None: + """The model asked for a tool to run before it continues answering one utterance.""" ... async def turn_completed(self) -> None: diff --git a/backend/src/timeflow/intelligence/realtime/schedule_tools.py b/backend/src/timeflow/intelligence/realtime/schedule_tools.py index e2c0f6a2..c4330ca9 100644 --- a/backend/src/timeflow/intelligence/realtime/schedule_tools.py +++ b/backend/src/timeflow/intelligence/realtime/schedule_tools.py @@ -254,11 +254,6 @@ class ToolResult: "type": "string", "description": "希望用户补充的字段名,例如 start_time、location", }, - "candidates": { - "type": "array", - "items": {"type": "object"}, - "description": "匹配到多条时的候选日程,供客户端展示", - }, }, "required": ["question_kind", "speech_text"], }, @@ -318,6 +313,18 @@ def __init__( self._client_location = client_location self._location_context: LocationSearchContext | None = None self._telemetry = telemetry if telemetry is not None else NOOP_TELEMETRY + # Set by _ask() when the question it just asked was recurrence_scope or + # confirmation; not consumed by a delete, so one confirmation still covers a + # batch of deletes the model makes in the same breath ("delete all three") -- + # only asking a different-kind question moves it off the delete flow. The + # composed agent gates schedule_delete the same way (see conversation/agent.py's + # _delete_authorized) by inspecting the user's own answer text -- realtime has + # no equivalent seam (the vendor session decides on its own when to call the + # tool again), so this only proves a qualifying question was asked at some + # point, not that the answer to it was affirmative. Still closes the gap this + # was added for: the model calling schedule_delete straight off a command with + # no question in between at all. + self._delete_authorization: str | None = None # The service is synchronous and reaches Postgres over a socket, so every call goes # to a worker thread: awaiting it inline would stall the loop streaming this turn's @@ -348,6 +355,16 @@ async def _run_tool(self, name: str, arguments: dict[str, Any]) -> ToolResult: if name == LOCATION_SEARCH: return await self._location_search(arguments) + if name == SCHEDULE_DELETE and self._delete_authorization is None: + # A delete is irreversible; refuse it on the spot rather than trust the + # prompt alone -- the model does sometimes skip the confirmation or + # recurrence-scope question it was told to ask first. + return _refusal( + "删除是不可逆操作,必须先调用 request_user_input(confirmation 或 " + "recurrence_scope)向用户确认删除目标或问清楚周期删除范围,得到回答后" + "才能 schedule_delete。" + ) + # Normalize datetime fields before mapping arguments = normalize_datetime_args(arguments, self._timezone) @@ -453,9 +470,9 @@ def _ask(self, arguments: dict[str, Any]) -> ToolResult: if not isinstance(speech_text, str) or not speech_text.strip(): return _refusal("speech_text 不能为空,要写出问用户的原话。") - candidates = _candidates(arguments.get("candidates")) - if kind == "ambiguous_target" and not candidates: - return _refusal("ambiguous_target 必须提供非空 candidates。先用 schedule_query 查询。") + # A qualifying question supersedes any earlier one; any other kind (missing_field, + # ambiguous_target) means the model moved off the delete flow, so it drops too. + self._delete_authorization = kind if kind in ("recurrence_scope", "confirmation") else None required = arguments.get("required_response") return ToolResult( @@ -464,7 +481,13 @@ def _ask(self, arguments: dict[str, Any]) -> ToolResult: "question_kind": kind, "speech_text": speech_text.strip(), "required_response": required if isinstance(required, str) and required else None, - "candidates": candidates, + # Always empty from this agent: making the model re-emit every matched + # schedule here cost 477 output tokens and 7.4s of generation for two of + # them (then another 4s, because the first attempt arrived as a JSON + # string and was refused) -- for a field no client reads. The choices go + # into speech_text instead, which is what the user actually hears. The + # protocol keeps the field for the composed agent, which still fills it. + "candidates": (), }, ) @@ -577,13 +600,6 @@ def _local_text(instant: datetime | None, tz: ZoneInfo) -> str: return instant.astimezone(tz).strftime("%Y-%m-%d %H:%M") -def _candidates(value: Any) -> tuple[dict[str, Any], ...]: - """Keep the choices that are actually objects, dropping anything else.""" - if not isinstance(value, list): - return () - return tuple(item for item in value if isinstance(item, dict)) - - def _json_value(value: object) -> object: if isinstance(value, datetime): return value.isoformat() diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py index 1f85e67d..d6e7dfd3 100644 --- a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -62,26 +62,34 @@ class RecordingObserver: def __init__(self) -> None: """Start with nothing observed.""" self.calls: list[tuple[str, Any]] = [] + # Recorded apart from calls so the existing assertions on wording stay readable; + # only the tests about pairing an utterance with its reply look at these. + self.turn_ids: list[tuple[str, str | None]] = [] - async def heard(self, text: str) -> None: + async def heard(self, text: str, turn_id: str | None = None) -> None: """Record the user's transcript.""" self.calls.append(("heard", text)) + self.turn_ids.append(("heard", turn_id)) async def user_started_speaking(self) -> None: """Record that the vendor detected user speech.""" self.calls.append(("user_started_speaking", None)) - async def spoke(self, text: str) -> None: + async def spoke(self, text: str, turn_id: str | None = None) -> None: """Record the assistant's own words.""" self.calls.append(("spoke", text)) + self.turn_ids.append(("spoke", turn_id)) async def audio(self, data: bytes) -> None: """Record one decoded audio chunk.""" self.calls.append(("audio", data)) - async def tool_requested(self, call_id: str, name: str, arguments: dict[str, Any]) -> None: + async def tool_requested( + self, call_id: str, name: str, arguments: dict[str, Any], turn_id: str | None = None + ) -> None: """Record a tool call request.""" self.calls.append(("tool", (call_id, name, arguments))) + self.turn_ids.append(("tool", turn_id)) async def turn_completed(self) -> None: """Record that one reply on a continuous stream finished normally.""" @@ -1025,24 +1033,34 @@ def test_continuous_pump_reports_multiple_replies_without_returning() -> None: async def scenario() -> None: transport = FakeTransport( + _event("input_audio_buffer.speech_started"), _event("response.created"), _event("response.audio_transcript.done", transcript="第一句"), _event("response.audio.delta", delta=base64.b64encode(b"pcm-1").decode()), _event("response.done"), + _event("input_audio_buffer.speech_started"), _event("response.created"), _event("response.audio_transcript.done", transcript="第二句"), _event("response.audio.delta", delta=base64.b64encode(b"pcm-2").decode()), _event("response.done"), _event("error", error={"message": "stream ended"}), ) + # Explicit clock, not the real one: the second speech_started's own barge-in + # check must land safely past the first reply's (near-instant) playable_until, + # or this flakes depending on how fast the test happens to run. + clock_reads = iter([0.0, 0.0, 0.0, 0.0, 1.0, 2.0, 2.0, 2.0, 2.0, 3.0]) observer = RecordingObserver() - await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) + await QwenAudioSession(transport, CONFIG, CONTINUOUS, clock=lambda: next(clock_reads)).pump( + observer + ) assert observer.calls == [ + ("user_started_speaking", None), ("spoke", "第一句"), ("audio", b"pcm-1"), ("turn_completed", None), + ("user_started_speaking", None), ("spoke", "第二句"), ("audio", b"pcm-2"), ("turn_completed", None), @@ -1061,6 +1079,7 @@ def test_continuous_pump_cancels_and_reports_an_interruption() -> None: async def scenario() -> None: transport = FakeTransport( + _event("input_audio_buffer.speech_started"), _event("response.created"), _event("response.audio_transcript.delta", delta="半句"), _event("input_audio_buffer.speech_started"), @@ -1077,6 +1096,7 @@ async def scenario() -> None: await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) assert observer.calls == [ + ("user_started_speaking", None), ("spoke", "半句"), ("user_started_speaking", None), ("interrupted", None), @@ -1107,8 +1127,12 @@ async def scenario() -> None: # takes to finish playing (~0.5s for the 24000 bytes below, 24kHz 16-bit # mono), then speech_started stamps itself and checks against that estimate # -- 0.1s later is still inside the playback window, so it is a real barge-in. - clock_reads = iter([0.0, 0.05, 0.05, 0.1, 0.1]) + # Leading pair (0.0, 0.0) is the reply's own opening speech_started -- legitimate, + # nothing to interrupt yet (_playable_until starts at 0.0) -- so this stays a + # harmless prefix ahead of the five reads the rest of the scenario already used. + clock_reads = iter([0.0, 0.0, 0.0, 0.05, 0.05, 0.1, 0.1]) transport = FakeTransport( + _event("input_audio_buffer.speech_started"), _event("response.created"), _event("response.audio.delta", delta=base64.b64encode(b"a" * 24000).decode()), _event("response.done"), @@ -1122,6 +1146,7 @@ async def scenario() -> None: ) assert observer.kinds() == [ + "user_started_speaking", "audio", "turn_completed", "user_started_speaking", @@ -1144,8 +1169,12 @@ async def scenario() -> None: # Five clock reads: created, first audio, done (playback estimate of ~0.5s for the # 24000 bytes below), then speech_started stamps itself and checks. 10s later is # far past the playback window, so nothing gets cancelled. - clock_reads = iter([0.0, 0.05, 0.05, 10.0, 10.0]) + # Leading pair (0.0, 0.0) is the reply's own opening speech_started -- legitimate, + # nothing to interrupt yet (_playable_until starts at 0.0) -- so this stays a + # harmless prefix ahead of the five reads the rest of the scenario already used. + clock_reads = iter([0.0, 0.0, 0.0, 0.05, 0.05, 10.0, 10.0]) transport = FakeTransport( + _event("input_audio_buffer.speech_started"), _event("response.created"), _event("response.audio.delta", delta=base64.b64encode(b"a" * 24000).decode()), _event("response.done"), @@ -1160,6 +1189,7 @@ async def scenario() -> None: ) assert observer.kinds() == [ + "user_started_speaking", "audio", "turn_completed", "user_started_speaking", @@ -1170,6 +1200,295 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_a_response_created_with_no_speech_or_followup_behind_it_is_canceled_on_sight() -> None: + """The vendor has been seen to start a response on its own -- no new speech_started, + no follow-up we asked for -- re-answering an already-settled query under a fresh + reply_id. Cancelled before any of its text can reach the client. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("input_audio_buffer.speech_started"), + _event("response.created"), + _event("response.audio_transcript.done", transcript="今天没有日程。"), + _event("response.done"), + _event("response.created"), # unsolicited: no speech_started, no follow-up + _event("response.audio_transcript.delta", delta="今天没有日程"), + _event("error", error={"message": "stream ended"}), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) + + assert observer.calls == [ + ("user_started_speaking", None), + ("spoke", "今天没有日程。"), + ("turn_completed", None), + ("failed", "stream ended"), + ] + assert transport.types() == ["response.cancel"] + + asyncio.run(scenario()) + + +def test_a_cancel_that_races_the_vendor_already_finishing_does_not_end_the_call() -> None: + """Found in production: cancelling an unsolicited response.created can lose the race + against the vendor finishing that same response a moment earlier on its own, which + comes back as an "error" event ("Conversation has no active response") rather than + silently succeeding. Treating every error as fatal hung up the whole call over what + was actually a no-op -- this must be swallowed and the call carries on. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("response.created"), # unsolicited: no speech_started yet at all + _event("error", error={"message": "Conversation has no active response."}), + _event("input_audio_buffer.speech_started"), + _event("response.created"), + _event("response.audio_transcript.done", transcript="真正的回复"), + _event("response.done"), + _event("error", error={"message": "stream ended"}), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) + + assert observer.calls == [ + ("user_started_speaking", None), + ("spoke", "真正的回复"), + ("turn_completed", None), + ("failed", "stream ended"), + ] + assert transport.types() == ["response.cancel"] + + asyncio.run(scenario()) + + +def test_an_unrelated_error_after_a_cancel_still_ends_the_call() -> None: + """The benign-race allowance is narrow: an error that is not the vendor saying + there was nothing to cancel must still be treated as fatal, even right after a + cancel_response() call. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("input_audio_buffer.speech_started"), + _event("response.created"), + _event("response.audio.delta", delta=base64.b64encode(b"a" * 24000).decode()), + _event("input_audio_buffer.speech_started"), # barge-in: cancels the reply + _event("error", error={"message": "internal server error"}), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) + + assert observer.calls[-1] == ("failed", "internal server error") + assert transport.types() == ["response.cancel"] + + asyncio.run(scenario()) + + +def test_a_response_done_between_the_cancel_and_its_error_does_not_end_the_call() -> None: + """The benign cancel race, in the order it actually arrives from the vendor. + + The vendor having already finished that response on its own is the very thing that + makes our cancel a no-op, so its response.done lands *between* the response.cancel + and the error saying there was nothing to cancel. An allowance that only covers the + single next event is spent on that response.done, and the call is then hung up over + exactly the race the allowance was added for. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("response.created"), # unsolicited: no speech_started, no follow-up + _event("response.done"), # the vendor had already finished it on its own + _event("error", error={"message": "Conversation has no active response."}), + _event("input_audio_buffer.speech_started"), + _event("response.created"), + _event("response.audio_transcript.done", transcript="真正的回复"), + _event("response.done"), + _event("error", error={"message": "stream ended"}), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) + + assert observer.calls == [ + ("turn_completed", None), + ("user_started_speaking", None), + ("spoke", "真正的回复"), + ("turn_completed", None), + ("failed", "stream ended"), + ] + assert transport.types() == ["response.cancel"] + + asyncio.run(scenario()) + + +def test_a_reused_session_does_not_inherit_the_previous_stream_s_expectations() -> None: + """A held session serves the next call too, so its per-stream state starts over. + + _expecting_response left true by the call that just ended would wave through the + first spontaneous response of the next one -- the duplicate reply this guard exists + to catch, slipping past on exactly the turn nobody is watching for it. + """ + + async def scenario() -> None: + transport = FakeTransport( + # First call: a reply we were legitimately waiting on, cut short. + _event("input_audio_buffer.speech_started"), + _event("response.created"), + _event("error", error={"message": "stream ended"}), + # Second call on the same session: the vendor speaks up unprompted. + _event("response.created"), + _event("error", error={"message": "stream ended"}), + ) + session = QwenAudioSession(transport, CONFIG, CONTINUOUS) + + await session.pump(RecordingObserver()) + second = RecordingObserver() + await session.pump(second) + + assert second.calls == [("failed", "stream ended")] + assert transport.types() == ["response.cancel"] + + asyncio.run(scenario()) + + +def test_a_reused_session_does_not_inherit_the_previous_reply_s_playback_estimate() -> None: + """The same reset, for the audio a previous call left counted as still playing. + + Within one call, `suppressed` shadows a stale playback estimate; across calls + nothing does, and the next call's opening speech_started gets reported as a barge-in + on a reply from the call before it -- a voice.tts.canceled the moment a fresh call + starts, over audio nobody is hearing. + """ + + async def scenario() -> None: + # First call: speech_started stamps and compares, response.created, first audio, + # then response.done estimates 0.5s of playback for the 24000 bytes below. The + # second call's speech_started stamps and compares 0.1s later -- inside that + # estimate, which is what makes the leak visible. + clock_reads = iter([0.0, 0.0, 0.0, 0.0, 0.0, 0.1, 0.1]) + transport = FakeTransport( + _event("input_audio_buffer.speech_started"), + _event("response.created"), + _event("response.audio.delta", delta=base64.b64encode(b"a" * 24000).decode()), + _event("response.done"), + _event("error", error={"message": "stream ended"}), + # Second call on the same session. + _event("input_audio_buffer.speech_started"), + _event("error", error={"message": "stream ended"}), + ) + session = QwenAudioSession(transport, CONFIG, CONTINUOUS, clock=lambda: next(clock_reads)) + + await session.pump(RecordingObserver()) + second = RecordingObserver() + await session.pump(second) + + assert second.kinds() == ["user_started_speaking", "failed"] + + asyncio.run(scenario()) + + +def test_a_reply_and_its_late_transcript_report_the_same_utterance_id() -> None: + """Captured from a real call: the vendor stamps the user's audio with an item id and + puts it on the transcript, which routinely lands after the reply it belongs to has + already started streaming. Reporting that id on both sides is what lets the client + pair them without guessing from arrival order. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("input_audio_buffer.speech_started", item_id="item_user_1"), + _event("input_audio_buffer.committed", item_id="item_user_1"), + _event("response.created"), + _event("response.audio_transcript.delta", delta="明天"), + # The real ordering: the transcript arrives mid-reply, not before it. + _event( + "conversation.item.input_audio_transcription.completed", + item_id="item_user_1", + transcript="明天我要看电影。", + ), + _event("response.audio_transcript.done", transcript="明天看电影,具体几点去?"), + _event("response.done"), + _event("error", error={"message": "stream ended"}), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) + + assert observer.turn_ids == [ + ("spoke", "item_user_1"), + ("heard", "item_user_1"), + ("spoke", "item_user_1"), + ] + + asyncio.run(scenario()) + + +def test_a_late_transcript_reports_its_own_utterance_not_the_one_now_running() -> None: + """A transcript carries its own item_id, so one that arrives after the user has + already started the next utterance still names the turn it actually transcribes. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("input_audio_buffer.speech_started", item_id="item_user_1"), + _event("input_audio_buffer.committed", item_id="item_user_1"), + _event("response.created"), + _event("response.audio_transcript.done", transcript="第一条回复"), + _event("response.done"), + # The next utterance is already under way when turn 1's transcript lands. + _event("input_audio_buffer.speech_started", item_id="item_user_2"), + _event( + "conversation.item.input_audio_transcription.completed", + item_id="item_user_1", + transcript="第一句话", + ), + _event("error", error={"message": "stream ended"}), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) + + assert observer.turn_ids == [("spoke", "item_user_1"), ("heard", "item_user_1")] + assert observer.calls[-2] == ("heard", "第一句话") + + asyncio.run(scenario()) + + +def test_a_tool_call_names_the_utterance_it_is_answering() -> None: + """A tool call happens before this reply has said anything, so the id has to come + from the session rather than from whatever the previous reply left behind -- a + question raised by the tool would otherwise be filed under the previous turn. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("input_audio_buffer.speech_started", item_id="item_user_1"), + _event("input_audio_buffer.committed", item_id="item_user_1"), + _event("response.created"), + _event("response.audio_transcript.done", transcript="第一条回复"), + _event("response.done"), + _event("input_audio_buffer.speech_started", item_id="item_user_2"), + _event("input_audio_buffer.committed", item_id="item_user_2"), + _event("response.created"), + _event( + "response.function_call_arguments.done", + call_id="call_1", + name="schedule_create", + arguments="{}", + ), + _event("error", error={"message": "stream ended"}), + ) + observer = RecordingObserver() + + await QwenAudioSession(transport, CONFIG, CONTINUOUS).pump(observer) + + assert ("tool", "item_user_2") in observer.turn_ids + + def test_continuous_pump_reports_tool_calls_and_a_bad_one_ends_the_stream() -> None: """Continuous mode reports tool calls the same way push-to-talk does.""" @@ -1484,16 +1803,18 @@ def test_a_second_audio_delta_does_not_re_stamp_first_audio_time() -> None: async def scenario() -> None: transport = FakeTransport( + _event("input_audio_buffer.speech_started"), _event("response.created"), _event("response.audio.delta", delta=base64.b64encode(b"pcm-1").decode()), _event("response.audio.delta", delta=base64.b64encode(b"pcm-2").decode()), _usage_event(), _event("error", error={"message": "stream ended"}), ) - # created, first delta stamps _first_audio_at, second delta's guard reads - # nothing further (short-circuited), response.done's playable_until estimate, - # then its own latency report. - clock_reads = iter([0.0, 0.2, 0.5, 0.5]) + # Leading pair (0.0, 0.0) is the reply's own opening speech_started (legitimate, + # nothing to interrupt yet). Then: created, first delta stamps _first_audio_at, + # second delta's guard reads nothing further (short-circuited), response.done's + # own latency report, then its playable_until estimate. + clock_reads = iter([0.0, 0.0, 0.0, 0.2, 0.5, 0.5]) observer = RecordingObserver() await QwenAudioSession(transport, CONFIG, CONTINUOUS, clock=lambda: next(clock_reads)).pump( @@ -1502,8 +1823,9 @@ async def scenario() -> None: usage = next(call[1] for call in observer.calls if call[0] == "usage_reported") assert usage["latency_first_audio_ms"] == 200.0 - assert observer.calls[0] == ("audio", b"pcm-1") - assert observer.calls[1] == ("audio", b"pcm-2") + assert observer.calls[0] == ("user_started_speaking", None) + assert observer.calls[1] == ("audio", b"pcm-1") + assert observer.calls[2] == ("audio", b"pcm-2") asyncio.run(scenario()) diff --git a/backend/tests/intelligence/realtime/test_realtime_agent.py b/backend/tests/intelligence/realtime/test_realtime_agent.py index 9b0daa34..d93a5268 100644 --- a/backend/tests/intelligence/realtime/test_realtime_agent.py +++ b/backend/tests/intelligence/realtime/test_realtime_agent.py @@ -784,3 +784,53 @@ async def open( assert second_session.audio_sent == [b"b"] asyncio.run(scenario()) + + +def test_a_farewell_without_end_conversation_still_hangs_up() -> None: + """The model sometimes says goodbye without remembering to call end_conversation; + composed guards this with the same fallback (agent.py's _is_farewell/_claims_success + pattern) and this mirrors it -- otherwise the call is left open after its own farewell. + """ + + async def scenario() -> None: + session = ScriptedSession( + [ + ("heard", ("先这样,拜拜",)), + ("spoke", ("好的,",)), + ("spoke", ("好的,再见",)), + ] + ) + sink = RecordingSink() + + await RealtimeAgent(ScriptedFactory(session), sink).handle_audio( + _chunks(b"a" * 3200), _Stream() + ) + + assert "session_end" in sink.kinds() + + asyncio.run(scenario()) + + +def test_a_farewell_in_a_reply_that_also_called_a_tool_does_not_auto_hang_up() -> None: + """Scoped to a reply that called no tool at all, same as composed: a reply that also + did something ("帮你删掉了,再见") should not risk a false hit on those words alone. + """ + + async def scenario() -> None: + session = ScriptedSession( + [ + ("heard", ("帮我把这条日程删了",)), + ("tool_requested", ("call_1", "schedule_delete", {})), + ("spoke", ("好的,",)), + ("spoke", ("好的,删掉了,再见",)), + ] + ) + sink = RecordingSink() + + await RealtimeAgent(ScriptedFactory(session), sink).handle_audio( + _chunks(b"a" * 3200), _Stream() + ) + + assert "session_end" not in sink.kinds() + + asyncio.run(scenario()) diff --git a/backend/tests/intelligence/realtime/test_realtime_sessions.py b/backend/tests/intelligence/realtime/test_realtime_sessions.py index 6c29092a..a32aa204 100644 --- a/backend/tests/intelligence/realtime/test_realtime_sessions.py +++ b/backend/tests/intelligence/realtime/test_realtime_sessions.py @@ -4,10 +4,13 @@ import asyncio import json +import logging from collections.abc import AsyncIterator, Awaitable, Callable from dataclasses import dataclass, field from typing import Any +import pytest + from timeflow.intelligence.location import ClientLocation, Coordinate from timeflow.intelligence.ports import ( AudioReply, @@ -515,3 +518,64 @@ async def scenario() -> None: assert "session_end" not in sink.kinds() asyncio.run(scenario()) + + +def test_a_turn_logs_what_the_user_said_and_what_the_model_asked_a_tool_for( + caplog: pytest.LogCaptureFixture, +) -> None: + """Both lines have to render their content, which extra= would silently not do. + + This project's log format string never references extra fields, so anything passed + that way prints nothing at all -- a trap this module has already fallen into twice + (see failed() and usage_reported()). Without these two lines a log shows replies + arriving with no way to tell what was said or what got written, which is what turned + every past investigation here into guesswork. + """ + + async def scenario() -> None: + tools = StubToolBox( + ToolResult( + output=json.dumps({"status": "applied"}), + outcome={"operation": "create_schedule", "status": "applied", "schedule": {}}, + ) + ) + factory = CountingFactory( + [ + ("heard", ("明天我将会去看电影",)), + ("tool_requested", ("call_1", "schedule_create", {"title": "看电影"})), + ] + ) + + with caplog.at_level(logging.INFO, logger="timeflow.intelligence.realtime.agent"): + await RealtimeAgent( + factory, + RecordingSink(), + tools_factory=_stub_factory(tools), # type: ignore[arg-type] + ).handle_audio(_chunks(b"a" * 3200), _Stream()) + + logged = [record.getMessage() for record in caplog.records] + # 中文原样可读,不能被转义成 \uXXXX。 + assert any("明天我将会去看电影" in line for line in logged) + assert any("schedule_create" in line and "看电影" in line for line in logged) + + asyncio.run(scenario()) + + +def test_tool_arguments_that_cannot_be_serialized_do_not_crash_the_turn() -> None: + """The log line must never be the thing that fails a turn.""" + + async def scenario() -> None: + tools = StubToolBox(ToolResult(output=json.dumps({"status": "applied"}))) + factory = CountingFactory( + [("tool_requested", ("call_1", "schedule_create", {"when": object()}))] + ) + + await RealtimeAgent( + factory, + RecordingSink(), + tools_factory=_stub_factory(tools), # type: ignore[arg-type] + ).handle_audio(_chunks(b"a" * 3200), _Stream()) + + assert [name for name, _ in tools.calls] == ["schedule_create"] + + asyncio.run(scenario()) diff --git a/backend/tests/intelligence/realtime/test_realtime_toolbox.py b/backend/tests/intelligence/realtime/test_realtime_toolbox.py index 8d83815d..f4dd8a01 100644 --- a/backend/tests/intelligence/realtime/test_realtime_toolbox.py +++ b/backend/tests/intelligence/realtime/test_realtime_toolbox.py @@ -159,6 +159,13 @@ def refusing_toolbox() -> ToolBox: ) +def _authorize_delete(box: ToolBox) -> ToolBox: + """Grant the delete-authorization gate directly, for tests exercising delete + mechanics rather than the gate itself (see test_realtime_agent.py for that).""" + box._delete_authorization = "confirmation" + return box + + def run(name: str, arguments: dict[str, Any], box: ToolBox | None = None) -> Any: return asyncio.run((box or refusing_toolbox()).run(name, arguments)) @@ -250,7 +257,8 @@ def test_a_refused_write_tells_the_model_and_not_the_client(name: str) -> None: "schedule_kind": "once", }, }[name] - result = run(name, arguments) + box = _authorize_delete(refusing_toolbox()) if name == "schedule_delete" else None + result = run(name, arguments, box) payload = json.loads(result.output) assert payload["status"] == "failed" assert payload["error"]["code"] == "revision_conflict" @@ -276,42 +284,56 @@ def test_a_question_reaches_the_client_and_not_the_calendar() -> None: assert result.outcome is None -def test_an_ambiguous_target_carries_the_candidates_it_found() -> None: +def test_an_ambiguous_target_only_needs_the_question_it_speaks_aloud() -> None: + """Asking which of several schedules costs one spoken sentence, nothing more. + + The model used to have to re-emit every matched schedule into a candidates array -- + measured at 477 output tokens and 7.4s of generation for two schedules, then another + 4s because the first attempt came back as a JSON string and was refused. Nothing + displayed it: the clients read only speech_text. So the choices are named in the + sentence the user actually hears, the same way ambiguous locations already work. + """ result = run( "request_user_input", { "question_kind": "ambiguous_target", - "speech_text": "是哪一个会?", - "candidates": [{"schedule_id": "sch_1"}, "not an object", {"schedule_id": "sch_2"}], + "speech_text": "找到两个开会,第一个是今天下午三点,第二个是每周一早上十点,删哪个?", }, ) - assert result.question is not None - # Anything that is not an object is dropped rather than passed to the client. - assert result.question["candidates"] == ({"schedule_id": "sch_1"}, {"schedule_id": "sch_2"}) - -def test_an_ambiguous_target_with_nothing_to_choose_between_is_refused() -> None: - result = run( - "request_user_input", - {"question_kind": "ambiguous_target", "speech_text": "是哪一个会?"}, - ) - assert json.loads(result.output)["status"] == "failed" - assert result.question is None + assert json.loads(result.output) == {"asked": True} + assert result.question is not None + assert result.question["candidates"] == () -def test_candidates_that_are_not_a_list_are_ignored() -> None: +def test_candidates_the_model_volunteers_anyway_are_not_passed_on() -> None: + """The parameter is gone from the schema; anything sent under that name is ignored + rather than quietly reaching the client as a shape nothing agreed on. + """ result = run( "request_user_input", { - "question_kind": "missing_field", - "speech_text": "哪天?", - "candidates": "sch_1", + "question_kind": "ambiguous_target", + "speech_text": "是哪一个会?", + "candidates": [{"schedule_id": "sch_1"}], }, ) + assert result.question is not None assert result.question["candidates"] == () +def test_request_user_input_no_longer_advertises_a_candidates_parameter() -> None: + """Left in the schema, the model keeps filling it -- that is where the seconds went.""" + (ask,) = [ + tool + for tool in ToolBox("acc_test", RecordingService()).tools() + if tool["function"]["name"] == "request_user_input" + ] + + assert "candidates" not in ask["function"]["parameters"]["properties"] + + @pytest.mark.parametrize( "arguments", [ @@ -329,6 +351,72 @@ def test_a_question_the_client_could_not_show_is_refused(arguments: dict[str, An assert result.outcome is None +_DELETE_ARGUMENTS = {"schedule_id": "sch_1", "expected_revision": 1, "schedule_kind": "once"} + + +def test_a_delete_straight_off_a_command_is_refused_without_asking_first() -> None: + # The model does sometimes skip the confirmation it was told to ask for in the + # prompt; this is the code-level backstop, mirroring composed's _delete_authorized. + box = ToolBox("acc_test", RecordingService()) + result = run("schedule_delete", _DELETE_ARGUMENTS, box) + payload = json.loads(result.output) + assert payload["status"] == "failed" + assert "request_user_input" in payload["error"]["message"] + assert result.outcome is None + + +@pytest.mark.parametrize("kind", ["confirmation", "recurrence_scope"]) +def test_a_delete_right_after_asking_confirmation_or_scope_is_allowed(kind: str) -> None: + box = ToolBox("acc_test", RecordingService()) + asked = run( + "request_user_input", + {"question_kind": kind, "speech_text": "确定要删除这条日程吗?"}, + box, + ) + assert asked.question is not None + + result = run("schedule_delete", _DELETE_ARGUMENTS, box) + assert json.loads(result.output)["status"] == "applied" + assert result.outcome is not None + + +def test_asking_something_else_does_not_carry_over_a_stale_delete_authorization() -> None: + box = ToolBox("acc_test", RecordingService()) + run( + "request_user_input", + {"question_kind": "confirmation", "speech_text": "确定要删除这条日程吗?"}, + box, + ) + # The model moved on to a different question instead of acting on the answer -- + # that earlier confirmation no longer authorizes a delete. + run( + "request_user_input", + {"question_kind": "missing_field", "speech_text": "还差开始时间,几点?"}, + box, + ) + + result = run("schedule_delete", _DELETE_ARGUMENTS, box) + assert json.loads(result.output)["status"] == "failed" + assert result.outcome is None + + +def test_one_confirmation_authorizes_a_whole_batch_of_deletes() -> None: + # A delete does not consume the authorization: "delete all three" only gets one + # confirmation from the model, then schedule_delete once per schedule -- if the + # first one cleared it, every delete after the first in a batch would be wrongly + # refused as unconfirmed. + box = ToolBox("acc_test", RecordingService()) + run( + "request_user_input", + {"question_kind": "confirmation", "speech_text": "确定要把这三条日程都删除吗?"}, + box, + ) + + for _ in range(3): + result = run("schedule_delete", _DELETE_ARGUMENTS, box) + assert json.loads(result.output)["status"] == "applied" + + @pytest.mark.parametrize( ("arguments", "expected"), [ @@ -343,7 +431,7 @@ def test_a_delete_reaches_the_call_that_matches_the_kind( run( "schedule_delete", {"schedule_id": "sch_1", "expected_revision": 1, **arguments}, - ToolBox("acc_test", service), + _authorize_delete(ToolBox("acc_test", service)), ) assert service.calls == [expected] @@ -375,7 +463,7 @@ def test_a_this_occurrence_delete_reports_the_override_it_produced() -> None: "schedule_kind": "recurring", "scope": "this_occurrence", }, - ToolBox("acc_test", OverrideProducingService()), + _authorize_delete(ToolBox("acc_test", OverrideProducingService())), ) assert result.outcome is not None assert result.outcome["schedule"] is None @@ -409,7 +497,7 @@ def test_a_this_occurrence_delete_with_existing_replacement_reports_both_schedul "schedule_kind": "recurring", "scope": "this_occurrence", }, - ToolBox("acc_test", MultiScheduleProducingService()), + _authorize_delete(ToolBox("acc_test", MultiScheduleProducingService())), ) assert result.outcome is not None # schedule(单数)保持只给第一条,向后兼容 @@ -429,7 +517,7 @@ def test_a_delete_with_nothing_left_to_report_still_says_it_applied() -> None: "schedule_kind": "recurring", "scope": "entire_series", }, - ToolBox("acc_test", RecordingService()), + _authorize_delete(ToolBox("acc_test", RecordingService())), ) assert json.loads(result.output) == {"status": "applied", "schedule": None} assert result.outcome is not None diff --git a/backend/tests/test_agent_delivery.py b/backend/tests/test_agent_delivery.py index 06c2fcd0..54c30b07 100644 --- a/backend/tests/test_agent_delivery.py +++ b/backend/tests/test_agent_delivery.py @@ -441,6 +441,9 @@ async def scenario() -> None: "reply_id": "reply_001", "speech_text": "好,明天下午三点", "done": False, + # Absent unless the producer can name the utterance this answers; the + # composed backend cannot, so the field ships as null there. + "turn_id": None, } asyncio.run(scenario()) @@ -504,6 +507,7 @@ async def scenario() -> None: "speech_text": "你说的是早会还是周会?", "required_response": "schedule_id", "candidates": [{"id": "schedule_1"}, {"id": "schedule_2"}], + "turn_id": None, } asyncio.run(scenario()) diff --git a/backend/tests/test_ws_voice_stream.py b/backend/tests/test_ws_voice_stream.py index d0890290..2394a681 100644 --- a/backend/tests/test_ws_voice_stream.py +++ b/backend/tests/test_ws_voice_stream.py @@ -83,6 +83,18 @@ async def consume(self, chunks: AsyncIterator[bytes], stream: StreamContext) -> raise RuntimeError("the sink is unavailable") +class AbandoningSink: + """A sink that walks away from the stream without raising. + + The realtime agent does exactly this when it cannot open a vendor session: it logs + and returns, so consume() succeeds while nothing ever reads the queue. + """ + + async def consume(self, chunks: AsyncIterator[bytes], stream: StreamContext) -> None: + """Return without reading anything.""" + return + + def _build_app( sink: AudioSink, *, @@ -466,6 +478,31 @@ def test_a_failing_sink_does_not_wedge_the_session() -> None: assert replies[0]["ok"] is False +def test_a_sink_that_gives_up_without_failing_does_not_wedge_the_session() -> None: + """A sink that returns early retires its stream too, not only one that raises. + + Found by walking the realtime path: it logs and returns when the vendor session + cannot be opened, so consume() succeeds and the exception handler never runs -- + yet nothing drains the queue either. The receive loop then parks on a full queue + and the whole connection stops answering, including the voice.stream.end that + would have been the way out of it. + """ + client = TestClient(_build_app(AbandoningSink(), queue_max_chunks=4)) + + with client.websocket_connect("/ws?device_id=device_001") as websocket: + websocket.send_json(VALID_HELLO) + websocket.receive_json() + websocket.send_json(START) + websocket.receive_json() + for _ in range(12): + websocket.send_bytes(b"\x00" * 320) + websocket.send_json({"type": "unknown.probe"}) + + reply = websocket.receive_json() + + assert reply["ok"] is False + + def test_disconnect_cancels_an_unfinished_stream() -> None: """Dropping the connection mid-stream leaves no work running.""" sink = CapturingSink() diff --git a/frontend/src/contracts/conversation.ts b/frontend/src/contracts/conversation.ts index bbc31b4c..72054784 100644 --- a/frontend/src/contracts/conversation.ts +++ b/frontend/src/contracts/conversation.ts @@ -59,7 +59,14 @@ export interface VoiceAsrCompletedMessage { type: 'voice.asr.completed'; request_id?: string; conversation_id: string; - payload: { transcript: string; language: string; duration_ms: number }; + /** turn_id:这句话对应的那段用户语音的 id,用来跟回答它的 reply 精确配对。 + * 只有 realtime 后端拿得到(来自 vendor 的 item_id),composed 后端为 null。 */ + payload: { + transcript: string; + language: string; + duration_ms: number; + turn_id?: string | null; + }; } /** 日程快照的形状由后端 `ScheduleSnapshot` 决定;前端只透传给本地落库,不在此处强约束字段。 */ @@ -100,6 +107,7 @@ export interface VoiceDialogueQuestionMessage { speech_text: string; required_response?: string; candidates: Record[]; + turn_id?: string | null; }; } @@ -107,7 +115,7 @@ export interface VoiceDialogueReplyMessage { type: 'voice.dialogue.reply'; request_id?: string; conversation_id: string; - payload: { reply_id: string; speech_text: string; done: boolean }; + payload: { reply_id: string; speech_text: string; done: boolean; turn_id?: string | null }; } export interface VoiceTtsStartMessage { diff --git a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts index 83f2fd32..7af4d890 100644 --- a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts +++ b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts @@ -26,6 +26,14 @@ const LOCATION_TIMEOUT_MS = 2000; // 单独的"等待用户输入 10~30 秒"档位按设计简化,不再单独实现——这一档就是 // 最终会触发动作的那一档。 const SESSION_IDLE_TIMEOUT_MS = 180_000; +// 高频内容更新合并成一次通知的窗口。vendor 的回复文字是逐词下发的(实测一句话约 +// 20 条、间隔 12~16ms),每条都通知一次就是每条都整屏重渲染;而播放链路每 ~50ms +// 就要把下一块 PCM 喂进原生播放器,被渲染挤掉就是一次可听见的空隙。50ms 一档, +// 文字看起来照样是流式的,重渲染次数降到四分之一以下。 +const NOTIFY_COALESCE_MS = 50; +// 等一个原生调用的上限。见 settleWithin 的注释:卡住不返回的原生调用会把挂断和 +// 播放两条路一起冻住,而原生停止本该是毫秒级的,1 秒已经宽裕得离谱。 +const NATIVE_TEARDOWN_TIMEOUT_MS = 1000; /** 允许暂停/由暂停恢复的阶段——除此之外调用 togglePause() 什么都不做。 */ const PAUSABLE_PHASES: ReadonlySet = new Set([ @@ -59,8 +67,12 @@ export class AssistantContinuousConversationService implements AssistantApplicat private streamId: string | null = null; /** 非 null 表示当前正处于 voice.tts.start 和 voice.tts.end/canceled 之间。 */ private currentAudioId: string | null = null; - /** 最近被打断的音频 id;服务端会在 canceled 后补发同 id 的 tts.end。 */ - private canceledAudioId: string | null = null; + /** 被打断、还没等到它自己 tts.end 的音频 id 集合;服务端会在 canceled 后补发 + * 同 id 的 tts.end。用 Set 而不是单个值:跟 AssistantConversationService(PTT) + * 里同名 bug 一样,连续两次打断前后脚发生、第一条还没等到它的 tts.end 就被 + * 第二条覆盖掉的话,第一条迟到的 tts.end 会落进正常收尾分支,多余地调一次 + * endStream() 并把第二条本该维持住的取消状态冲掉。 */ + private readonly canceledAudioIds = new Set(); private streamStartedWaiter: ((conversationId: string) => void) | null = null; /** 跟 streamStartedWaiter 配对;传输层报错/连接掉线时用它让等待方结束,不然会永远卡住。 */ private streamStartRejecter: ((error: Error) => void) | null = null; @@ -70,16 +82,17 @@ export class AssistantContinuousConversationService implements AssistantApplicat private soundLevel: number | null = null; /** 本次开麦以来的问答历史,追加不覆盖;startTurn() 清空,跟 replyText 各管各的 * ——replyText 是当前这一轮的气泡内容,turns 是完整历史。气泡列表由 turns 推导 - * (每轮一条用户句 + 一条助手句)。连续模式一次 voice.stream.start 不带 - * request_id,服务端回显的 request_id 恒为空,所以轮次只能按到达顺序排: - * 每个 voice.asr.completed 追加一轮,随后的 reply/question 写进当前这轮。 - * 后端(realtime 在 speech_started 打断、composed 在 VAD 打断)都保证上一轮 - * 回复在下一轮转写前落地,顺序天然正确;turns 推导的意义在于让"追问"和它 - * 的流式回复落在同一轮同一个气泡里,而不是用 request_id 做跨轮关联。 */ + * (每轮一条用户句 + 一条助手句)。 + * + * 轮次归属优先看消息带的 turn_id:realtime 后端把 vendor 给这段用户语音的 + * item_id 同时贴在转写和回答它的 reply/question 上,两半谁先到都能按同一个 id + * 找回同一轮。这很重要——实测 vendor 的转写经常晚于它对应的回复到达,甚至晚于 + * response.done。 + * + * 没有 turn_id 时(composed 后端拿不到 vendor 的 id)退回按到达顺序归属:每个 + * voice.asr.completed 追加一轮,随后的 reply/question 写进当前这轮。这套只在 + * "一轮的两半挨着到"时成立,两轮的回复都赶在任何一条转写之前完成就会配错人。 */ private turns: ConversationTurnRecord[] = []; - /** 回复/追问先于任何 transcript 到达(正常流程不会发生)时挂在这里的零散助手句, - * getMessages() 里接在 turns 推导结果后面,避免乱序把内容丢掉。 */ - private orphanMessages: VoiceChatMessage[] = []; /** 当前是否正在流式输出一条回复(voice.dialogue.reply done=false 起、 * done=true 止)。连续模式同一时刻最多只有一条回复在流,所以一个布尔就够; * 展示层据此给最后一条助手气泡加波形。composed 代理打断回复时只发 @@ -91,9 +104,18 @@ export class AssistantContinuousConversationService implements AssistantApplicat * 必须开一个占位轮挂上去,否则会覆盖上一条已完成的回复(改前实测:上一条 * 回复被后一条覆盖成一样的内容)。 */ private lastTurnReplyDone = false; + /** 最后一条轮次正在写入的 reply_id(没有则为 null,voice.dialogue.question + * 不带 reply_id)。lastTurnReplyDone 为 true 时用来加一道防御:如果新消息 + * 跟这个 reply_id 相同,说明是同一条已经收尾的回复的迟到/重复投递,不是 + * 真的新一轮——不能开占位轮,否则会造成重复气泡(跟 vendor 凭空多起 + * response 那次实测的现象一样,这里是给"到达顺序"这个唯一依据加一层身份 + * 校验)。 */ + private lastReplyId: string | null = null; /** endTurn() 主动关闭连接期间为 true,让 handleClose 认出这是预期内的挂断。 */ private endingCall = false; private idleTimer: ReturnType | null = null; + /** notifyThrottled() 排着的那次通知;非 null 表示这一档里已经有一次待发。 */ + private notifyTimer: ReturnType | null = null; /** true 期间麦克风回调采到的帧不再往 WS 发;连接和录音本身不受影响。 */ private muted = false; /** endTurn() 执行期间为 true:期间到达的服务端消息一律丢弃,防止挂断过程中 @@ -172,11 +194,12 @@ export class AssistantContinuousConversationService implements AssistantApplicat if (turn.transcript.length > 0) { messages.push({ id: `${turn.id}:user`, role: 'user', text: turn.transcript }); } - if (turn.replyText !== null && turn.replyText.length > 0) { + // trim 后判空:流式回复的第一条分片可能只有空白(vendor 先起了 response、 + // 字还没吐出来),渲染出来就是一个空气泡。 + if (turn.replyText !== null && turn.replyText.trim().length > 0) { messages.push({ id: `${turn.id}:assistant`, role: 'assistant', text: turn.replyText }); } } - messages.push(...this.orphanMessages); if (this.replyStreaming) { for (let index = messages.length - 1; index >= 0; index -= 1) { if (messages[index].role === 'assistant') { @@ -202,9 +225,10 @@ export class AssistantContinuousConversationService implements AssistantApplicat this.replyText = null; this.soundLevel = null; this.turns = []; - this.orphanMessages = []; this.replyStreaming = false; this.lastTurnReplyDone = false; + this.lastReplyId = null; + this.canceledAudioIds.clear(); // 上一通电话可能是在暂停期间被空闲超时兜底挂断的,muted 只在用户手动 // togglePause() 里恢复;不在这里清一次,新开的电话会继承上一通的静音, // UI 显示 listening 但麦克风帧全被吞掉。 @@ -250,7 +274,12 @@ export class AssistantContinuousConversationService implements AssistantApplicat } connection.sendAudioFrame(chunk); this.soundLevel = soundLevel; - this.notifyListeners(); + // 正在放 TTS 时不为音量刷界面:这时候光球显示的是麦克风能量,本来就没有 + // 意义,而每刷一次都要跟播放链路抢同一条 JS 线程。音量照常记着,下一次 + // 真正的状态变化会把它一起带出去。 + if (this.state.phase !== 'speaking') { + this.notifyThrottled(); + } }); } catch (error) { // 服务端这时候已经确认开流了(stream_id 拿到手了):跟 endTurn() 一样的 @@ -277,29 +306,46 @@ export class AssistantContinuousConversationService implements AssistantApplicat /** 关闭连续会话:关麦、结束流、断开连接。 */ async endTurn(stopPlayback: boolean = true): Promise { + await this.hangUp({ phase: 'idle' }, stopPlayback); + } + + /** + * 服务端报错就等于这条流已经不能用了:走跟挂断完全一样的收尾,只是把原因 + * 留在界面上,而不是悄悄归 idle 让通话凭空消失。 + * + * 光设一个 error 状态是不够的——麦克风还开着、音频帧还在往一条服务端已经回收 + * 掉的流上发,服务端就一帧回一条错误(100ms 一条),用户看到"出错了"却发现通话 + * 还在继续,只能自己想起来去点"结束对话"。而且不发 voice.stream.end 的话,服务端 + * 那条流还挂着,下一次 voice.stream.start 会被当成"已有活跃流"直接拒绝。 + */ + private async abortCall(message: string): Promise { + await this.hangUp({ message, phase: 'error' }, true); + } + + private async hangUp(finalState: ConversationTurnState, stopPlayback: boolean): Promise { // 挂断可能同时从两个地方触发(用户点"结束对话" vs 服务端 voice.session.end), // 也可能在它还没跑完时被同一个来源再触发一次——不加门槛会重复关连接、 - // 重复发 voice.stream.end。 + // 重复发 voice.stream.end。收尾途中到达的服务端报错同样被这道门槛挡掉: + // 已经在挂了,不该再被掰成 error。 if (this.endTurnInFlight) { return; } this.endTurnInFlight = true; this.clearIdleTimer(); + this.clearNotifyTimer(); this.endingCall = true; try { - await this.deps.capture.stop(); - } catch { - // 原生停止失败也不能让挂断卡在半路——跟 dispose() 的兜底思路一致, - // 后面几步(发 stream.end、关连接、状态归位)必须照常走完。 - } - this.soundLevel = null; - // 手动挂断(用户点"结束对话")要立刻停掉正在播的 TTS;语音挂断(服务端 - // voice.session.end,AI 已道别)则让道别音频播完,不截断。 - if (stopPlayback) { - await this.stopPlaybackImmediately(); - } - const connection = this.connection; - try { + // 两个原生调用都限时。原来它们各自裸 await 在清 endTurnInFlight 的 finally + // 前面:抛错有 catch 兜着,卡住不返回没有——而卡住就再也走不到后面的收尾, + // endTurnInFlight 永远是 true,结束对话从此按不动(实测症状)。 + await settleWithin(this.deps.capture.stop(), NATIVE_TEARDOWN_TIMEOUT_MS); + this.soundLevel = null; + // 手动挂断(用户点"结束对话")要立刻停掉正在播的 TTS;语音挂断(服务端 + // voice.session.end,AI 已道别)则让道别音频播完,不截断。 + if (stopPlayback) { + await this.stopPlaybackImmediately(); + } + const connection = this.connection; if (connection !== null && this.streamId !== null) { connection.send({ payload: { stream_id: this.streamId }, @@ -309,12 +355,12 @@ export class AssistantContinuousConversationService implements AssistantApplicat this.unsubscribeConnection?.(); connection?.close(); } catch { - // 收尾步骤失败也不能拦住下面的状态归位。 + // 收尾任何一步失败都不能拦住下面的状态归位。 } finally { this.streamId = null; this.unsubscribeConnection = null; this.connection = null; - this.setState({ phase: 'idle' }); + this.setState(finalState); this.endTurnInFlight = false; } } @@ -358,6 +404,7 @@ export class AssistantContinuousConversationService implements AssistantApplicat this.disposed = true; this.pendingCategoryUpdates.clear(); this.clearIdleTimer(); + this.clearNotifyTimer(); this.unsubscribeAppState(); this.listeners.clear(); // 连续模式麦克风开着的时间远长于按住说话的一次按住,组件卸载时更可能还在 @@ -410,8 +457,8 @@ export class AssistantContinuousConversationService implements AssistantApplicat return; } if (isTransportError(message)) { - this.setState({ message: message.error.message, phase: 'error' }); this.rejectPendingStreamStart(new Error(message.error.message)); + void this.abortCall(message.error.message); return; } @@ -428,24 +475,7 @@ export class AssistantContinuousConversationService implements AssistantApplicat this.armIdleTimer(); const transcript = message.payload.transcript.trim(); if (transcript.length > 0) { - const last = this.turns[this.turns.length - 1]; - if (last !== undefined && last.transcript === '' && last.replyText !== null) { - // 上一条回复先于它的转写到达,开过占位轮:现在把转写补进同一轮。 - this.turns = this.turns.map((turn, i) => - i === this.turns.length - 1 ? { ...turn, transcript } : turn, - ); - } else { - this.turns = [ - ...this.turns, - { - id: message.request_id ?? `turn-${this.turns.length}`, - replyText: null, - transcript, - }, - ]; - // 新一轮等着它自己的回复。 - this.lastTurnReplyDone = false; - } + this.writeTurnTranscript(message.payload.turn_id ?? null, message.request_id, transcript); } this.notifyListeners(); return; @@ -472,11 +502,13 @@ export class AssistantContinuousConversationService implements AssistantApplicat // 追问本身就是这轮回复的文字:后续 voice.dialogue.reply 的流式文字还会 // 继续写进同一轮。这里不再单独生成一个 question_id 气泡——否则同一句 // 追问会以两个气泡重复出现(改前实测)。 - if (this.turns.length === 0) { - this.upsertOrphanQuestion(message.payload.question_id, message.payload.speech_text); - } else { - this.writeTurnReply(message.request_id, message.payload.speech_text, false); - } + this.writeTurnReply( + message.payload.turn_id ?? null, + message.request_id, + null, + message.payload.speech_text, + false, + ); this.setState({ conversationId: message.conversation_id, phase: 'asking', @@ -486,23 +518,23 @@ export class AssistantContinuousConversationService implements AssistantApplicat case 'voice.dialogue.reply': { this.replyText = message.payload.speech_text; this.replyStreaming = !message.payload.done; - if (this.turns.length === 0) { - // 回复先于任何 transcript 到达(正常流程不会发生):没有轮次可写, - // 挂到零散气泡列表兜底。 - this.upsertOrphanReply(message.payload.reply_id, message.payload.speech_text); + this.writeTurnReply( + message.payload.turn_id ?? null, + message.request_id, + message.payload.reply_id, + message.payload.speech_text, + message.payload.done, + ); + // 收尾那条不合并,让最终文案立刻落地;中间的分片合并掉。 + if (message.payload.done) { + this.notifyListeners(); } else { - this.writeTurnReply( - message.request_id, - message.payload.speech_text, - message.payload.done, - ); + this.notifyThrottled(); } - this.notifyListeners(); return; } case 'voice.tts.start': this.playbackGeneration += 1; - this.canceledAudioId = null; this.currentAudioId = message.audio_id; this.setState({ conversationId: message.conversation_id, phase: 'speaking' }); this.chainPlayback(() => @@ -514,22 +546,23 @@ export class AssistantContinuousConversationService implements AssistantApplicat return; case 'voice.tts.end': // canceled 后服务端仍会补发同 id 的 tts.end;它不能收尾新流,也不能把 - // interrupted 状态提前改回 listening。 - if ( - (this.canceledAudioId !== null && message.audio_id === this.canceledAudioId) || - (this.currentAudioId !== null && message.audio_id !== this.currentAudioId) - ) { + // interrupted 状态提前改回 listening。用 Set.delete() 而不是比较单个 + // 变量:见 canceledAudioIds 的字段注释,两次打断前后脚发生时,单个变量 + // 会让第一条的 id 被第二条覆盖掉。 + if (this.canceledAudioIds.delete(message.audio_id)) { + return; + } + if (this.currentAudioId !== null && message.audio_id !== this.currentAudioId) { return; } this.currentAudioId = null; - this.canceledAudioId = null; this.chainPlayback(() => this.deps.playback.endStream()); this.setState({ conversationId: message.conversation_id, phase: 'listening' }); // 播报完成后给一个全新的窗口,对应"播报完成后进入短暂等待"——不用单独 // 再搞一个计时器,这次重置就当作那个等待。 this.armIdleTimer(); return; - case 'voice.tts.canceled': + case 'voice.tts.canceled': { // 用户开口打断了正在播的回复:stop 必须绕过 playbackChain 立即执行,否则 // 已排队的 PCM 会先继续喂给原生播放器;旧队列随后由代次检查丢弃。 if ( @@ -538,7 +571,10 @@ export class AssistantContinuousConversationService implements AssistantApplicat ) { return; } - this.canceledAudioId = message.audio_id || this.currentAudioId; + const canceledAudioId = message.audio_id || this.currentAudioId; + if (canceledAudioId !== null) { + this.canceledAudioIds.add(canceledAudioId); + } this.currentAudioId = null; // composed 代理打断回复时只发 voice.tts.canceled、不会再补 done=true, // 所以这里要顺手清掉流式标记,否则被打断那条气泡的波形会一直跳。 @@ -551,6 +587,7 @@ export class AssistantContinuousConversationService implements AssistantApplicat void this.stopPlaybackImmediately(); this.setState({ conversationId: message.conversation_id, phase: 'interrupted' }); return; + } case 'voice.session.end': // 模型识别到用户想结束对话("结束对话"「先这样」等)。语音挂断时 AI 已先 // 道别,这里不截断道别音频、让它播完;连接照常关闭。 @@ -667,7 +704,17 @@ export class AssistantContinuousConversationService implements AssistantApplicat /** 立即清空原生播放器,并把后续新操作排在 stop 完成之后。 */ private async stopPlaybackImmediately(): Promise { this.playbackGeneration += 1; - const stop = this.deps.playback.stop().catch(() => {}); + // 排在 playbackChain 末尾而不是直接替换:在途的 pushChunk 正在逐块 playAudio, + // 让 stopAudio 跟原生侧的写入并发,正是按住说话那边同名方法的注释点名过的危险 + // 动作。排队的成本现在很低——pushChunk 每块之前会确认这条流还在不在,代次一变 + // 就当场停手,所以最多等一块。 + // 再限一次时:原生 stop 卡住不返回的话,playbackChain 会停在一个永远不会 settle + // 的 promise 上,之后每一块音频都排在它后面,这通电话再也不出声。宁可不等, + // 也不能让这条链死掉。 + const stop = settleWithin( + this.playbackChain.then(() => this.deps.playback.stop()), + NATIVE_TEARDOWN_TIMEOUT_MS, + ); this.playbackChain = stop; await stop; } @@ -708,17 +755,72 @@ export class AssistantContinuousConversationService implements AssistantApplicat return this.connection; } + /** 把一句转写写进它所属的轮次。 + * 后端给了 turn_id 就按 id 认(realtime:vendor 给这段用户语音的 item_id,回答它 + * 的 reply 带的是同一个),一句转写无论多晚到达都能找回自己那一轮。 + * 没有 turn_id(composed 后端)时退回原来那套按到达顺序的归属。 */ + private writeTurnTranscript( + turnId: string | null, + requestId: string | undefined, + transcript: string, + ): void { + if (turnId !== null) { + this.turns = upsertTurn(this.turns, turnId, (turn) => ({ ...turn, transcript }), { + id: turnId, + replyText: null, + transcript, + }); + return; + } + const last = this.turns[this.turns.length - 1]; + if (last !== undefined && last.transcript === '' && last.replyText !== null) { + // 上一条回复先于它的转写到达,开过占位轮:现在把转写补进同一轮。 + this.turns = this.turns.map((turn, i) => + i === this.turns.length - 1 ? { ...turn, transcript } : turn, + ); + return; + } + this.turns = [ + ...this.turns, + { id: requestId ?? `turn-${this.turns.length}`, replyText: null, transcript }, + ]; + // 新一轮等着它自己的回复。 + this.lastTurnReplyDone = false; + } + /** 把一条助手消息(回复/追问)写进它所属的轮次。连续模式没有 per-turn * request_id,回复只能按到达顺序归属到"当前轮"(最后一条): * - 正常:最后一条轮次的回复还没完成,直接写进去(done 驱动波形)。 - * - 上一轮已回复完成、又来一条新回复:说明它属于"还没到的新一轮"(转写 - * 晚到或被 VAD 跳过)。必须开一个占位轮挂上去,等它的 asr 到了再补 - * 转写——否则会覆盖上一条已完成的回复(改前实测:上一条回复被后一条 - * 覆盖成一样的内容)。 + * - 一条轮次都还没有,或者上一轮已回复完成又来了新回复:这条属于"转写还没到 + * 的一轮",开一个占位轮挂上去,等它的 asr 到了再由 voice.asr.completed 补 + * 进同一轮。不开占位轮的话,前者会让这句话没有落脚点,后者会覆盖上一条已 + * 完成的回复(改前实测:上一条回复被后一条覆盖成一样的内容)。除非新消息的 + * reply_id 跟刚收尾那条完全一样:那说明这不是新一轮,是同一条回复的迟到/ + * 重复投递,直接丢弃——到达顺序是唯一依据时挡不住这种情况,reply_id 相等 + * 能挡住。 + * "一条轮次都还没有"不是异常分支,是每通电话第一轮的常态:后端日志实测 vendor + * 的 transcription.completed 就是在回复开始之后才发的,有时甚至晚于 + * response.done。这里原来是把这种回复挂进一个单独的零散列表,那个列表永远渲染 + * 在最后、也永远不会被随后到达的转写认领,于是同一句话既出现在它该在的轮次里、 + * 又有一条挂在对话最底下整通电话不消失(实测复现)。 * request_id 分支只为向前兼容:万一后端将来给连续模式的每条消息带上独立 - * request_id,就能按它认轮次。 */ - private writeTurnReply(requestId: string | undefined, replyText: string, done: boolean): void { - if (this.turns.length === 0) { + * request_id,就能按它认轮次。voice.dialogue.question 没有 reply_id,传 null + * 即可,占位轮逻辑照常。 */ + private writeTurnReply( + turnId: string | null, + requestId: string | undefined, + replyId: string | null, + replyText: string, + done: boolean, + ): void { + if (turnId !== null) { + this.turns = upsertTurn(this.turns, turnId, (turn) => ({ ...turn, replyText }), { + id: turnId, + replyText, + transcript: '', + }); + this.lastTurnReplyDone = done; + this.lastReplyId = replyId; return; } if (requestId !== undefined) { @@ -726,52 +828,23 @@ export class AssistantContinuousConversationService implements AssistantApplicat if (found >= 0) { this.turns = this.turns.map((turn, i) => (i === found ? { ...turn, replyText } : turn)); this.lastTurnReplyDone = done; + this.lastReplyId = replyId; return; } } - if (this.lastTurnReplyDone) { + if (this.turns.length === 0 || this.lastTurnReplyDone) { + if (replyId !== null && replyId === this.lastReplyId) { + return; + } this.turns = [...this.turns, { id: `turn-${this.turns.length}`, replyText, transcript: '' }]; this.lastTurnReplyDone = done; + this.lastReplyId = replyId; return; } const index = this.turns.length - 1; this.turns = this.turns.map((turn, i) => (i === index ? { ...turn, replyText } : turn)); this.lastTurnReplyDone = done; - } - - /** 零散助手句(回复先于任何转写到达时的兜底):同 id 覆盖;波形由 - * getMessages() 里的 replyStreaming 统一决定,这里不再单独带 pending。 */ - private upsertOrphanReply(id: string, speechText: string): void { - const text = speechText.trim(); - if (text.length === 0) { - return; - } - const next: VoiceChatMessage = { id, role: 'assistant', text }; - const existing = this.orphanMessages.findIndex((message) => message.id === id); - if (existing >= 0) { - this.orphanMessages = this.orphanMessages.map((message, index) => - index === existing ? next : message, - ); - return; - } - this.orphanMessages = [...this.orphanMessages, next]; - } - - /** 追问先于任何转写到达时的兜底:有轮次可写时一律走 writeTurnReply, - * 这里只在 turns 为空时把内容挂到零散列表。 */ - private upsertOrphanQuestion(questionId: string, speechText: string): void { - if (this.turns.length > 0 || speechText.trim().length === 0) { - return; - } - const text = speechText.trim(); - const existing = this.orphanMessages.findIndex((message) => message.id === questionId); - if (existing >= 0) { - this.orphanMessages = this.orphanMessages.map((message, index) => - index === existing ? { ...message, text } : message, - ); - return; - } - this.orphanMessages = [...this.orphanMessages, { id: questionId, role: 'assistant', text }]; + this.lastReplyId = replyId; } private armIdleTimer(): void { @@ -791,9 +864,64 @@ export class AssistantContinuousConversationService implements AssistantApplicat this.notifyListeners(); } + /** 把一串高频内容更新合并成一次通知。状态本身是同步改好的——getMessages()、 + * getReplyText() 立刻就是新值——这里推迟的只是"通知界面重画",所以合并不会让 + * 任何人读到旧数据,只是少刷几帧,把 JS 线程让给播放链路。 */ + private notifyThrottled(): void { + if (this.notifyTimer !== null) { + return; + } + this.notifyTimer = setTimeout(() => { + this.notifyTimer = null; + this.notifyListeners(); + }, NOTIFY_COALESCE_MS); + } + private notifyListeners(): void { + // 立刻通知已经把最新状态送出去了,排着的那次就是多余的一帧。 + this.clearNotifyTimer(); for (const listener of this.listeners) { listener(this.state); } } + + private clearNotifyTimer(): void { + if (this.notifyTimer !== null) { + clearTimeout(this.notifyTimer); + this.notifyTimer = null; + } + } +} + +/** 按 id 找到那一轮就地更新,找不到就用 fallback 追加一条新的。 + * turn_id 到齐之前,一轮的两半(转写和回复)谁先到都可能——先到的那半建轮次, + * 后到的那半按同一个 id 找回来填进去,跟到达顺序无关。 */ +function upsertTurn( + turns: readonly ConversationTurnRecord[], + turnId: string, + update: (turn: ConversationTurnRecord) => ConversationTurnRecord, + fallback: ConversationTurnRecord, +): ConversationTurnRecord[] { + const found = turns.findIndex((turn) => turn.id === turnId); + if (found >= 0) { + return turns.map((turn, index) => (index === found ? update(turn) : turn)); + } + return [...turns, fallback]; +} + +/** 等一个原生调用,但最多等这么久,且从不抛出。 + * + * 原生桥调用失败会抛(catch 得住),卡住不返回不会。而在挂断路径上卡住意味着后面 + * 的收尾永远走不到、结束对话按钮从此失效;在播放路径上卡住意味着 playbackChain 停 + * 在一个不会 settle 的 promise 上、整通电话再也不出声。这两个症状实测是一起出现的。 + * 超时不代表原生那边真的结束了,只代表我们不再等它、把控制权还给用户。 */ +function settleWithin(work: Promise, ms: number): Promise { + return new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + const settle = () => { + clearTimeout(timer); + resolve(); + }; + void work.then(settle, settle); + }); } diff --git a/frontend/src/features/assistant/application/AssistantConversationService.ts b/frontend/src/features/assistant/application/AssistantConversationService.ts index 6e955cec..c2f9c28a 100644 --- a/frontend/src/features/assistant/application/AssistantConversationService.ts +++ b/frontend/src/features/assistant/application/AssistantConversationService.ts @@ -255,11 +255,22 @@ export class AssistantConversationService implements AssistantApplicationPort { this.soundLevel = null; this.replyText = null; this.currentAudioId = null; - // 连接马上就要整个关掉,不会再有消息进来;清空避免长会话里攒一堆再也 - // 用不上的 id。 + // 清空避免长会话里攒一堆再也用不上的 id;连接不会整个关掉(见下面), + // 但这一路监听马上就要摘掉,不会再有消息进来。 this.abandonedAudioIds.clear(); const connection = this.connection; + const streamId = this.streamId; + // 服务端收到 voice.stream.start 那一刻就把这个 session 标成"有一条活跃 + // 流",只有匹配的 voice.stream.end 才能解开。connection.close() 解决不了 + // 这件事——语音这条连接是共享的 AuthenticatedWebSocketClient,close() 只 + // 解绑本地监听,不断真实的 WebSocket(见 AuthenticatedVoiceTransport 的 + // 注释)。不发这一条,服务端会把这条流永远当成活跃的,下一次按住说话会 + // 被直接拒绝——而且这条拒绝路径服务端不打日志,表现成"按下去完全没反应" + // (实测复现,上滑取消手势最容易触发)。 + if (connection !== null && streamId !== null) { + connection.send({ payload: { stream_id: streamId }, type: 'voice.stream.end' }); + } this.unsubscribeConnection?.(); this.unsubscribeConnection = null; this.connection = null; @@ -290,6 +301,12 @@ export class AssistantConversationService implements AssistantApplicationPort { this.disposed = true; this.pendingCategoryUpdates.clear(); this.listeners.clear(); + // 同样的原因见 cancelTurn():卸载时如果还有一条活跃流(比如按住说话时 + // 组件被卸载),必须先结束它,否则服务端会把这个 session 永远卡在 + // "有活跃流",后续同一条连接上的每一次按住说话都会被拒绝。 + if (this.connection !== null && this.streamId !== null) { + this.connection.send({ payload: { stream_id: this.streamId }, type: 'voice.stream.end' }); + } this.unsubscribeConnection?.(); this.unsubscribeConnection = null; this.connection?.close(); diff --git a/frontend/src/features/assistant/data/audio/ExpoAudioPlayback.ts b/frontend/src/features/assistant/data/audio/ExpoAudioPlayback.ts index 3d17f85c..933457cc 100644 --- a/frontend/src/features/assistant/data/audio/ExpoAudioPlayback.ts +++ b/frontend/src/features/assistant/data/audio/ExpoAudioPlayback.ts @@ -20,6 +20,24 @@ const BYTES_PER_SAMPLE = 2; */ const SPLIT_MS = 100; +/** + * 开播前先攒够这么久的音频,再一次性交给原生播放器。 + * + * 读原生模块的源码(AudioPlaybackManager.playChunk)才看明白卡在哪:它的队列是 + * Channel.UNLIMITED,数据一旦交过去就跟 JS 无关了;但播放循环是「阻塞写一块 → 空等 + * 这块时长的 50% → 再取下一块」,稳态下供给和消费刚好打平,而 AudioTrack 的缓冲 + * (minBufferSize*2)只有几十毫秒。也就是说余量薄到几乎贴着底线,队列一见底就是一个 + * 听得见的空隙。 + * + * 队列什么时候最薄?每条回复刚开始那几百毫秒——第一帧到了就立刻开播,后面一帧稍微 + * 晚一点就断。往后 vendor 生成快于播放,队列很快攒起来,就再也断不了了。所以要补的 + * 是起跑余量,不是「在 JS 里囤数据防 JS 卡顿」(JS 卡顿在队列有存货时根本不影响)。 + * + * 代价是首字音频晚这么久。200ms 够盖住一帧的到达抖动,相对整轮 ~2 秒的响应约 10%。 + * 还听得到断音就调大,嫌慢就调小——只有这一个常量。 + */ +const PREBUFFER_MS = 200; + /** * TTS 回复的流式播放真实实现:voice.tts.start 开一条流、陆续 pushChunk、 * voice.tts.end 收尾。同一条流内所有分片用同一个 streamId,播放端靠它保序。 @@ -34,14 +52,32 @@ const SPLIT_MS = 100; export class ExpoAudioPlayback implements AssistantAudioPlaybackPort { private streamId: string | null = null; private splitBytes = 0; + /** 光靠 Date.now() 不够:两条回复前后脚开流会拿到同一个毫秒、同一个 id, + * pushChunk 里"这条流还是不是当前那条"的判断就形同虚设,原生播放器也没法 + * 靠 id 把两条流分开。 */ + private streamCounter = 0; + /** 起跑余量的门槛,按这条流的采样率算出来的字节数。 */ + private prebufferBytes = 0; + /** 还没交给播放器、正攒着凑起跑余量的块;攒够或收尾时一次性交出去。 */ + private pending: ArrayBuffer[] = []; + private pendingBytes = 0; + /** 起跑余量已经交出去了,之后的块直通,不再攒。 */ + private primed = false; async startStream(format: { sampleRateHz: number; encoding: 'pcm_s16le' }): Promise { - this.streamId = `assistant-tts-${Date.now()}`; + this.streamCounter += 1; + this.streamId = `assistant-tts-${Date.now()}-${this.streamCounter}`; // 每块字节数 = 采样率 × 每样本字节数 × 目标时长(秒)。 this.splitBytes = Math.max( 1, Math.round(format.sampleRateHz * BYTES_PER_SAMPLE * (SPLIT_MS / 1000)), ); + this.prebufferBytes = Math.max( + 1, + Math.round(format.sampleRateHz * BYTES_PER_SAMPLE * (PREBUFFER_MS / 1000)), + ); + // 每条回复各攒各的:上一条播完之后原生队列就空了,新的一条同样是从零起跑。 + this.discardPending(); await ExpoPlayAudioStream.setSoundConfig({ // 服务端 TTS 是 24000Hz,而包里 SoundConfig.sampleRate 的 TS 类型只列了 // 16000|44100|48000。这个类型比原生窄:Android 侧是 @@ -53,24 +89,72 @@ export class ExpoAudioPlayback implements AssistantAudioPlaybackPort { } async pushChunk(chunk: ArrayBuffer): Promise { - if (this.streamId === null || this.splitBytes <= 0) { + const streamId = this.streamId; + if (streamId === null || this.splitBytes <= 0) { throw new Error('pushChunk called before startStream'); } - for (const piece of splitPcm(chunk, this.splitBytes)) { - await ExpoPlayAudioStream.playAudio( - arrayBufferToBase64(piece), - this.streamId, - EncodingTypes.PCM_S16LE, - ); + if (this.primed) { + await this.write(streamId, [chunk]); + return; + } + this.pending.push(chunk); + this.pendingBytes += chunk.byteLength; + if (this.pendingBytes < this.prebufferBytes) { + return; } + this.primed = true; + await this.write(streamId, this.takePending()); } async endStream(): Promise { + const streamId = this.streamId; + // 「好的。」这种短回复整条都可能不到门槛。收尾时必须把攒着的交出去,否则这句 + // 回复一个字都不会响。 + if (streamId !== null && this.pendingBytes > 0) { + this.primed = true; + await this.write(streamId, this.takePending()); + } this.streamId = null; + this.discardPending(); } async stop(): Promise { this.streamId = null; + // 攒着的那段属于被放弃的那条回复,不能留到下一条流开起来再补播出去。 + this.discardPending(); await ExpoPlayAudioStream.stopAudio(); } + + /** 把这些块切成小片依次写进播放器,中途这条流被换掉就停手。 */ + private async write(streamId: string, chunks: readonly ArrayBuffer[]): Promise { + for (const chunk of chunks) { + for (const piece of splitPcm(chunk, this.splitBytes)) { + // 每片之前都重新确认这条流还是不是当前那条:上一片还卡在原生桥上的时候, + // stop()(打断)或下一条回复的 startStream() 可能已经把它换掉了。不确认的话 + // 剩下的片会在 stopAudio() 之后继续写进播放器,把刚被打断那句的尾巴放出来; + // 而且那时 this.streamId 已经是 null,等于用一条野生的流去播。 + if (this.streamId !== streamId) { + return; + } + await ExpoPlayAudioStream.playAudio( + arrayBufferToBase64(piece), + streamId, + EncodingTypes.PCM_S16LE, + ); + } + } + } + + private takePending(): ArrayBuffer[] { + const queued = this.pending; + this.pending = []; + this.pendingBytes = 0; + return queued; + } + + private discardPending(): void { + this.pending = []; + this.pendingBytes = 0; + this.primed = false; + } } diff --git a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts index 7897960d..1b4e6b01 100644 --- a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts +++ b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts @@ -19,6 +19,8 @@ import type { import type { LocationProvider } from '../../../../../src/infrastructure/location/LocationProvider'; const SESSION_IDLE_TIMEOUT_MS = 180_000; +const NOTIFY_COALESCE_MS = 50; +const NATIVE_TEARDOWN_TIMEOUT_MS = 1000; /** 跟 AssistantConversationService.test.ts 用同一套理由:startTurn() 里好几层 await,固定多轮 flush 比猜跳数稳。 */ async function flushAsync(iterations = 20): Promise { @@ -255,6 +257,324 @@ describe('AssistantContinuousConversationService', () => { ]); }); + it('does not leave a second copy of a reply that started before its transcript', async () => { + // 后端日志实测:vendor 的 transcription.completed 是在回复开始之后才发的,有时 + // 甚至晚于 response.done。也就是说"回复先于转写到达"根本不是异常分支,而是每 + // 通电话第一轮的常态。回复的头几个分片落进 turns 为空时的兜底列表、剩下的分片 + // 落进转写开的那一轮,同一句话就变成两个气泡——而且兜底列表永远渲染在最后、 + // 整通电话都不会消失(用户描述的"有一条一直存在")。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: false, reply_id: 'reply_1', speech_text: '电影' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { duration_ms: 500, language: 'zh', transcript: '明天我将会去看电影。' }, + type: 'voice.asr.completed', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_1', speech_text: '电影几点开始?' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getMessages()).toEqual([ + { id: 'turn-0:user', role: 'user', text: '明天我将会去看电影。' }, + { id: 'turn-0:assistant', role: 'assistant', text: '电影几点开始?' }, + ]); + }); + + it('lets a whole reply that beat its transcript still collect that transcript', async () => { + // 同一件事的另一半时序:整条回复(含 done=true)都在转写之前到达。回复开的占位 + // 轮必须能被随后到达的转写补上,而不是各自成为一条孤立的消息。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_1', speech_text: '电影几点开始?' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { duration_ms: 500, language: 'zh', transcript: '明天我将会去看电影。' }, + type: 'voice.asr.completed', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getTurns()).toEqual([ + { id: 'turn-0', replyText: '电影几点开始?', transcript: '明天我将会去看电影。' }, + ]); + }); + + it('pairs each reply with its own utterance when the backend names the turn', async () => { + // 到达顺序在这个时序下必然猜错:两轮的回复都在任何一条转写之前完成,按"填最后 + // 一轮"会把问题A配到回复B上,回复A丢掉提问、问题B丢掉回复(实测三条全错位)。 + // realtime 后端现在把 vendor 的输入 item id 带在两边,配对不再靠猜。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_A', speech_text: '回复A', turn_id: 'item_1' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_B', speech_text: '回复B', turn_id: 'item_2' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { duration_ms: 1, language: 'zh', transcript: '问题A', turn_id: 'item_1' }, + type: 'voice.asr.completed', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { duration_ms: 1, language: 'zh', transcript: '问题B', turn_id: 'item_2' }, + type: 'voice.asr.completed', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getTurns()).toEqual([ + { id: 'item_1', replyText: '回复A', transcript: '问题A' }, + { id: 'item_2', replyText: '回复B', transcript: '问题B' }, + ]); + }); + + it('files a clarifying question under the utterance it asks about', async () => { + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { duration_ms: 1, language: 'zh', transcript: '明天看电影', turn_id: 'item_1' }, + type: 'voice.asr.completed', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { + candidates: [], + question_id: 'q_1', + question_kind: 'missing_field', + speech_text: '电影几点开始?', + turn_id: 'item_1', + }, + type: 'voice.dialogue.question', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getTurns()).toEqual([ + { id: 'item_1', replyText: '电影几点开始?', transcript: '明天看电影' }, + ]); + }); + + it('still falls back to arrival order when the backend names no turn', async () => { + // composed 后端拿不到 vendor 的 item id,字段是 null,老那套按到达顺序的归属 + // 必须继续有效——不能因为加了 id 就把没有 id 的那条路走坏。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { duration_ms: 1, language: 'zh', transcript: '明天看电影' }, + type: 'voice.asr.completed', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_A', speech_text: '好的' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getMessages()).toEqual([ + { id: 'turn-0:user', role: 'user', text: '明天看电影' }, + { id: 'turn-0:assistant', role: 'assistant', text: '好的' }, + ]); + }); + + it('coalesces streaming reply fragments into one notification without delaying the text', async () => { + // vendor 的回复文字是逐词下发的(实测一句话约 20 条、间隔 12~16ms)。每条都通知 + // 一次就是每条都整屏重渲染,而播放链路每 ~50ms 就得把下一块 PCM 喂进原生播放器 + // ——被渲染挤掉就是一次听得见的空隙。合并的只是"通知界面重画",状态本身同步更新。 + jest.useFakeTimers(); + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + await startListening(fake, service); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { duration_ms: 1, language: 'zh', transcript: '明天看电影' }, + type: 'voice.asr.completed', + } as AssistantServerMessage); + await flushAsync(); + + let notifications = 0; + service.subscribe(() => { + notifications += 1; + }); + for (const text of ['明天', '明天中午', '明天中午12点', '明天中午12点看电影']) { + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: false, reply_id: 'reply_1', speech_text: text }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + } + await flushAsync(); + + // 四条分片之间一次都没通知,但内容已经是最新的了。 + expect(notifications).toBe(0); + expect(service.getReplyText()).toBe('明天中午12点看电影'); + + await advanceAndFlush(NOTIFY_COALESCE_MS); + expect(notifications).toBe(1); + }); + + it('does not make the last fragment of a reply wait for the coalescing window', async () => { + jest.useFakeTimers(); + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + await startListening(fake, service); + + let notifications = 0; + service.subscribe(() => { + notifications += 1; + }); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_1', speech_text: '好的' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + await flushAsync(); + + expect(notifications).toBe(1); + }); + + it('stops re-rendering for microphone level while a reply is playing', async () => { + // 放 TTS 时光球显示的是麦克风能量,本来就没有意义,而每刷一次都要跟播放链路 + // 抢同一条 JS 线程。音量照常记着,只是不为它单独刷界面。 + jest.useFakeTimers(); + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + await startListening(fake, service); + fake.emitMessage({ + audio_id: 'audio_1', + conversation_id: 'conv_001', + payload: { format: 'pcm', purpose: 'command_result', sample_rate_hz: 24000 }, + type: 'voice.tts.start', + } as AssistantServerMessage); + await flushAsync(); + + let notifications = 0; + service.subscribe(() => { + notifications += 1; + }); + deps.emitMicChunk(new ArrayBuffer(8), -20); + await advanceAndFlush(NOTIFY_COALESCE_MS * 2); + + expect(notifications).toBe(0); + // 但音量本身记下来了,下一次真正的状态变化会把它一起带出去。 + expect(service.getSoundLevel()).toBe(-20); + }); + + it('still hangs up when the native player never answers stop()', async () => { + // 实测症状:结束对话按了没反应,而且那之后 TTS 也不出声。两件事同一个根: + // hangUp 里清 endTurnInFlight 的 finally 只盖住最后一个 try,前面两个 await + // 卡住(不是抛错,是不返回)就永远走不到,之后每一次按都被开头那道门槛挡掉。 + jest.useFakeTimers(); + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + deps.playback.stop = jest.fn(() => new Promise(() => {})); + const service = createService(deps); + await startListening(fake, service); + + void service.endTurn(); + // 先让 hangUp 跑到装定时器那一步,再推时钟——限时是在 await 之后才装上的。 + await flushAsync(); + await advanceAndFlush(NATIVE_TEARDOWN_TIMEOUT_MS); + await advanceAndFlush(NATIVE_TEARDOWN_TIMEOUT_MS); + + expect(service.getState()).toEqual({ phase: 'idle' }); + // 服务端那条流也必须放掉,否则下一次 voice.stream.start 会被当成"已有活跃流"拒绝。 + expect(fake.sent).toContainEqual({ + payload: { stream_id: 'stream_001' }, + type: 'voice.stream.end', + }); + }); + + it('still hangs up when the native recorder never answers stop()', async () => { + jest.useFakeTimers(); + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + deps.capture.stop = jest.fn(() => new Promise(() => {})); + const service = createService(deps); + await startListening(fake, service); + + void service.endTurn(); + // 先让 hangUp 跑到装定时器那一步,再推时钟——限时是在 await 之后才装上的。 + await flushAsync(); + await advanceAndFlush(NATIVE_TEARDOWN_TIMEOUT_MS); + await advanceAndFlush(NATIVE_TEARDOWN_TIMEOUT_MS); + + expect(service.getState()).toEqual({ phase: 'idle' }); + }); + + it('does not let a stop() that never answers silence the rest of the call', async () => { + // playbackChain 会被换成那个永不 settle 的 promise,之后每一块音频都排在它 + // 后面——整通电话再也不出声。打断(tts.canceled)就会走到这条路上。 + jest.useFakeTimers(); + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + deps.playback.stop = jest.fn(() => new Promise(() => {})); + const service = createService(deps); + await startListening(fake, service); + + fake.emitMessage({ + audio_id: 'audio_1', + conversation_id: 'conv_001', + payload: { format: 'pcm', purpose: 'command_result', sample_rate_hz: 24000 }, + type: 'voice.tts.start', + } as AssistantServerMessage); + await flushAsync(); + fake.emitMessage({ + audio_id: 'audio_1', + conversation_id: 'conv_001', + type: 'voice.tts.canceled', + } as AssistantServerMessage); + await flushAsync(); + await advanceAndFlush(NATIVE_TEARDOWN_TIMEOUT_MS); + + // 打断之后模型接着回答:这一轮的音频必须照常送进播放器。 + fake.emitMessage({ + audio_id: 'audio_2', + conversation_id: 'conv_001', + payload: { format: 'pcm', purpose: 'command_result', sample_rate_hz: 24000 }, + type: 'voice.tts.start', + } as AssistantServerMessage); + await flushAsync(); + fake.emitAudioFrame(new ArrayBuffer(4)); + await flushAsync(); + + expect(deps.playback.pushChunk).toHaveBeenCalled(); + }); + it('keeps a streaming assistant reply pending until it is done', async () => { const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); @@ -268,7 +588,7 @@ describe('AssistantContinuousConversationService', () => { } as AssistantServerMessage); await flushAsync(); expect(service.getMessages()).toEqual([ - { id: 'reply_1', pending: true, role: 'assistant', text: '明天' }, + { id: 'turn-0:assistant', pending: true, role: 'assistant', text: '明天' }, ]); fake.emitMessage({ @@ -278,7 +598,7 @@ describe('AssistantContinuousConversationService', () => { } as AssistantServerMessage); await flushAsync(); expect(service.getMessages()).toEqual([ - { id: 'reply_1', role: 'assistant', text: '明天下午三点' }, + { id: 'turn-0:assistant', role: 'assistant', text: '明天下午三点' }, ]); }); @@ -644,6 +964,47 @@ describe('AssistantContinuousConversationService', () => { ]); }); + it('drops a stale duplicate of the just-completed reply instead of opening a new turn', async () => { + // 防御性加固:正常情况下后端不会在 done=true 之后又发一条同 reply_id 的 + // 重复消息,但连续模式判断"是不是新一轮"完全靠 lastTurnReplyDone 这个 + // 到达顺序猜测——一旦真的收到一条迟到/重复的消息,旧逻辑会把它当成 + // "还没到的新一轮",凭空开一个占位轮,造成重复气泡(跟 vendor 凭空多起 + // response 那次实测的现象一样)。reply_id 没变就不该开新轮。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { duration_ms: 800, language: 'zh', transcript: '明天几点开会' }, + type: 'voice.asr.completed', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_1', speech_text: '明天下午三点' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + await flushAsync(); + expect(service.getMessages()).toEqual([ + { id: 'turn-0:user', role: 'user', text: '明天几点开会' }, + { id: 'turn-0:assistant', role: 'assistant', text: '明天下午三点' }, + ]); + + // 同一个 reply_id 的重复投递。 + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_1', speech_text: '明天下午三点' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getMessages()).toEqual([ + { id: 'turn-0:user', role: 'user', text: '明天几点开会' }, + { id: 'turn-0:assistant', role: 'assistant', text: '明天下午三点' }, + ]); + }); + it('records an assistant bubble even when a reply arrives before any transcript', async () => { const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); @@ -658,7 +1019,9 @@ describe('AssistantContinuousConversationService', () => { await flushAsync(); expect(service.getReplyText()).toBe('好的'); - expect(service.getMessages()).toEqual([{ id: 'reply_1', role: 'assistant', text: '好的' }]); + expect(service.getMessages()).toEqual([ + { id: 'turn-0:assistant', role: 'assistant', text: '好的' }, + ]); }); it('ignores a reply that arrives before any transcript with blank speech text', async () => { @@ -677,10 +1040,10 @@ describe('AssistantContinuousConversationService', () => { expect(service.getMessages()).toEqual([]); }); - it('records a clarifying question as an orphan bubble when it arrives before any transcript', async () => { - // voice.dialogue.question 跟 voice.dialogue.reply 一样,正常流程不会先于 - // 任何 transcript 到达,但要是真的发生了,得挂到零散气泡列表兜底,而不是 - // 丢掉。同一个 question_id 再来一次要原地更新,不能变成两条气泡。 + it('opens a turn for a clarifying question that arrives before any transcript', async () => { + // voice.dialogue.question 跟 voice.dialogue.reply 一样会先于转写到达,得开一个 + // 占位轮挂上去,等转写到了补进同一轮,而不是丢掉。同一个 question_id 再来一次 + // 要原地更新,不能变成两条气泡。 const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); const service = createService(deps); @@ -699,7 +1062,7 @@ describe('AssistantContinuousConversationService', () => { await flushAsync(); expect(service.getMessages()).toEqual([ - { id: 'q_1', role: 'assistant', text: '你是想订哪一天的会议室?' }, + { id: 'turn-0:assistant', role: 'assistant', text: '你是想订哪一天的会议室?' }, ]); fake.emitMessage({ @@ -715,7 +1078,11 @@ describe('AssistantContinuousConversationService', () => { await flushAsync(); expect(service.getMessages()).toEqual([ - { id: 'q_1', role: 'assistant', text: '你是想订哪一天的会议室,上午还是下午?' }, + { + id: 'turn-0:assistant', + role: 'assistant', + text: '你是想订哪一天的会议室,上午还是下午?', + }, ]); }); @@ -1249,7 +1616,7 @@ describe('AssistantContinuousConversationService', () => { disposeService(service); }); - it('stops immediately and drops queued chunks when TTS is canceled', async () => { + it('stops the player once the in-flight write lands, and drops what was queued behind it', async () => { const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); const service = createService(deps); @@ -1292,7 +1659,11 @@ describe('AssistantContinuousConversationService', () => { conversation_id: 'conv_001', type: 'voice.tts.canceled', } as AssistantServerMessage); - expect(calls).toEqual(['push-1', 'stop']); + // stop 排在在途的 push-1 后面,不跟它并发。原来是绕过队列立刻发的,理由写的是 + // "否则已排队的 PCM 会先继续喂给播放器"——那个理由不成立(代次检查已经把排队的 + // 都作废了,见下面 push-2),而并发的代价是真的:stopAudio 撞上原生侧正在进行 + // 的写入,实测会不返回,然后 playbackChain 就死在那儿,整通电话再不出声。 + expect(calls).toEqual(['push-1']); resolveFirst(); await flushAsync(); @@ -1338,6 +1709,9 @@ describe('AssistantContinuousConversationService', () => { await flushAsync(); disposeService(service); + // stop 现在排在 playbackChain 末尾(不再跟在途的原生写入并发),所以晚一个 + // 微任务才真正调用——保证没变,时机变了。 + await flushAsync(); expect(deps.playback.stop).toHaveBeenCalled(); }); @@ -1434,6 +1808,51 @@ describe('AssistantContinuousConversationService', () => { disposeService(service); }); + it('tracks every abandoned audio id, not just the most recently abandoned one', async () => { + // 跟 AssistantConversationService(PTT)里同名 bug 一样的结构:canceledAudioId + // 原来是单个变量,两次打断前后脚发生、第一条的 tts.end 还没到就轮到第二条时, + // 第一条的 id 会被第二条覆盖掉——它自己迟到的 tts.end 就会被误判成"正常收尾", + // 多余地调一次 endStream() 并把第二条也应该维持住的取消状态冲掉。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + const startMessage = (audioId: string): AssistantServerMessage => + ({ + audio_id: audioId, + conversation_id: 'conv_001', + payload: { format: 'pcm_s16le', purpose: 'reply', sample_rate_hz: 24000, speech_text: '' }, + type: 'voice.tts.start', + }) as AssistantServerMessage; + const canceledMessage = (audioId: string): AssistantServerMessage => + ({ + audio_id: audioId, + conversation_id: 'conv_001', + type: 'voice.tts.canceled', + }) as AssistantServerMessage; + const endMessage = (audioId: string): AssistantServerMessage => + ({ + audio_id: audioId, + conversation_id: 'conv_001', + type: 'voice.tts.end', + }) as AssistantServerMessage; + + await startListening(fake, service); + fake.emitMessage(startMessage('audio_001')); + fake.emitMessage(canceledMessage('audio_001')); + fake.emitMessage(startMessage('audio_002')); + fake.emitMessage(canceledMessage('audio_002')); + await flushAsync(); + + // 两条都迟到的收尾消息,谁先谁后都不该被当成真正的收尾。 + fake.emitMessage(endMessage('audio_001')); + fake.emitMessage(endMessage('audio_002')); + await flushAsync(); + + expect(deps.playback.endStream).not.toHaveBeenCalled(); + disposeService(service); + }); + it('dismissReply() clears the reply bubble and stops playback immediately', async () => { const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); @@ -1558,4 +1977,58 @@ describe('AssistantContinuousConversationService', () => { expect(retryOnChunk).not.toBeNull(); expect(service.getState()).toEqual({ conversationId: 'conv_001', phase: 'listening' }); }); + + it('hangs up the call when the server rejects an audio frame mid-conversation', async () => { + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + // 服务端这一侧的流已经没了(后端 agent 提前退出/流被回收),每一个音频帧都会 + // 换回来一条这样的错误。只把 phase 设成 error 是不够的:麦克风还开着,帧还在 + // 发,服务端就一帧回一条错误,直到用户自己想起来点"结束对话"。 + fake.emitMessage({ + error: { + code: 'AUDIO_INVALID', + message: 'Audio frames require voice.stream.start first', + retryable: false, + }, + ok: false, + type: 'voice.command.error', + } as AssistantServerMessage); + await flushAsync(); + + expect(deps.capture.stop).toHaveBeenCalled(); + // 服务端那条流必须放掉,否则下一次 voice.stream.start 会被当成"已有活跃流"拒绝。 + expect(fake.sent).toContainEqual({ + payload: { stream_id: 'stream_001' }, + type: 'voice.stream.end', + }); + expect(fake.unsubscribeCalls).toEqual({ audio: 1, close: 1, message: 1 }); + // 报错原因要留在界面上,不能像正常挂断那样直接归 idle 悄悄消失。 + expect(service.getState()).toEqual({ + message: 'Audio frames require voice.stream.start first', + phase: 'error', + }); + }); + + it('does not fight a hangup already in progress when an error arrives during it', async () => { + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + + await startListening(fake, service); + const ending = service.endTurn(); + // 我们自己发的 voice.stream.end 之后,服务端可能补一条"这条流没有音频"的错误。 + // 挂断已经在路上了,它不该把已经归位的状态又掰成 error。 + fake.emitMessage({ + error: { code: 'AUDIO_INVALID', message: 'The audio stream carried no audio' }, + ok: false, + type: 'voice.command.error', + } as AssistantServerMessage); + await ending; + await flushAsync(); + + expect(service.getState()).toEqual({ phase: 'idle' }); + }); }); diff --git a/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts index 321020c3..0ba1447c 100644 --- a/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts +++ b/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts @@ -254,7 +254,11 @@ describe('AssistantConversationService', () => { expect(service.getState()).toEqual({ conversationId: 'conv_002', phase: 'recording' }); }); - it('cancels a recording without ending the stream or applying a late command', async () => { + it('ends the active stream before closing so the server does not keep it marked active forever', async () => { + // 回归:connection.close() 不会真的断开共享连接(AuthenticatedVoiceTransport + // 只解绑本地监听),服务端只认 voice.stream.end。不发这一条,服务端会把 + // 这个 session 永远标成"有活跃流",下一次按住说话直接被拒绝,且这条拒绝 + // 服务端不打日志,表现成按下去完全没反应——上滑取消手势实测复现过。 const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); @@ -265,7 +269,10 @@ describe('AssistantConversationService', () => { await service.cancelTurn(); expect(deps.capture.stop).toHaveBeenCalledTimes(1); - expect(fake.sent.filter((message) => message.type === 'voice.stream.end')).toHaveLength(0); + expect(fake.sent).toContainEqual({ + payload: { stream_id: 'stream_001' }, + type: 'voice.stream.end', + }); expect(fake.closeCalls.count).toBe(1); expect(service.getState()).toEqual({ phase: 'idle' }); @@ -279,6 +286,22 @@ describe('AssistantConversationService', () => { expect(deps.localScheduleWriter.applyCommandResult).not.toHaveBeenCalled(); }); + it('does not send voice.stream.end when cancelling before any stream was opened', async () => { + const fake = createFakeConnection(); + const deps = createDeps({ + connection: fake.connection, + requestPermission: () => new Promise(() => {}), + }); + const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); + + void service.startTurn(); + await flushAsync(); + + await service.cancelTurn(); + + expect(fake.sent.filter((message) => message.type === 'voice.stream.end')).toHaveLength(0); + }); + it('sends voice.stream.end and reports an error when capture.start() fails after the stream opened', async () => { const fake = createFakeConnection(); const deps = createDeps({ @@ -836,6 +859,21 @@ describe('AssistantConversationService', () => { expect(fake.closeCalls.count).toBe(1); }); + it('ends the active stream on dispose() too, same reason as cancelTurn()', async () => { + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); + + await completeStreamStart(fake, service.startTurn()); + + service.dispose(); + + expect(fake.sent).toContainEqual({ + payload: { stream_id: 'stream_001' }, + type: 'voice.stream.end', + }); + }); + it('tags every voice.stream.start with a fresh per-turn request_id', async () => { const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); diff --git a/frontend/tests/unit/features/assistant/data/audio/ExpoAudioPlayback.test.ts b/frontend/tests/unit/features/assistant/data/audio/ExpoAudioPlayback.test.ts new file mode 100644 index 00000000..fe5f8d21 --- /dev/null +++ b/frontend/tests/unit/features/assistant/data/audio/ExpoAudioPlayback.test.ts @@ -0,0 +1,165 @@ +import { beforeEach, describe, expect, it, jest } from '@jest/globals'; + +const mockPlayAudio = jest.fn<(...args: unknown[]) => Promise>(); +const mockSetSoundConfig = jest.fn<(...args: unknown[]) => Promise>(); +const mockStopAudio = jest.fn<(...args: unknown[]) => Promise>(); + +jest.mock('@irvingouj/expo-audio-stream', () => ({ + EncodingTypes: { PCM_S16LE: 'pcm_s16le' }, + ExpoPlayAudioStream: { + playAudio: (...args: unknown[]) => mockPlayAudio(...args), + setSoundConfig: (...args: unknown[]) => mockSetSoundConfig(...args), + stopAudio: (...args: unknown[]) => mockStopAudio(...args), + }, +})); + +// eslint-disable-next-line import/first +import { ExpoAudioPlayback } from '../../../../../../src/features/assistant/data/audio/ExpoAudioPlayback'; + +/** 24000Hz、16bit 单声道下 100ms 一片 = 4800 字节,跟 SPLIT_MS 对齐。 */ +const PIECE_BYTES = 4800; + +const FORMAT = { encoding: 'pcm_s16le', sampleRateHz: 24000 } as const; + +describe('ExpoAudioPlayback (assistant TTS stream)', () => { + beforeEach(() => { + mockPlayAudio.mockReset(); + mockSetSoundConfig.mockReset(); + mockStopAudio.mockReset(); + mockPlayAudio.mockResolvedValue(undefined); + mockSetSoundConfig.mockResolvedValue(undefined); + mockStopAudio.mockResolvedValue(undefined); + }); + + it('splits one chunk into pieces and writes them under the same stream id', async () => { + const playback = new ExpoAudioPlayback(); + await playback.startStream(FORMAT); + + await playback.pushChunk(new ArrayBuffer(PIECE_BYTES * 3)); + + expect(mockPlayAudio).toHaveBeenCalledTimes(3); + const streamIds = new Set(mockPlayAudio.mock.calls.map(([, streamId]) => streamId)); + expect(streamIds.size).toBe(1); + expect([...streamIds][0]).toEqual(expect.stringContaining('assistant-tts-')); + }); + + it('stops writing the rest of a chunk once the stream it belongs to is gone', async () => { + // 打断的真实时序:stop() 在上一片还卡在原生桥上时到达。不重新确认这条流还 + // 是不是当前那条的话,剩下的片会在 stopAudio() 之后继续写进播放器,把刚被 + // 打断那句的尾巴放出来——而且那时 streamId 已经是 null,等于用一条野生的流播。 + const playback = new ExpoAudioPlayback(); + await playback.startStream(FORMAT); + + let releaseFirstPiece: () => void = () => {}; + mockPlayAudio.mockImplementationOnce( + () => + new Promise((resolve) => { + releaseFirstPiece = resolve; + }), + ); + + const pushing = playback.pushChunk(new ArrayBuffer(PIECE_BYTES * 3)); + await playback.stop(); + releaseFirstPiece(); + await pushing; + + expect(mockStopAudio).toHaveBeenCalledTimes(1); + expect(mockPlayAudio).toHaveBeenCalledTimes(1); + expect(mockPlayAudio.mock.calls.every(([, streamId]) => streamId !== null)).toBe(true); + }); + + it('stops writing the rest of a chunk once a newer stream has taken over', async () => { + // 同一件事的另一半:没有 stop(),直接来了下一条回复的 voice.tts.start。 + const playback = new ExpoAudioPlayback(); + await playback.startStream(FORMAT); + + let releaseFirstPiece: () => void = () => {}; + mockPlayAudio.mockImplementationOnce( + () => + new Promise((resolve) => { + releaseFirstPiece = resolve; + }), + ); + + const pushing = playback.pushChunk(new ArrayBuffer(PIECE_BYTES * 3)); + await playback.startStream(FORMAT); + releaseFirstPiece(); + await pushing; + + expect(mockPlayAudio).toHaveBeenCalledTimes(1); + }); + + it('holds the opening audio back until the player has a head start', async () => { + // 原生模块的播放循环是「写一块、空等这块时长的 50%、再取下一块」,稳态下刚好 + // 打平,而 AudioTrack 的缓冲只有几十毫秒。队列一见底就是一个听得见的空隙,而 + // 每条回复刚开始的那几百毫秒队列恰恰最薄。先攒够一段再一次性交过去,让播放 + // 循环一开始就有存货。 + const playback = new ExpoAudioPlayback(); + await playback.startStream(FORMAT); + + await playback.pushChunk(new ArrayBuffer(PIECE_BYTES)); + expect(mockPlayAudio).not.toHaveBeenCalled(); + + await playback.pushChunk(new ArrayBuffer(PIECE_BYTES)); + // 攒够 200ms,两块一起交出去。 + expect(mockPlayAudio).toHaveBeenCalledTimes(2); + }); + + it('stops holding audio back once the player has its head start', async () => { + const playback = new ExpoAudioPlayback(); + await playback.startStream(FORMAT); + await playback.pushChunk(new ArrayBuffer(PIECE_BYTES * 2)); + mockPlayAudio.mockClear(); + + await playback.pushChunk(new ArrayBuffer(PIECE_BYTES)); + + expect(mockPlayAudio).toHaveBeenCalledTimes(1); + }); + + it('still plays a reply too short to reach the head start', async () => { + // 「好的。」这种一句话可能整条都不够门槛。收尾时必须把攒着的交出去,否则这句 + // 回复一个字都不会响。 + const playback = new ExpoAudioPlayback(); + await playback.startStream(FORMAT); + await playback.pushChunk(new ArrayBuffer(PIECE_BYTES)); + expect(mockPlayAudio).not.toHaveBeenCalled(); + + await playback.endStream(); + + expect(mockPlayAudio).toHaveBeenCalledTimes(1); + }); + + it('throws away audio still being held when playback is stopped', async () => { + // 打断时攒着的那段属于被放弃的那条回复,不能等下一条流开起来再补播出去。 + const playback = new ExpoAudioPlayback(); + await playback.startStream(FORMAT); + await playback.pushChunk(new ArrayBuffer(PIECE_BYTES)); + + await playback.stop(); + await playback.startStream(FORMAT); + await playback.pushChunk(new ArrayBuffer(PIECE_BYTES)); + + // 新流自己还没攒够,旧流那块也不该冒出来。 + expect(mockPlayAudio).not.toHaveBeenCalled(); + }); + + it('refuses a chunk pushed before any stream was opened', async () => { + const playback = new ExpoAudioPlayback(); + + await expect(playback.pushChunk(new ArrayBuffer(PIECE_BYTES))).rejects.toThrow( + 'pushChunk called before startStream', + ); + }); + + it('endStream() closes the stream without touching the native player', async () => { + const playback = new ExpoAudioPlayback(); + await playback.startStream(FORMAT); + + await playback.endStream(); + + expect(mockStopAudio).not.toHaveBeenCalled(); + await expect(playback.pushChunk(new ArrayBuffer(PIECE_BYTES))).rejects.toThrow( + 'pushChunk called before startStream', + ); + }); +});