fix(realtime): 批量创建/删除日程时连续对话被强制挂断，且本地日历静默丢数据

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,25 +196,31 @@ 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(
             {
                 "type": "conversation.item.create",
                 "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:
@@ -269,8 +288,19 @@ 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
+                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
             elif kind == "error":
@@ -330,13 +360,24 @@ 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
-                    # 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
+                # 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

backend/tests/infrastructure/external/realtime/test_qwen_audio.py

@@ -327,23 +327,33 @@ 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",
             "call_id": "call_1",
             "output": '{"count":2}',
         }
 
+        await session.pump(RecordingObserver())
+
+        assert transport.types() == ["conversation.item.create", "response.create"]
+
     asyncio.run(scenario())
 
 
@@ -370,6 +380,90 @@ 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
+    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(
@@ -1070,6 +1164,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."""
 

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<string, ScheduleCategory>();
+  /** 串起每条 voice.command.result 的本地落库；一次连续通话里模型可能连续触发多个
+   * 工具调用（批量新建/删除），command.result 消息前后脚到达时若不排队，会有两个
+   * applyCommandResultLocally() 同时各开一个 withExclusiveTransactionAsync（各自
+   * 新建一条原生连接），互相踩 "database is locked" 甚至留下损坏的原生连接状态——
+   * 跟 playbackChain 一个模式：不管上一条成功与否都排到下一条前面。 */
+  private commandResultChain: Promise<void> = 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,17 @@ 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))
+      .catch(() => {});
+  }
+
   private async applyCommandResultLocally(
     command: AppliedCommand,
     messageId: string,

frontend/src/features/assistant/application/AssistantConversationService.ts

@@ -47,6 +47,10 @@ export class AssistantConversationService implements AssistantApplicationPort {
   private pendingStartTurn: Promise<void> | null = null;
   /** Category events can arrive before the command result creates their local row. */
   private readonly pendingCategoryUpdates = new Map<string, ScheduleCategory>();
+  /** 串起每条 voice.command.result 的本地落库，见 AssistantContinuousConversationService
+   * 里同名字段的注释——一次按住说话也可能在一句话里触发多个工具调用（批量新建/
+   * 删除），不排队会让两个 applyCommandResultLocally() 并发抢同一个 SQLite 连接。 */
+  private commandResultChain: Promise<void> = 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,17 @@ 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))
+      .catch(() => {});
+  }
+
   private async applyCommandResultLocally(
     command: AppliedCommand,
     messageId: string,