From d9864e20e06f54b01f4a3b4d0bc3dc09ba0116c6 Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Fri, 21 Aug 2026 14:16:41 +0800 Subject: [PATCH 1/4] fix(realtime): serialize batch tool-call follow-ups and local command writes Batch voice commands that call several tools within one realtime response raced two ways: send_tool_result() asked the vendor for a follow-up reply right after each tool call instead of waiting for that response's own response.done, so a second in-batch tool call collided with the vendor's "one response in flight" invariant and force-ended the call; and the frontend applied each voice.command.result fire-and-forget, letting concurrent local SQLite transactions race on withExclusiveTransactionAsync and silently drop writes the cloud had already committed. Fixes 1024XEngineer/timeflow#341. Co-Authored-By: Claude Sonnet 5 --- .../external/realtime/qwen_audio.py | 61 ++++++++++++++----- .../external/realtime/test_qwen_audio.py | 18 ++++-- .../AssistantContinuousConversationService.ts | 15 ++++- .../AssistantConversationService.ts | 13 +++- 4 files changed, 87 insertions(+), 20 deletions(-) diff --git a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py index 4f13e0ba..8d509170 100644 --- a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py +++ b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py @@ -136,6 +136,17 @@ def __init__( # ran the tool, then whatever response.create it triggered actually carries the # reply. Both pump loops use this to wait for the last one before settling. self._open_responses = 0 + # Set by send_tool_result() while the response that requested the tool(s) is + # still in progress; consumed at that response's own response.done (see the two + # pump loops). A batch turn can carry several function calls in one response -- + # asking for the follow-up as each tool finishes, instead of once the response + # itself is done, would send response.create while the vendor still considers + # that response in progress and it rejects the request outright. + self._followup_requested = False + # Set instead of _followup_requested when a tool in the current response ended + # the conversation: no follow-up is wanted even if an earlier tool call in the + # same batch asked for one. + self._followup_suppressed = False # Continuous mode only: true between a response.created and its matching # response.done (or a cancellation). cancel_response() reads this to know # whether there is anything to tell the vendor to abandon. @@ -185,14 +196,20 @@ async def finish_input(self) -> None: self._open_responses += 1 async def send_tool_result(self, call_id: str, output: str, *, respond: bool = True) -> None: - """Write a tool's output back and let the model continue from it. + """Write a tool's output back and mark whether it should be followed by a reply. The output is written back either way, so the vendor's conversation history stays - complete for later turns. Asking for a reply is separate: response.create is - required in every mode when one is wanted -- the vendor's own turn_detection only - starts a turn from the user's audio, never from a tool result on its own -- but a - tool that ends the conversation has nothing to follow up, and asking anyway just - buys a second goodbye on top of the one the model already spoke. + complete for later turns. Asking for a reply is separate, and deliberately deferred: + response.create is required in every mode when one is wanted -- the vendor's own + turn_detection only starts a turn from the user's audio, never from a tool result on + its own -- but the response that requested this tool may still be in progress (a + batch turn can call several tools before that response is done), and the vendor + rejects a response.create sent while another response is still open. The two pump + loops send it exactly once, at that response's own response.done, once every tool + call belonging to it has been recorded here. A tool that ends the conversation has + nothing to follow up, and asking anyway just buys a second goodbye on top of the one + the model already spoke -- that takes priority over any other tool in the same batch + that asked for a reply. """ await self._send( { @@ -200,10 +217,10 @@ async def send_tool_result(self, call_id: str, output: str, *, respond: bool = T "item": {"type": "function_call_output", "call_id": call_id, "output": output}, } ) - if not respond: - return - await self._send({"type": "response.create"}) - self._open_responses += 1 + if respond: + self._followup_requested = True + else: + self._followup_suppressed = True async def cancel_response(self) -> None: """Tell the vendor to abandon its in-flight reply, if there is one. @@ -218,6 +235,8 @@ async def cancel_response(self) -> None: return self._responding = False self._open_responses = 0 + self._followup_requested = False + self._followup_suppressed = False await self._send({"type": "response.cancel"}) async def close(self) -> None: @@ -271,6 +290,13 @@ async def _pump_single_turn(self, observer: Observer) -> None: elif kind == "response.done": # Not the turn's end if a tool ran: the next response is the one that speaks. self._open_responses -= 1 + if self._followup_requested or self._followup_suppressed: + if self._followup_requested and not self._followup_suppressed: + await self._send({"type": "response.create"}) + self._open_responses += 1 + self._followup_requested = False + self._followup_suppressed = False + continue if self._open_responses <= 0: return elif kind == "error": @@ -330,13 +356,20 @@ async def _pump_continuous(self, observer: Observer) -> None: await observer.tool_requested(**requested) elif kind == "response.done": if self._open_responses > 0: - # A tool call inside this response asked the vendor for a - # follow-up (send_tool_result's response.create) -- that follow-up, - # not this response, is what actually finishes the turn. Settling + self._open_responses -= 1 + if self._followup_requested or self._followup_suppressed: + # One or more tool calls happened inside the response that just + # finished; ask for the follow-up only now that the vendor + # considers it done (see send_tool_result). That follow-up, not + # this response, is what actually finishes the turn -- settling # here would let a caller like end_conversation hang up before the # model's actual reply to it has even started. Same pattern as # _pump_single_turn's push-to-talk handling below. - self._open_responses -= 1 + if self._followup_requested and not self._followup_suppressed: + await self._send({"type": "response.create"}) + self._open_responses += 1 + self._followup_requested = False + self._followup_suppressed = False continue self._responding = False # The bytes just sent still take this long to actually play out on the diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py index 1569d828..6c8f179f 100644 --- a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -327,16 +327,22 @@ async def scenario() -> None: def test_sending_a_tool_result_lets_the_model_continue() -> None: - """The output is written back as a conversation item, then a reply is requested.""" + """The output is written back as a conversation item immediately; the reply is only + requested once the response that called the tool is itself done (see + send_tool_result's docstring -- asking any earlier collides with that still-open + response on the vendor's side). + """ async def scenario() -> None: - """Send a tool result and read back what was sent.""" - transport = FakeTransport() + """Send a tool result, then read back what was sent before and after settling.""" + transport = FakeTransport(_event("response.done"), _event("response.done")) session = QwenAudioSession(transport, CONFIG, PUSH_TO_TALK) + await session.finish_input() + transport.sent.clear() await session.send_tool_result("call_1", '{"count":2}') - assert transport.types() == ["conversation.item.create", "response.create"] + assert transport.types() == ["conversation.item.create"] item = transport.sent[0]["item"] assert item == { "type": "function_call_output", @@ -344,6 +350,10 @@ async def scenario() -> None: "output": '{"count":2}', } + await session.pump(RecordingObserver()) + + assert transport.types() == ["conversation.item.create", "response.create"] + asyncio.run(scenario()) diff --git a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts index 86a775c2..4e228f97 100644 --- a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts +++ b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts @@ -96,6 +96,12 @@ export class AssistantContinuousConversationService implements AssistantApplicat private playbackGeneration = 0; /** Category events can arrive before the command result creates their local row. */ private readonly pendingCategoryUpdates = new Map(); + /** 串起每条 voice.command.result 的本地落库；一次连续通话里模型可能连续触发多个 + * 工具调用（批量新建/删除），command.result 消息前后脚到达时若不排队，会有两个 + * applyCommandResultLocally() 同时各开一个 withExclusiveTransactionAsync（各自 + * 新建一条原生连接），互相踩 "database is locked" 甚至留下损坏的原生连接状态—— + * 跟 playbackChain 一个模式：不管上一条成功与否都排到下一条前面。 */ + private commandResultChain: Promise = Promise.resolve(); private disposed = false; private readonly unsubscribeAppState: () => void; @@ -383,7 +389,7 @@ export class AssistantContinuousConversationService implements AssistantApplicat }; // 状态立刻回到 listening（麦克风还开着），不等写库；message.ack 必须等 // 写库成功后才发，避免向服务端谎报已落库。 - void this.applyCommandResultLocally(command, message.message_id); + this.queueCommandResult(command, message.message_id); this.setState({ conversationId: message.conversation_id, phase: 'listening' }); return; } @@ -458,6 +464,13 @@ export class AssistantContinuousConversationService implements AssistantApplicat } } + private queueCommandResult(command: AppliedCommand, messageId: string): void { + this.commandResultChain = this.commandResultChain.then( + () => this.applyCommandResultLocally(command, messageId), + () => this.applyCommandResultLocally(command, messageId), + ); + } + private async applyCommandResultLocally( command: AppliedCommand, messageId: string, diff --git a/frontend/src/features/assistant/application/AssistantConversationService.ts b/frontend/src/features/assistant/application/AssistantConversationService.ts index 8b6383cf..72acb14f 100644 --- a/frontend/src/features/assistant/application/AssistantConversationService.ts +++ b/frontend/src/features/assistant/application/AssistantConversationService.ts @@ -47,6 +47,10 @@ export class AssistantConversationService implements AssistantApplicationPort { private pendingStartTurn: Promise | null = null; /** Category events can arrive before the command result creates their local row. */ private readonly pendingCategoryUpdates = new Map(); + /** 串起每条 voice.command.result 的本地落库，见 AssistantContinuousConversationService + * 里同名字段的注释——一次按住说话也可能在一句话里触发多个工具调用（批量新建/ + * 删除），不排队会让两个 applyCommandResultLocally() 并发抢同一个 SQLite 连接。 */ + private commandResultChain: Promise = Promise.resolve(); private disposed = false; constructor( @@ -239,7 +243,7 @@ export class AssistantConversationService implements AssistantApplicationPort { status: message.payload.status, }; // 状态立刻回到 idle，不等写库；message.ack 必须等写库成功才发（AGENTS.md §6）。 - void this.applyCommandResultLocally(command, message.message_id); + this.queueCommandResult(command, message.message_id); this.setState({ phase: 'idle' }); return; } @@ -277,6 +281,13 @@ export class AssistantConversationService implements AssistantApplicationPort { } } + private queueCommandResult(command: AppliedCommand, messageId: string): void { + this.commandResultChain = this.commandResultChain.then( + () => this.applyCommandResultLocally(command, messageId), + () => this.applyCommandResultLocally(command, messageId), + ); + } + private async applyCommandResultLocally( command: AppliedCommand, messageId: string, From 1c7488556b7344cc01f7a4858c3c5bff95d3dbd4 Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Fri, 21 Aug 2026 14:37:50 +0800 Subject: [PATCH 2/4] test(realtime): add regression coverage for batch tool-call serialization Both bugs fixed in 407ab61 only surface when a single turn triggers several tool calls (batch create/delete). These tests exercise that shape directly through the real service/session code: - Two voice.command.result messages arriving back-to-back (no flush between them) must apply strictly in order; the second must not start until the first's transaction settles. - Two tool calls landing in one still-open realtime response must produce exactly one deferred response.create, sent only after that response's own response.done. Verified each test actually catches the regression by reverting its corresponding source fix locally and confirming the new test fails, then restoring the fix and confirming it passes. Co-Authored-By: Claude Sonnet 5 --- .../external/realtime/test_qwen_audio.py | 71 +++++++++++++++++++ ...stantContinuousConversationService.test.ts | 70 ++++++++++++++++++ .../AssistantConversationService.test.ts | 70 ++++++++++++++++++ 3 files changed, 211 insertions(+) diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py index 6c8f179f..6baa60b9 100644 --- a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -380,6 +380,41 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_two_tool_calls_in_one_response_only_ask_for_a_single_followup() -> None: + """A batch turn can call several tools before the response that requested them is + done (e.g. a batch create/delete voice command). Each call must not eagerly ask + for its own follow-up -- doing so, one per tool call, sends a second + response.create while the vendor still considers the first response in progress, + and it rejects the request outright ("Cannot create response while another + response is in progress."). Exactly one follow-up should be asked for, once, + after that response's own response.done. + """ + + async def scenario() -> None: + transport = FakeTransport(_event("response.done"), _event("response.done")) + session = QwenAudioSession(transport, CONFIG, PUSH_TO_TALK) + await session.finish_input() + transport.sent.clear() + + await session.send_tool_result("call_1", '{"count":2}') + await session.send_tool_result("call_2", '{"count":3}') + + # Both tool outputs are written back immediately; neither asks for a reply yet. + assert transport.types() == ["conversation.item.create", "conversation.item.create"] + + await session.pump(RecordingObserver()) + + # The follow-up is requested exactly once, after the tool-calling response's + # own response.done -- not once per tool call. + assert transport.types() == [ + "conversation.item.create", + "conversation.item.create", + "response.create", + ] + + asyncio.run(scenario()) + + def test_first_response_done_does_not_end_a_tool_extended_turn() -> None: async def scenario() -> None: transport = FakeTransport( @@ -1080,6 +1115,42 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_continuous_two_tool_calls_in_one_response_only_ask_for_a_single_followup() -> None: + """Continuous mode's version of test_two_tool_calls_in_one_response_only_ask_for_a + _single_followup above: a batch voice command (e.g. "create three schedules") + can make the model call several tools inside one response before that response + is done. Asking for a follow-up per call, instead of once the response settles, + is what force-ends the whole conversation in production -- the vendor's + rejection is reported through failed(), which continuous mode treats as the + call having to hang up. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("response.created"), + _event("response.done"), + _event("response.created"), + _event("response.audio_transcript.done", transcript="好的，都办好了"), + _event("response.done"), + _event("error", error={"message": "stream ended"}), + ) + session = QwenAudioSession(transport, CONFIG, CONTINUOUS) + observer = RecordingObserver() + await session.send_tool_result("call-1", "{}") + await session.send_tool_result("call-2", "{}") + + await session.pump(observer) + + assert observer.calls == [ + ("spoke", "好的，都办好了"), + ("turn_completed", None), + ("failed", "stream ended"), + ] + assert transport.types().count("response.create") == 1 + + asyncio.run(scenario()) + + def test_continuous_pump_stops_when_a_frame_cannot_be_parsed() -> None: """A malformed frame ends a continuous stream the same way it ends push-to-talk.""" diff --git a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts index 9a1d105f..4c3d0908 100644 --- a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts +++ b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts @@ -453,6 +453,76 @@ describe('AssistantContinuousConversationService', () => { disposeService(service); }); + it('serializes local writes so a batch of back-to-back command results cannot race', async () => { + // Regression test for a batch create/delete race: the model can call several + // tools inside one turn, so voice.command.result messages can arrive only + // milliseconds apart (observed as low as 12ms in production logs). Each write + // opens its own withExclusiveTransactionAsync() on the real SQLite adapter -- + // running two at once opens two native connections that fight over the same + // exclusive lock ("database is locked"), and the loser's write is silently + // dropped even though the cloud already committed it. queueCommandResult() + // must serialize these instead of firing them concurrently. + const fake = createFakeConnection(); + const order: string[] = []; + let resolveFirst: (() => void) | undefined; + let inFlight = 0; + let overlapped = false; + let callCount = 0; + const deps = createDeps({ + applyCommandResult: async () => { + callCount += 1; + inFlight += 1; + if (inFlight > 1) overlapped = true; + if (callCount === 1) { + order.push('first-start'); + await new Promise((resolve) => { + resolveFirst = resolve; + }); + order.push('first-end'); + } else { + order.push('second-start'); + order.push('second-end'); + } + inFlight -= 1; + }, + connection: fake.connection, + }); + const service = createService(deps); + await startListening(fake, service); + + fake.emitMessage({ + conversation_id: 'conv_001', + message_id: 'msg_1', + payload: { operation: 'create_schedule', schedule: { id: 'a' }, status: 'applied' }, + type: 'voice.command.result', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + message_id: 'msg_2', + payload: { operation: 'create_schedule', schedule: { id: 'b' }, status: 'applied' }, + type: 'voice.command.result', + } as AssistantServerMessage); + await flushAsync(); + + // The second write must not have started yet -- the first is still stuck + // mid-transaction, waiting on resolveFirst. + expect(order).toEqual(['first-start']); + expect(overlapped).toBe(false); + + resolveFirst?.(); + await flushAsync(); + + expect(order).toEqual(['first-start', 'first-end', 'second-start', 'second-end']); + expect(overlapped).toBe(false); + expect(fake.sent).toContainEqual( + expect.objectContaining({ message_id: 'msg_1', type: 'message.ack' }), + ); + expect(fake.sent).toContainEqual( + expect.objectContaining({ message_id: 'msg_2', type: 'message.ack' }), + ); + disposeService(service); + }); + it('patches an asynchronous category update while the call remains active', async () => { const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); diff --git a/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts index 4a9f6c16..30c255d1 100644 --- a/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts +++ b/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts @@ -292,6 +292,76 @@ describe('AssistantConversationService', () => { expect(service.getLastAppliedCommand()).toBeNull(); }); + it('serializes local writes so a batch of back-to-back command results cannot race', async () => { + // Regression test: a single utterance can make the model call several tools in + // one turn, so voice.command.result messages can arrive only milliseconds + // apart. Each write opens its own withExclusiveTransactionAsync() on the real + // SQLite adapter -- running two at once opens two native connections that + // fight over the same exclusive lock ("database is locked"), and the loser's + // write is silently dropped even though the cloud already committed it. + // queueCommandResult() must serialize these instead of firing them + // concurrently. + const fake = createFakeConnection(); + const order: string[] = []; + let resolveFirst: (() => void) | undefined; + let inFlight = 0; + let overlapped = false; + let callCount = 0; + const deps = createDeps({ + applyCommandResult: async () => { + callCount += 1; + inFlight += 1; + if (inFlight > 1) overlapped = true; + if (callCount === 1) { + order.push('first-start'); + await new Promise((resolve) => { + resolveFirst = resolve; + }); + order.push('first-end'); + } else { + order.push('second-start'); + order.push('second-end'); + } + inFlight -= 1; + }, + connection: fake.connection, + }); + const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); + + const turn = service.startTurn(); + await completeStreamStart(fake, turn); + fake.emitMessage({ + conversation_id: 'conv_001', + message_id: 'msg_1', + payload: { operation: 'create_schedule', schedule: { id: 'a' }, status: 'applied' }, + type: 'voice.command.result', + } as AssistantServerMessage); + fake.emitMessage({ + conversation_id: 'conv_001', + message_id: 'msg_2', + payload: { operation: 'create_schedule', schedule: { id: 'b' }, status: 'applied' }, + type: 'voice.command.result', + } as AssistantServerMessage); + await flushAsync(); + + // The second write must not have started yet -- the first is still stuck + // mid-transaction, waiting on resolveFirst. + expect(order).toEqual(['first-start']); + expect(overlapped).toBe(false); + + resolveFirst?.(); + await flushAsync(); + + expect(order).toEqual(['first-start', 'first-end', 'second-start', 'second-end']); + expect(overlapped).toBe(false); + expect(fake.sent).toContainEqual( + expect.objectContaining({ message_id: 'msg_1', type: 'message.ack' }), + ); + expect(fake.sent).toContainEqual( + expect.objectContaining({ message_id: 'msg_2', type: 'message.ack' }), + ); + }); + it('patches an asynchronous category update without requiring a revision change', async () => { const fake = createFakeConnection(); const deps = createDeps({ connection: fake.connection }); From 7fc71379b0a2e5e71399700f828c3f88022d0c75 Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Fri, 21 Aug 2026 15:18:53 +0800 Subject: [PATCH 3/4] fix(realtime): settle the turn when a tool suppresses the follow-up response.done treated "a tool ran in this response" and "this response needs a follow-up" as the same condition and kept looping either way. A tool that ends the conversation asks for no follow-up (send_tool_result(..., respond=False)), so that response.done is the turn's actual last event: continuing past it left continuous mode never reporting turn_completed() (the call never hangs up) and push-to-talk reading past the end of the stream and reporting a spurious transport failure instead of settling cleanly. Only continue when a follow-up was actually requested and not suppressed; every other case -- including a suppressed one -- now falls through to the normal settlement path. Reported by the fennoai review bot on PR #343. Co-Authored-By: Claude Sonnet 5 --- .../external/realtime/qwen_audio.py | 48 ++++++++++-------- .../external/realtime/test_qwen_audio.py | 49 +++++++++++++++++++ 2 files changed, 77 insertions(+), 20 deletions(-) diff --git a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py index 8d509170..af410027 100644 --- a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py +++ b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py @@ -288,14 +288,18 @@ async def _pump_single_turn(self, observer: Observer) -> None: return await observer.tool_requested(**requested) elif kind == "response.done": - # Not the turn's end if a tool ran: the next response is the one that speaks. + # Not the turn's end if a tool asked for a follow-up: the next response + # is the one that speaks. A tool that ended the conversation instead + # (respond=False, _followup_suppressed) has no follow-up coming -- that + # must fall through to the normal settlement below, not skip it, or the + # turn never ends and the next recv() reads past the finished stream. self._open_responses -= 1 - if self._followup_requested or self._followup_suppressed: - if self._followup_requested and not self._followup_suppressed: - await self._send({"type": "response.create"}) - self._open_responses += 1 - self._followup_requested = False - self._followup_suppressed = False + wants_followup = self._followup_requested and not self._followup_suppressed + self._followup_requested = False + self._followup_suppressed = False + if wants_followup: + await self._send({"type": "response.create"}) + self._open_responses += 1 continue if self._open_responses <= 0: return @@ -357,19 +361,23 @@ async def _pump_continuous(self, observer: Observer) -> None: elif kind == "response.done": if self._open_responses > 0: self._open_responses -= 1 - if self._followup_requested or self._followup_suppressed: - # One or more tool calls happened inside the response that just - # finished; ask for the follow-up only now that the vendor - # considers it done (see send_tool_result). That follow-up, not - # this response, is what actually finishes the turn -- settling - # here would let a caller like end_conversation hang up before the - # model's actual reply to it has even started. Same pattern as - # _pump_single_turn's push-to-talk handling below. - if self._followup_requested and not self._followup_suppressed: - await self._send({"type": "response.create"}) - self._open_responses += 1 - self._followup_requested = False - self._followup_suppressed = False + # A tool call inside the response that just finished may have asked for + # a follow-up; ask for it only now that the vendor considers this + # response done (see send_tool_result). That follow-up, not this + # response, is what actually finishes the turn -- settling here would + # let a caller like end_conversation hang up before the model's actual + # reply to it has even started. Same pattern as _pump_single_turn's + # push-to-talk handling below. A tool that ended the conversation + # instead (respond=False, _followup_suppressed) has no follow-up + # coming, so it must NOT take this branch -- it needs the normal + # settlement below to actually report turn_completed and let the + # caller close out the call. + wants_followup = self._followup_requested and not self._followup_suppressed + self._followup_requested = False + self._followup_suppressed = False + if wants_followup: + await self._send({"type": "response.create"}) + self._open_responses += 1 continue self._responding = False # The bytes just sent still take this long to actually play out on the diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py index 6baa60b9..3ac61f8a 100644 --- a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py +++ b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py @@ -380,6 +380,55 @@ async def scenario() -> None: asyncio.run(scenario()) +def test_a_suppressed_followup_settles_the_turn_instead_of_reading_past_it() -> None: + """A tool that ends the conversation asks for no follow-up + (send_tool_result(..., respond=False)). Its response.done is the turn's actual + last event and must settle the pump immediately, not be treated the same as the + "a follow-up is coming, keep waiting" case above -- doing that leaves nothing to + stop the loop, so the next recv() reads past the end of the stream. + """ + + async def scenario() -> None: + transport = FakeTransport(_event("response.done")) + session = QwenAudioSession(transport, CONFIG, PUSH_TO_TALK) + observer = RecordingObserver() + await session.finish_input() + await session.send_tool_result("call-1", "{}", respond=False) + + await session.pump(observer) + + assert observer.calls == [] + + asyncio.run(scenario()) + + +def test_continuous_suppressed_followup_reports_turn_completed() -> None: + """Continuous-mode counterpart: a tool ending the conversation must still report + turn_completed() so the caller (agent.py's _finish_reply/deliver_session_end) + actually closes out the call, instead of being swallowed by the "a follow-up is + coming" branch and leaving the call open. + """ + + async def scenario() -> None: + transport = FakeTransport( + _event("response.created"), + _event("response.done"), + _event("error", error={"message": "stream ended"}), + ) + session = QwenAudioSession(transport, CONFIG, CONTINUOUS) + observer = RecordingObserver() + await session.send_tool_result("call-1", "{}", respond=False) + + await session.pump(observer) + + assert observer.calls == [ + ("turn_completed", None), + ("failed", "stream ended"), + ] + + asyncio.run(scenario()) + + def test_two_tool_calls_in_one_response_only_ask_for_a_single_followup() -> None: """A batch turn can call several tools before the response that requested them is done (e.g. a batch create/delete voice command). Each call must not eagerly ask From 0ad1919edf977d20b4782015d72a47a9d66ab012 Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Fri, 21 Aug 2026 15:37:55 +0800 Subject: [PATCH 4/4] fix(assistant): stop queueCommandResult from crashing on a rejected write commandResultChain used .then(onFulfilled, onRejected) to keep applying queued voice.command.result writes after one fails. If the rejected branch actually fires, that rejection sits unhandled until some later queueCommandResult() call chains onto it -- and if there isn't one (e.g. it was the last command result of the call), Node/Hermes treats it as an unhandled rejection and crashes the process outright. Reproduced with a state-subscriber listener that throws during markScheduleDataChanged()'s notification. Switch to .then(onFulfilled).catch(() => {}), the same idiom already used by chainPlayback() in the same file: the .catch() is attached in the same statement, so the rejection is neutralized immediately instead of waiting on a future call that may never come. Co-Authored-By: Claude Sonnet 5 --- .../AssistantContinuousConversationService.ts | 12 +++-- .../AssistantConversationService.ts | 12 +++-- ...stantContinuousConversationService.test.ts | 47 ++++++++++++++++++ .../AssistantConversationService.test.ts | 48 +++++++++++++++++++ 4 files changed, 111 insertions(+), 8 deletions(-) diff --git a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts index 4e228f97..eb923cf7 100644 --- a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts +++ b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts @@ -464,11 +464,15 @@ export class AssistantContinuousConversationService implements AssistantApplicat } } + /** 跟 chainPlayback 一个模式：.catch(() => {}) 必须紧跟在同一条语句里同步接上—— + * 迟一拍再接（比如靠下一次 queueCommandResult 调用里的第二个 then 参数兜底） + * 这段窗口期这个被拒绝的 promise 没有任何 handler，Node 的 unhandled rejection + * 检测在下一次调用到达前就已经判定“没人接”，直接把整个进程带崩——不是理论 + * 风险，用一个会抛的 state 订阅者复现过。 */ private queueCommandResult(command: AppliedCommand, messageId: string): void { - this.commandResultChain = this.commandResultChain.then( - () => this.applyCommandResultLocally(command, messageId), - () => this.applyCommandResultLocally(command, messageId), - ); + this.commandResultChain = this.commandResultChain + .then(() => this.applyCommandResultLocally(command, messageId)) + .catch(() => {}); } private async applyCommandResultLocally( diff --git a/frontend/src/features/assistant/application/AssistantConversationService.ts b/frontend/src/features/assistant/application/AssistantConversationService.ts index 72acb14f..bb6073b4 100644 --- a/frontend/src/features/assistant/application/AssistantConversationService.ts +++ b/frontend/src/features/assistant/application/AssistantConversationService.ts @@ -281,11 +281,15 @@ export class AssistantConversationService implements AssistantApplicationPort { } } + /** .catch(() => {}) 必须紧跟在同一条语句里同步接上——迟一拍再接（比如靠下一次 + * queueCommandResult 调用里的第二个 then 参数兜底）这段窗口期这个被拒绝的 + * promise 没有任何 handler，Node 的 unhandled rejection 检测在下一次调用到达前 + * 就已经判定"没人接"，直接把整个进程带崩——不是理论风险，用一个会抛的 state + * 订阅者复现过。 */ private queueCommandResult(command: AppliedCommand, messageId: string): void { - this.commandResultChain = this.commandResultChain.then( - () => this.applyCommandResultLocally(command, messageId), - () => this.applyCommandResultLocally(command, messageId), - ); + this.commandResultChain = this.commandResultChain + .then(() => this.applyCommandResultLocally(command, messageId)) + .catch(() => {}); } private async applyCommandResultLocally( diff --git a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts index 4c3d0908..c38f8b66 100644 --- a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts +++ b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts @@ -453,6 +453,53 @@ describe('AssistantContinuousConversationService', () => { disposeService(service); }); + it('keeps applying queued command results even if a subscriber listener throws', async () => { + // markScheduleDataChanged() synchronously notifies every subscriber; a listener + // that throws (e.g. a buggy re-render) rejects applyCommandResultLocally()'s + // promise. queueCommandResult() must neutralize that with .catch(() => {}) + // chained in the same statement (same idiom as chainPlayback) -- deferring the + // catch to a later call leaves the rejection unhandled for a few microtask + // ticks, which crashes the process outright (reproduced while writing this + // test, before adding the immediate .catch()). + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = createService(deps); + await startListening(fake, service); + + // handleMessage's own setState() notifies once synchronously before + // applyCommandResultLocally's markScheduleDataChanged() notifies again + // asynchronously; only the second one is the one queueCommandResult's chain + // needs to survive. + let notifyCount = 0; + const unsubscribe = service.subscribe(() => { + notifyCount += 1; + if (notifyCount === 2) { + throw new Error('listener boom'); + } + }); + fake.emitMessage({ + conversation_id: 'conv_001', + message_id: 'msg_1', + payload: { operation: 'create_schedule', schedule: { id: 'a' }, status: 'applied' }, + type: 'voice.command.result', + } as AssistantServerMessage); + await flushAsync(); + unsubscribe(); + + fake.emitMessage({ + conversation_id: 'conv_001', + message_id: 'msg_2', + payload: { operation: 'create_schedule', schedule: { id: 'b' }, status: 'applied' }, + type: 'voice.command.result', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getLastAppliedCommand()).toEqual( + expect.objectContaining({ operation: 'create_schedule', schedule: { id: 'b' } }), + ); + disposeService(service); + }); + it('serializes local writes so a batch of back-to-back command results cannot race', async () => { // Regression test for a batch create/delete race: the model can call several // tools inside one turn, so voice.command.result messages can arrive only diff --git a/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts index 30c255d1..0124a296 100644 --- a/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts +++ b/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts @@ -292,6 +292,54 @@ describe('AssistantConversationService', () => { expect(service.getLastAppliedCommand()).toBeNull(); }); + it('keeps applying queued command results even if a subscriber listener throws', async () => { + // markScheduleDataChanged() synchronously notifies every subscriber; a listener + // that throws (e.g. a buggy re-render) rejects applyCommandResultLocally()'s + // promise. queueCommandResult() must neutralize that with .catch(() => {}) + // chained in the same statement (same idiom as + // AssistantContinuousConversationService/chainPlayback) -- deferring the catch + // leaves the rejection unhandled for a few microtask ticks, which crashes the + // process outright (reproduced while writing this test, before adding the + // immediate .catch()). + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); + + const turn = service.startTurn(); + await completeStreamStart(fake, turn); + // handleMessage's own setState() notifies once synchronously before + // applyCommandResultLocally's markScheduleDataChanged() notifies again + // asynchronously; only the second one is the one queueCommandResult's chain + // needs to survive. + let notifyCount = 0; + const unsubscribe = service.subscribe(() => { + notifyCount += 1; + if (notifyCount === 2) { + throw new Error('listener boom'); + } + }); + fake.emitMessage({ + conversation_id: 'conv_001', + message_id: 'msg_1', + payload: { operation: 'create_schedule', schedule: { id: 'a' }, status: 'applied' }, + type: 'voice.command.result', + } as AssistantServerMessage); + await flushAsync(); + unsubscribe(); + + fake.emitMessage({ + conversation_id: 'conv_001', + message_id: 'msg_2', + payload: { operation: 'create_schedule', schedule: { id: 'b' }, status: 'applied' }, + type: 'voice.command.result', + } as AssistantServerMessage); + await flushAsync(); + + expect(service.getLastAppliedCommand()).toEqual( + expect.objectContaining({ operation: 'create_schedule', schedule: { id: 'b' } }), + ); + }); + it('serializes local writes so a batch of back-to-back command results cannot race', async () => { // Regression test: a single utterance can make the model call several tools in // one turn, so voice.command.result messages can arrive only milliseconds