fix(voice): preserve pending message intent across reconnects

talkingq-url/docs/voice-target-recovery.md

@@ -1,14 +1,22 @@
 # 留言状态与后续 AI 对话恢复
 
-留言目标只属于一次录音和一个 WebSocket 连接。目标设置后默认空闲有效期为
+留言分为待录音意图和已关联录音两个阶段。刷卡或短按先创建待录音意图，不绑定
-120 秒，可用 `VOICE_TARGET_TIMEOUT_SECONDS` 调整；录音开始及有效留言音频会刷新
+连接表中的旧 WebSocket；录音开始包再将意图关联到唯一的连接和 session。
-空闲期限。每次访问均检查过期状态，后台会定期回收过期目标及缓存。
+新连接接入、尚未录音便断开以及旧连接延迟退出，均不清除待录音意图；
+已关联录音的目标仍在其连接断开或被替换时清理。
 
-开始包统一返回一次 `START_ACK`。存在留言目标时，根据首个音频包选择流程：
+目标设置后默认空闲有效期为 120 秒，可用 `VOICE_TARGET_TIMEOUT_SECONDS` 调整；
+录音开始及有效留言音频会刷新空闲期限，单纯连接或重连不会延长有效期。
+每次访问均检查过期状态，后台会定期回收过期目标及缓存。
+
+开始包统一返回一次 `START_ACK`，先登记会话，再根据首个音频包选择流程：
 类型 `0` 清除待处理留言目标并完成 AI 初始化，原始首包随后正常送入 ASR；
 类型 `4` 写入本次留言的独立缓存。录音开始传输音频后不允许切换类型。
+没有有效目标的留言音频不会启动 AI，结束时返回 `END_ACK` 和既有留言保存失败
+提示音；旧录音延迟结束时不向正在进行的新会话插入该失败提示。
+没有音频的普通 START/END 会话仍走原有静音识别响应。
 
-结束、取消、异常、空音频、超时和连接断开均会释放对应目标及音频缓存。
+已关联录音的目标及音频缓存，在结束、取消、异常、空音频、超时或连接断开时释放。
 结束时先消费本次目标，再处理上传和通知，避免旧处理完成时清除新设置的目标。
 旧连接断开只清理该连接的会话和任务，不影响替换它的新连接。
 音频缓存键按发送设备及本次目标独立生成，多个发送方对同一接收方留言时互不覆盖。
@@ -34,3 +42,23 @@ AI 中断后留言验证了启动、收音及取消，尚未串联验证这两
 状态测试：`.venv/bin/python -m pytest tests/test_voice_target_lifecycle.py -q`
 
 完整结果：[本地测试汇总](../output/voice-target-recovery-20261028/summary.json)。
+
+## 待录音目标重连修复（2026-10-30）
+
+设备先刷卡、后建立录音 WebSocket 时，旧实现将新意图绑定到旧连接，随后以
+`connection_changed` 清除有效意图。本次取消创建意图和连接接入时的提前绑定，
+保留已关联录音的连接与 session 隔离，并将 AI 初始化推迟到首个普通音频包
+（纯静音会话则在结束包）以避免无目标留言误走 AI。
+
+- 新增用例先在旧代码复现：8 种刷卡后换连接/旧连接退出组合失败，
+  另有 1 项无目标留言误走 AI 的测试失败。
+- 修改后留言状态测试 49/49，完整后端回归 264/264。
+- 覆盖家长及设备留言、旧连接在新连接前/后及录音开始后退出、重连不刷新
+  待录音有效期、旧包隔离、静音、无目标留言失败以及取消/断线/超时后的再次留言。
+- 两项本地集成用例串联实际 MQTT 005/011 业务处理函数、`/ws` 路由和
+  留言处理函数，核对完整音频、成功提示、家长会话保存或设备归档/待投递接口参数
+  及缓存清理；鉴权、MQTT 发布、存储、数据库写入及通知均使用隔离替身。
+- 本轮未连接真实 MQTT Broker、未调用真实 ASR/LLM/TTS、未验证物理设备，
+  未访问或部署生产服务。
+
+测试证据：`output/voice-target-reconnect-20261030/pytest.xml`。

