diff --git a/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py b/backend/src/timeflow/infrastructure/external/realtime/qwen_audio.py index 4f13e0ba..af410027 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: @@ -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 diff --git a/backend/tests/infrastructure/external/realtime/test_qwen_audio.py b/backend/tests/infrastructure/external/realtime/test_qwen_audio.py index 1569d828..3ac61f8a 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()) @@ -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.""" diff --git a/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts b/frontend/src/features/assistant/application/AssistantContinuousConversationService.ts index 86a775c2..eb923cf7 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,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, diff --git a/frontend/src/features/assistant/application/AssistantConversationService.ts b/frontend/src/features/assistant/application/AssistantConversationService.ts index 8b6383cf..bb6073b4 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,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, diff --git a/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantContinuousConversationService.test.ts index 9a1d105f..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,123 @@ 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 + // 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..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,124 @@ 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 + // 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 });