From aa4cb0ed884c60c31ae97c469014f5c3f6c72f66 Mon Sep 17 00:00:00 2001 From: LUPENGHAN Date: Mon, 24 Aug 2026 21:57:00 +0800 Subject: [PATCH] =?UTF-8?q?fix(assistant):=20=E6=8C=89=E4=BD=8F=E8=AF=B4?= =?UTF-8?q?=E8=AF=9D=E6=97=B6=E4=B8=A2=E5=BC=83=E4=B8=8A=E4=B8=80=E8=BD=AE?= =?UTF-8?q?=E8=BF=9F=E5=88=B0=E7=9A=84=E5=9B=9E=E5=A4=8D=EF=BC=8C=E9=81=BF?= =?UTF-8?q?=E5=85=8D=E6=97=A7=E6=B0=94=E6=B3=A1=E6=8C=A1=E4=BD=8F=E8=AF=AD?= =?UTF-8?q?=E9=9F=B3=E6=9D=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 每一轮按住说话生成自增的 request_id 随 voice.stream.start 发出，服务端会把它 回显到该轮的 voice.dialogue.reply 上；handleMessage 据此认轮次，对不上的直接 丢弃。上一轮被 interrupt 后迟到的回复不会再把已经点掉的气泡弹回来——气泡是一层 全屏 Pressable，会挡住语音条，用户得多点一次才能说话。 只 gate voice.dialogue.reply 这一条：错误信封是 startTurn() 唯一的解套途径， voice.stream.started 是 streamId 的唯一来源，voice.command.result 代表服务端 已提交的写入，按轮次丢弃它们的代价都远大于显示一条过期气泡。request_id 为 null/undefined 时一律放行——后端 model_dump() 把缺省值序列化成 null 而非省略 字段，只判 undefined 会把正常回复误吞。 Co-Authored-By: Claude --- .../AssistantConversationService.ts | 24 ++++ .../AssistantConversationService.test.ts | 111 ++++++++++++++++++ 2 files changed, 135 insertions(+) diff --git a/frontend/src/features/assistant/application/AssistantConversationService.ts b/frontend/src/features/assistant/application/AssistantConversationService.ts index cbbe9a08..f50edc9b 100644 --- a/frontend/src/features/assistant/application/AssistantConversationService.ts +++ b/frontend/src/features/assistant/application/AssistantConversationService.ts @@ -20,6 +20,8 @@ const EMPTY_MESSAGES: readonly VoiceChatMessage[] = []; // 共享连接的握手超时（AuthenticatedWebSocketClient 内部固定 5s）已经不归这里管； // 这个只是给定位单独留的预算，拿不到就不带，不能让 connect() 本身被定位拖住。 const LOCATION_TIMEOUT_MS = 2000; +/** 每个按住说话的 turn 一个自增 id，随 voice.stream.start 发出、由服务端回显。 */ +let turnRequestIdCounter = 0; /** * 一次"按住说话"编排的真实实现，按 AGENTS.md 第 6 节的时序把 transport / capture / @@ -44,6 +46,8 @@ export class AssistantConversationService implements AssistantApplicationPort { private unsubscribeConnection: (() => void) | null = null; /** voice.dialogue.reply 的流式文字，展示层拿来当气泡内容；新一轮开始时清空。 */ private replyText: string | null = null; + /** 当前 turn 的 request_id；voice.dialogue.reply 拿它认轮次，对不上的是上一轮迟到的。 */ + private turnRequestId: string | null = null; /** 当前这一帧麦克风音量（dBFS），给波形展示；不在录音时是 null。 */ private soundLevel: number | null = null; // 展示层的 onPressIn/onPressOut 不等待彼此:快速按放会让 endTurn() 在 @@ -132,7 +136,12 @@ export class AssistantConversationService implements AssistantApplicationPort { this.streamStartedWaiter = resolve; this.streamStartRejecter = reject; }); + // 紧挨着 send 换 id：connect() 和权限框都可能停留好几秒，提前换掉的话这段窗口 + // 里线上跑的还是上一轮的流，它自己的 reply 会被当成"别人的"误丢。 + const turnRequestId = `ptt-turn-${++turnRequestIdCounter}`; + this.turnRequestId = turnRequestId; connection.send({ + request_id: turnRequestId, payload: { audio_format: AUDIO_FORMAT, channels: CHANNELS, @@ -267,6 +276,12 @@ export class AssistantConversationService implements AssistantApplicationPort { }); return; case 'voice.dialogue.reply': + // 上一轮被打断时它的 reply 可能晚到，把已经点掉的旧气泡重新弹出来——气泡 + // 是全屏点击层，会挡住语音条，用户得再点一次才能说话。只认当前这一轮的 + // request_id；服务端把 voice.stream.start 上带的 id 回显在这条消息上。 + if (!this.isCurrentTurnReply(message.request_id)) { + return; + } this.replyText = message.payload.speech_text; this.notifyListeners(); return; @@ -290,6 +305,15 @@ export class AssistantConversationService implements AssistantApplicationPort { } } + /** 拿不到判断依据时一律放行：后端把缺省的 request_id 序列化成 null 而不是省略字段， + * 所以 null 和 undefined 都要当作"不知道是哪一轮"，不能误伤。 */ + private isCurrentTurnReply(requestId: string | null | undefined): boolean { + if (this.turnRequestId === null || requestId === null || requestId === undefined) { + return true; + } + return requestId === this.turnRequestId; + } + /** .catch(() => {}) 必须紧跟在同一条语句里同步接上——迟一拍再接（比如靠下一次 * queueCommandResult 调用里的第二个 then 参数兜底）这段窗口期这个被拒绝的 * promise 没有任何 handler，Node 的 unhandled rejection 检测在下一次调用到达前 diff --git a/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts b/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts index 0124a296..ec3c5c03 100644 --- a/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts +++ b/frontend/tests/unit/features/assistant/application/AssistantConversationService.test.ts @@ -610,4 +610,115 @@ describe('AssistantConversationService', () => { expect(fake.unsubscribeCalls).toEqual({ audio: 1, close: 1, message: 1 }); expect(fake.closeCalls.count).toBe(1); }); + + it('tags every voice.stream.start with a fresh per-turn request_id', async () => { + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); + + await completeStreamStart(fake, service.startTurn()); + await completeStreamStart(fake, service.startTurn()); + + const starts = fake.sent.filter((message) => message.type === 'voice.stream.start'); + expect(starts).toHaveLength(2); + expect(starts[0].request_id).toBeTruthy(); + expect(starts[1].request_id).toBeTruthy(); + expect(starts[1].request_id).not.toBe(starts[0].request_id); + }); + + it('drops a stale reply from the previous turn so it cannot repop the bubble', async () => { + // 上一轮被新一轮 voice.stream.start 打断时，execute 的 reply/tts 事件可能晚到； + // 不按 request_id 过滤的话，气泡被点掉后会被这条迟到回复重新弹回来，把输入条 + // 整个挡住（AssistantVoiceOverlay 在有 replyText 时铺了一层全屏点击层）。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); + + // turn 1：回复到了，用户点掉气泡。计数器是模块级的，用例之间会累加，所以 + // 轮次 id 一律从实际发出的 voice.stream.start 上取，不写死字面量。 + await completeStreamStart(fake, service.startTurn()); + const turn1Id = fake.sent.filter((message) => message.type === 'voice.stream.start')[0] + .request_id; + fake.emitMessage({ + conversation_id: 'conv_001', + request_id: turn1Id, + payload: { done: true, reply_id: 'reply_1', speech_text: '第一条回复' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + await flushAsync(); + await service.dismissReply(); + expect(service.getReplyText()).toBeNull(); + + // turn 2：新的 stream.start 已经带上不同的 request_id（这里故意不先把新流 + // started 发下去，复现旧事件在新一轮启动窗口里晚到的情形）。 + const nextTurn = service.startTurn(); + await flushAsync(); + fake.emitMessage({ + conversation_id: 'conv_001', + request_id: turn1Id, + payload: { done: true, reply_id: 'reply_1', speech_text: '第一条回复' }, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + await flushAsync(); + expect(service.getReplyText()).toBeNull(); + + // 新 turn 自己的回复到了才显示。 + const starts = fake.sent.filter((message) => message.type === 'voice.stream.start'); + const turn2Id = starts[1].request_id; + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_2', speech_text: '第二条回复' }, + request_id: turn2Id, + type: 'voice.dialogue.reply', + } as AssistantServerMessage); + await flushAsync(); + expect(service.getReplyText()).toBe('第二条回复'); + + await completeStreamStart(fake, nextTurn); + }); + + it('still shows a reply whose request_id is null', async () => { + // 后端 model_dump() 把缺省的 request_id 序列化成 null（不是省掉字段），所以 + // null 必须当作"不知道是哪一轮"放行，否则回复永远显示不出来。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); + + await completeStreamStart(fake, service.startTurn()); + fake.emitMessage({ + conversation_id: 'conv_001', + payload: { done: true, reply_id: 'reply_1', speech_text: '没带轮次的回复' }, + request_id: null, + type: 'voice.dialogue.reply', + } as unknown as AssistantServerMessage); + await flushAsync(); + + expect(service.getReplyText()).toBe('没带轮次的回复'); + }); + + it('resolves startTurn on a transport error that carries a stale request_id', async () => { + // 错误信封也带 request_id，而且可能是上一条流的（后端 voice_stream.py 的 + // "A stream is already active" / "Audio frame is empty" 都会这样）。这条路径 + // 是 startTurn() 唯一的解套机会，按轮次丢掉它会让按住说话永久卡死。 + const fake = createFakeConnection(); + const deps = createDeps({ connection: fake.connection }); + const service = new AssistantConversationService({ accountId: 'acc_001' }, deps); + + await completeStreamStart(fake, service.startTurn()); + const staleId = fake.sent.filter((message) => message.type === 'voice.stream.start')[0] + .request_id; + const turn = service.startTurn(); + await flushAsync(); + fake.emitMessage({ + error: { code: 'X', message: 'A stream is already active for this session' }, + ok: false, + request_id: staleId, + } as AssistantServerMessage); + + await expect(turn).resolves.toBeUndefined(); + expect(service.getState()).toEqual({ + message: 'A stream is already active for this session', + phase: 'error', + }); + }); });