talkingq-url/handlers/websocket_message_handler.py

@@ -176,15 +176,13 @@ async def handle_binary_message(websocket: WebSocket, device_id: str, serial_num
             return current_active_session, first_audio_received_time
         session.start_received = True
         session.voice_target = await device_target_cache.claim_target(device_id, websocket, session_id)
-        if session.voice_target:
+        # START does not identify AI versus message audio. Decide on the first
-            # Acknowledge START now; the first audio packet selects AI (0) or
+        # audio packet, even without a target, so an invalid message cannot
-            # message (4). A pending message must not swallow an AI recording.
+        # start ASR and later report an unrelated recognition failure.
-            await websocket.send_text("START_ACK")
+        current_active_session = await handle_start_packet(
-            session.start_ack_sent = True
+            device_id, session_id, session_key, session, current_active_session)
-            return session_key, None
+        await websocket.send_text("START_ACK")
-        current_active_session = await start_ai_recording(
+        session.start_ack_sent = True
-            websocket, device_id, session_id, session_key, session, current_active_session, sample_rate
-        )
         return current_active_session, None
 
     if not session.start_received or session.recording_finished:
@@ -209,26 +207,32 @@ async def handle_binary_message(websocket: WebSocket, device_id: str, serial_num
             session_logger.info(device_id, session_id, f"收到第一个音频数据包时间: {first_audio_received_time}")
     elif packet_type == 4:
         target = session.voice_target
-        if target and not session.ai_started:
+        if not session.ai_started:
-            if await handle_target_audio_packet(device_id, audio_data, target, websocket, session_id):
+            session.message_audio_received = True
-                session.message_audio_received = True
+            if target:
+                await handle_target_audio_packet(device_id, audio_data, target, websocket, session_id)
+            else:
+                session_logger.warning(device_id, session_id, "留言录音没有对应目标，将返回留言失败提示")
         else:
-            session_logger.warning(device_id, session_id, "忽略没有对应留言目标的留言音频包")
+            session_logger.warning(device_id, session_id, "忽略普通对话中混入的留言音频")
         return current_active_session, first_audio_received_time
     elif packet_type == 3:
-        if session.voice_target:
+        if not session.ai_started:
-            await device_target_cache.remove_target(device_id, expected=session.voice_target, reason="interrupted")
+            if session.voice_target:
+                await device_target_cache.remove_target(device_id, expected=session.voice_target, reason="interrupted")
             session.voice_target = None
             session.recording_finished = True
             await websocket.send_text("INTERRUPT_ACK")
             await websocket.send_text("INTERRUPT_PROCESSED")
+            await interrupt_handler.remove_session(session_key)
             return None if current_active_session == session_key else current_active_session, None
         current_active_session = await handle_interrupt_packet(
             device_id, session_id, session_key, session, current_active_session
         )
     elif packet_type == 2:
-        if session.voice_target:
+        if session.voice_target or session.message_audio_received:
-            target = await device_target_cache.take_target(device_id, session.voice_target, websocket, session_id)
+            target = (await device_target_cache.take_target(device_id, session.voice_target, websocket, session_id)
+                      if session.voice_target else None)
             session.voice_target = None
             session.recording_finished = True
             try:
@@ -240,10 +244,19 @@ async def handle_binary_message(websocket: WebSocket, device_id: str, serial_num
                     else:
                         await process_cached_audio(device_id, target.target_device_id, serial_number,
                                                    audio_cache_key=target.audio_cache_key)
+                elif current_active_session is None or current_active_session == session_key:
+                    await websocket.send_text(
+                        f"PROMPT_SOUND_URL:http://{settings.server_host}:{settings.server_port}/assets/audio/save_audio_fail.mp3")
             finally:
                 if target:
                     await target_audio_cache.clear_audio_data(target.audio_cache_key)
+                await interrupt_handler.remove_session(session_key)
             return None if current_active_session == session_key else current_active_session, None
+        if not session.ai_started:
+            # Preserve the existing silent-recording response for START/END
+            # without audio while keeping unowned messages out of the AI path.
+            current_active_session = await start_ai_recording(
+                websocket, device_id, session_id, session_key, session, current_active_session, sample_rate)
         if current_active_session == session_key:
             current_active_session = None
 
@@ -254,9 +267,6 @@ async def handle_binary_message(websocket: WebSocket, device_id: str, serial_num
 
 
 async def start_ai_recording(websocket, device_id, session_id, session_key, session, current_active_session, sample_rate):
-    current_active_session = await handle_start_packet(
-        device_id, session_id, session_key, session, current_active_session
-    )
     await handle_websocket_data(websocket, session_key, session, 1, b"", sample_rate)
     session.ai_started = True
     return current_active_session

talkingq-url/services/device_target_cache.py

@@ -62,7 +62,6 @@ class DeviceTargetCache:
             self.targets[device_id] = VoiceTarget(
                 kind=kind, target_device_id=target_device_id, uuid=uuid,
                 audio_cache_key=f"voice:{device_id}:{uuid4().hex}",
-                websocket=await connection_manager.get_connection(device_id),
                 expires_at=self.clock() + self.timeout_seconds,
             )
             session_logger.info(device_id, "target", f"留言目标已设置: kind={kind}, timeout={self.timeout_seconds}s")
@@ -94,17 +93,15 @@ class DeviceTargetCache:
             return target
 
     async def bind_connection(self, device_id, websocket):
-        """Adopt commands received before connect; discard older socket intents."""
+        """Keep pending commands; discard recordings owned by another socket."""
         async with self.lock:
             if await connection_manager.get_connection(device_id) is not websocket:
                 return
             target = await self._get_locked(device_id)
             if target is None:
                 return
-            if target.websocket is not None and target.websocket is not websocket:
+            if target.session_id is not None and target.websocket is not websocket:
                 await self._remove_locked(device_id, target, "connection_changed")
-            else:
-                target.websocket = websocket
 
     async def append_audio(self, device_id, target, websocket, session_id, audio):
         async with self.lock:

talkingq-url/tests/test_voice_target_lifecycle.py

@@ -20,6 +20,7 @@ from services.target_audio_cache import TargetAudioCache
 
 DEVICE = 'TalkingQ_local_voice'
 PROCESS_PARENT_MESSAGE = messages.process_parent_leave_message
+PROCESS_DEVICE_MESSAGE = messages.process_cached_audio
 
 
 def packet(sid, kind, audio=b''):
@@ -79,6 +80,223 @@ async def set_target(state, kind):
     return await state.targets.get_voice_target(DEVICE)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize('kind', ['parent', 'device'])
+@pytest.mark.parametrize('old_disconnect', ['before_connect', 'after_connect', 'after_start', 'none'])
+async def test_fresh_command_survives_socket_replacement_and_completes_message(state, kind, old_disconnect):
+    """Reproduce 03111: card command, then a new recording socket nine seconds later."""
+    old = state.ws
+    await state.connections.add_connection(DEVICE, old)
+    target = await set_target(state, kind)
+    deadline = target.expires_at
+    state.now[0] += 9
+    if old_disconnect == 'before_connect':
+        await state.connections.remove_connection(DEVICE, old)
+        await cleanup.cleanup_device_sessions(DEVICE, old)
+    new = SimpleNamespace(send_text=AsyncMock())
+    await state.connections.add_connection(DEVICE, new)
+    await state.targets.bind_connection(DEVICE, new)
+    if old_disconnect == 'after_connect':
+        await cleanup.cleanup_device_sessions(DEVICE, old)
+    assert await state.targets.get_voice_target(DEVICE) is target
+    assert target.websocket is None
+    assert target.expires_at == deadline
+    await send(state, 'fresh-message', 1, ws=new)
+    if old_disconnect == 'after_start':
+        await cleanup.cleanup_device_sessions(DEVICE, old)
+    await send(state, 'old-late', 1, ws=old)
+    await send(state, 'fresh-message', 4, b'first-', ws=new)
+    await send(state, 'fresh-message', 4, b'tail', ws=new)
+    processor = messages.process_parent_leave_message if kind == 'parent' else messages.process_cached_audio
+    archived = []
+
+    async def process(*args, **kwargs):
+        archived.append(await state.buffers.get_audio_data(target.audio_cache_key))
+
+    processor.side_effect = process
+    await send(state, 'fresh-message', 2, ws=new)
+    processor.assert_awaited_once()
+    assert archived == [b'first-tail']
+    assert state.started == []
+    assert [c.args[0] for c in new.send_text.call_args_list] == ['START_ACK', 'END_ACK']
+    assert await state.targets.get_voice_target(DEVICE) is None
+    assert await state.buffers.get_audio_data(target.audio_cache_key) is None
+
+
+@pytest.mark.asyncio
+async def test_unowned_message_reports_failure_without_starting_ai_or_consuming_new_intent(state):
+    await send(state, 'unowned', 1)
+    await send(state, 'unowned', 4, b'unowned-audio')
+    fresh = await set_target(state, 'parent')
+    await send(state, 'unowned', 0, b'wrong-mode')
+    await send(state, 'unowned', 2)
+    assert state.started == []
+    assert state.ended == []
+    assert await state.targets.get_voice_target(DEVICE) is fresh
+    replies = [c.args[0] for c in state.ws.send_text.call_args_list]
+    assert replies[:2] == ['START_ACK', 'END_ACK']
+    assert replies[2].endswith('/assets/audio/save_audio_fail.mp3')
+    messages.process_parent_leave_message.assert_not_awaited()
+    messages.process_cached_audio.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_silent_ai_still_starts_and_finishes_once(state):
+    await send(state, 'silent-ai', 1)
+    await send(state, 'silent-ai', 2)
+    assert state.started == ['silent-ai']
+    assert state.ended == [('silent-ai', b'')]
+    assert [c.args[0] for c in state.ws.send_text.call_args_list] == ['START_ACK', 'END_ACK']
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('kind', ['parent', 'device'])
+@pytest.mark.parametrize('previous', ['message_cancelled', 'message_disconnected', 'message_expired', 'ai_disconnected'])
+async def test_incomplete_recording_then_fresh_message_saves_only_new_audio(state, kind, previous):
+    await state.connections.add_connection(DEVICE, state.ws)
+    old_target = None
+    if previous.startswith('message'):
+        old_target = await set_target(state, kind)
+    await send(state, 'incomplete', 1)
+    await send(state, 'incomplete', 0 if previous == 'ai_disconnected' else 4, b'old-audio')
+    if previous == 'message_cancelled':
+        await send(state, 'incomplete', 3)
+    elif previous == 'message_expired':
+        state.now[0] += 10
+        assert await state.targets.get_voice_target(DEVICE) is None
+    else:
+        session = await state.sessions.get_session((DEVICE, 'incomplete'))
+        session.set_interrupted = AsyncMock()
+        await state.connections.remove_connection(DEVICE, state.ws)
+        await cleanup.cleanup_device_sessions(DEVICE, state.ws)
+    fresh = await set_target(state, kind)
+    new = SimpleNamespace(send_text=AsyncMock())
+    await state.connections.add_connection(DEVICE, new)
+    await state.targets.bind_connection(DEVICE, new)
+    await send(state, 'fresh', 1, ws=new)
+    await send(state, 'fresh', 4, b'new-audio', ws=new)
+    processor = messages.process_parent_leave_message if kind == 'parent' else messages.process_cached_audio
+    archived = []
+
+    async def process(*args, **kwargs):
+        archived.append(await state.buffers.get_audio_data(fresh.audio_cache_key))
+
+    processor.side_effect = process
+    await send(state, 'fresh', 2, ws=new)
+    assert archived == [b'new-audio']
+    processor.assert_awaited_once()
+    assert await state.targets.get_voice_target(DEVICE) is None
+    assert await state.buffers.get_audio_data(fresh.audio_cache_key) is None
+    if old_target:
+        assert await state.buffers.get_audio_data(old_target.audio_cache_key) is None
+    assert state.started == (['incomplete'] if previous == 'ai_disconnected' else [])
+
+
+@pytest.mark.asyncio
+async def test_reconnect_does_not_extend_pending_target_lifetime(state):
+    target = await set_target(state, 'parent')
+    for _ in range(2):
+        new = SimpleNamespace(send_text=AsyncMock())
+        state.now[0] += 4
+        await state.connections.add_connection(DEVICE, new)
+        await state.targets.bind_connection(DEVICE, new)
+        assert await state.targets.get_voice_target(DEVICE) is target
+    state.now[0] += 2
+    await state.targets.bind_connection(DEVICE, new)
+    assert await state.targets.get_voice_target(DEVICE) is None
+    await send(state, 'expired-message', 1, ws=new)
+    await send(state, 'expired-message', 4, b'late', ws=new)
+    await send(state, 'expired-message', 2, ws=new)
+    assert state.started == []
+    assert new.send_text.call_args.args[0].endswith('/assets/audio/save_audio_fail.mp3')
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('kind', ['parent', 'device'])
+async def test_mqtt_command_then_replacement_ws_reaches_real_message_processor(state, monkeypatch, kind):
+    """Use actual MQTT handlers, /ws and message processors; stub external I/O."""
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+    from api.websocket import router
+    from handlers import mqtt_handler as mqtt
+
+    monkeypatch.setattr(mqtt, 'device_target_cache', state.targets)
+    monkeypatch.setattr(messages, 'process_parent_leave_message', PROCESS_PARENT_MESSAGE)
+    monkeypatch.setattr(messages, 'process_cached_audio', PROCESS_DEVICE_MESSAGE)
+    monkeypatch.setattr(mqtt.card_service, 'get_card_by_uuid',
+                        AsyncMock(return_value=SimpleNamespace(device_id='TalkingQ_receiver', status=1)))
+    monkeypatch.setattr(mqtt.card_service, 'is_active_for_device', AsyncMock(return_value=False))
+    upload = AsyncMock(return_value=SimpleNamespace(file_key='local-test/parent.wav'))
+    create = AsyncMock(return_value=(SimpleNamespace(child_name='local-test'), None, []))
+    notify = Mock()
+    save = AsyncMock(return_value=('http://localhost/peer.wav', '/local-test/peer.wav'))
+    archive = AsyncMock(return_value=SimpleNamespace(message_id=123, media_file_key='local-test/peer.wav'))
+    pending = AsyncMock()
+    monkeypatch.setattr(messages, 'upload_message_audio', upload)
+    monkeypatch.setattr(messages.im_conversation_service, 'create_device_parent_leave_message', create)
+    monkeypatch.setattr(messages.wechat_mp_notification_service, 'schedule_leave_message_notification', notify)
+    monkeypatch.setattr(messages.device_audio_cache_service, 'save_device_audio', save)
+    monkeypatch.setattr(messages.device_voice_archive_service, 'archive_peer_voice_message', archive)
+    monkeypatch.setattr(messages.pending_voice_message_service, 'add_pending_message', pending)
+    service = mqtt.TalkingQMQTTService({'broker': ''})
+    service._publish = AsyncMock(return_value=True)
+
+    async def authenticate(ws):
+        await state.connections.add_connection(DEVICE, ws)
+        await ws.send_text('LOCAL_AUTHENTICATED')
+        return True, DEVICE, 'LOCAL-SERIAL'
+
+    monkeypatch.setattr(endpoint, 'authenticate_websocket', authenticate)
+    app = FastAPI()
+    app.include_router(router)
+
+    @app.post('/local-command')
+    async def command():
+        if kind == 'parent':
+            await service._handle_short_press_message(DEVICE, {'params': {'uuid': DEVICE}})
+        else:
+            await service._handle_nfc_listen_report(DEVICE, {'params': {'uuid': 'LOCAL-CARD'}})
+        return {'ok': True}
+
+    audio = b'\x01\x00' * 800
+    with TestClient(app) as client:
+        with client.websocket_connect('/ws') as old:
+            assert old.receive_text() == 'LOCAL_AUTHENTICATED'
+            assert client.post('/local-command').json() == {'ok': True}
+            payload = service._publish.call_args.args[1]
+            assert payload['msg_id'] == ('011' if kind == 'parent' else '005')
+            if kind == 'device':
+                assert payload['type'] == 1
+            state.now[0] += 9
+            with client.websocket_connect('/ws') as new:
+                assert new.receive_text() == 'LOCAL_AUTHENTICATED'
+                new.send_bytes(packet('replacement', 1))
+                assert new.receive_text() == 'START_ACK'
+                old.close()
+                new.send_bytes(packet('replacement', 4, audio[:800]))
+                new.send_bytes(packet('replacement', 4, audio[800:]))
+                new.send_bytes(packet('replacement', 2))
+                assert new.receive_text() == 'END_ACK'
+                assert new.receive_text().endswith('/assets/audio/message_ok.mp3')
+    assert state.started == []
+    assert await state.targets.get_voice_target(DEVICE) is None
+    assert not state.buffers.audio_data_cache
+    assert await state.sessions.get_all_sessions() == []
+    if kind == 'parent':
+        assert create.call_args.kwargs['audio_content'][44:] == audio
+        upload.assert_awaited_once()
+        notify.assert_called_once()
+        pending.assert_not_awaited()
+    else:
+        save.assert_awaited_once_with(audio, device_id='TalkingQ_receiver')
+        archive.assert_awaited_once_with(sender_device_id=DEVICE, receiver_device_id='TalkingQ_receiver',
+                                        local_audio_path='/local-test/peer.wav')
+        assert pending.call_args.kwargs['im_message_id'] == 123
+        assert pending.call_args.kwargs['sender_device_id'] == DEVICE
+        assert pending.call_args.kwargs['target_device_id'] == 'TalkingQ_receiver'
+        create.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize('kind', ['parent', 'device'])
 async def test_ai_audio_recovers_from_pending_message_without_losing_first_chunk(state, kind):
@@ -213,7 +431,7 @@ async def test_two_senders_to_same_receiver_use_separate_buffers(state):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize('claimed', [False, True])
-async def test_disconnect_clears_pending_or_partial_target(state, claimed):
+async def test_disconnect_only_clears_target_claimed_by_that_recording(state, claimed):
     await state.connections.add_connection(DEVICE, state.ws)
     target = await set_target(state, 'parent')
     if claimed:
@@ -222,7 +440,7 @@ async def test_disconnect_clears_pending_or_partial_target(state, claimed):
         session = await state.sessions.get_session((DEVICE, 'message'))
         session.set_interrupted = AsyncMock()
     await cleanup.cleanup_device_sessions(DEVICE, state.ws)
-    assert await state.targets.get_voice_target(DEVICE) is None
+    assert await state.targets.get_voice_target(DEVICE) is (None if claimed else target)
     assert await state.buffers.get_audio_data(target.audio_cache_key) is None
     assert await state.sessions.get_all_sessions() == []
 
@@ -256,12 +474,15 @@ async def test_old_socket_cleanup_and_late_packets_preserve_new_connection(state
 
 
 @pytest.mark.asyncio
-async def test_command_before_connect_is_adopted_and_cleared_on_early_disconnect(state):
+async def test_pending_command_survives_early_disconnect_without_extending_deadline(state):
     target = await set_target(state, 'parent')
     assert target.websocket is None
     await state.connections.add_connection(DEVICE, state.ws)
     await state.targets.bind_connection(DEVICE, state.ws)
     await cleanup.cleanup_device_sessions(DEVICE, state.ws)
+    assert await state.targets.get_voice_target(DEVICE) is target
+    assert target.websocket is None
+    state.now[0] += 10
     assert await state.targets.get_voice_target(DEVICE) is None
 
 
@@ -343,3 +564,4 @@ async def test_new_recording_recovers_abandoned_message_without_reusing_audio(st
     assert state.ended == [('new-ai', b'question')]
     assert await state.buffers.get_audio_data(abandoned.audio_cache_key) is None
     messages.process_parent_leave_message.assert_not_awaited()
+    assert not any('save_audio_fail.mp3' in c.args[0] for c in state.ws.send_text.call_args_list)