fix(voice): preserve pending message intent across reconnects

banban-mini/src/components/firmware-progress/view-model.ts

@@ -1,17 +1,20 @@
 import type { DeviceFirmwareStatus } from '@/services/device'
-import { isFirmwareCurrent, isFirmwareTargetReached, isFirmwareUpdateInProgress } from '../../utils/firmware-status'
+import { isFirmwareCurrent, isFirmwareTargetReached, isFirmwareUpdateInProgress, normalizeFirmwareStatus } from '../../utils/firmware-status'
 
 export function firmwareProgressView(status: DeviceFirmwareStatus | null, refreshError = false) {
-  const state = String(status?.update_status || '').trim().toLowerCase()
+  const normalizedStatus = normalizeFirmwareStatus(status)
+  const state = String(normalizedStatus?.update_status || '').trim().toLowerCase()
   const complete = state === 'success' || state === 'completed'
-    || (isFirmwareUpdateInProgress(state) && isFirmwareTargetReached(status))
-  const current = !complete && !isFirmwareUpdateInProgress(state) && isFirmwareCurrent(status)
+    || (isFirmwareUpdateInProgress(state) && isFirmwareTargetReached(normalizedStatus))
+  const current = !complete && !isFirmwareUpdateInProgress(state) && isFirmwareCurrent(normalizedStatus)
   const failed = state === 'failed' && !current
-  const value = Number(status?.progress)
+  const value = Number(normalizedStatus?.progress)
   const progress = complete ? 100 : Number.isFinite(value) ? Math.max(0, Math.min(100, value)) : 0
   const hasProgress = complete || progress > 0
-  const target = status?.target_version
-  const version = current ? status?.current_version : target && !['unknown', 'updating'].includes(target) ? target : status?.latest_version
+  // MQTT 003 carries target_version; latest_version is display-only fallback
+  // for older status rows and must not be used for completion detection.
+  const target = normalizedStatus?.target_version
+  const version = current ? normalizedStatus?.current_version : target && !['unknown', 'updating'].includes(target) ? target : normalizedStatus?.latest_version
   const title = complete ? '固件更新完成' : current ? '当前已是最新版本' : failed ? '固件更新失败' : '固件更新中'
   const message = complete ? '设备已完成更新，可以继续使用。'
     : current ? '当前设备已运行最新版本，无需更新。'

banban-mini/src/pages/bind/index.config.ts

@@ -1,3 +1,4 @@
 export default definePageConfig({
-  navigationStyle: 'custom'
+  navigationStyle: 'custom',
+  enableShareAppMessage: true,
 })

banban-mini/src/pages/bind/index.tsx

@@ -20,13 +20,13 @@ import {
   CHILD_NAME_RULE_MESSAGE,
   createChild,
   getChildren,
-  sanitizeChildNameInput,
   resolveChildSelection,
   setSelectedChildId as setStoredSelectedChildId,
   validateChildNameInput,
 } from '@/services/child'
 import { useSystemBanner } from '@/components/system-banner/use-system-banner'
 import { trafficPageUrl } from '@/services/traffic-binding'
+import { usePageShare } from '@/utils/page-share'
 
 import './index.scss'
 
@@ -161,6 +161,7 @@ function parseBindingPayload(rawValue: string): { deviceId: string; serialNumber
 }
 
 export default function Bind() {
+  usePageShare()
   const router = useRouter()
   const isAssignMode = String(router.params?.mode || '').trim() === 'assign'
   const isCardMode = String(router.params?.mode || '').trim() === 'card'
@@ -802,9 +803,10 @@ export default function Bind() {
                   ? children.length > 0 ? '不选已有宝贝时，可填写2-5位简体中文姓名（暂不支持数字和英文）' : CHILD_NAME_RULE_MESSAGE
                   : children.length > 0 ? '不选已有宝贝时，可填写2-5位简体中文姓名（暂不支持数字和英文）' : '可不填，稍后再关联宝贝'
               }
-              maxlength={5}
+              maxlength={-1}
               onInput={(event) => {
-                const nextChildName = sanitizeChildNameInput(event.detail.value)
+                // Preserve pinyin composition; validate the final name on submit.
+                const nextChildName = event.detail.value
                 setNewChildName(nextChildName)
                 if (nextChildName) {
                   setSelectedChildId(null)

banban-mini/src/pages/chat/detail/index.config.ts

@@ -1,3 +1,4 @@
 export default definePageConfig({
-  navigationStyle: 'custom'
+  navigationStyle: 'custom',
+  enableShareAppMessage: true,
 })

banban-mini/src/pages/chat/detail/index.tsx

@@ -13,6 +13,7 @@ import {
   PeerKind,
 } from '../../../services/chat'
 import { useSystemBanner } from '@/components/system-banner/use-system-banner'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 function normalizePeerKind(value?: string): PeerKind {
@@ -194,6 +195,7 @@ function getAudioUrlDebugInfo(rawUrl: string) {
 }
 
 export default function ChatDetail() {
+  usePageShare()
   const systemBanner = useSystemBanner()
   const router = useRouter()
   const params = router.params

banban-mini/src/pages/chat/index.config.ts

@@ -1,4 +1,5 @@
 export default {
   navigationBarTitleText: '聊天记录',
-  navigationStyle: 'custom'
-}
+  navigationStyle: 'custom',
+  enableShareAppMessage: true,
+}

banban-mini/src/pages/chat/index.tsx

@@ -9,9 +9,11 @@ import Taro, { useDidShow } from '@tarojs/taro'
 import { getToken } from '@/services/auth'
 import { getConversations, ChatConversation } from '../../services/chat'
 import { useSystemBanner } from '@/components/system-banner/use-system-banner'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 export default function Chat() {
+  usePageShare()
   const systemBanner = useSystemBanner()
   const [conversations, setConversations] = useState<ChatConversation[]>([])
   const [loading, setLoading] = useState(true)

banban-mini/src/pages/device/index.config.ts

@@ -1,3 +1,4 @@
 export default definePageConfig({
-  navigationStyle: 'custom'
-})
+  navigationStyle: 'custom',
+  enableShareAppMessage: true,
+})

banban-mini/src/pages/device/index.tsx

@@ -10,6 +10,7 @@ import { Binding, loadCurrentChildBindingContext } from '@/services/binding'
 import { Child } from '@/services/child'
 import { DeviceAlarmItem, DeviceStatus, getDeviceAlarms, getDeviceStatus, getDeviceVolumeCommandStatus, setDeviceRemoteSleepWake, setDeviceVolume } from '@/services/device'
 import { useSystemBanner } from '@/components/system-banner/use-system-banner'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 function formatTime(value?: string | null): string {
@@ -119,6 +120,7 @@ function getAlarmLocationHint(alarm: DeviceAlarmItem | null): string {
 }
 
 export default function Device() {
+  usePageShare()
   const systemBanner = useSystemBanner()
   const [binding, setBinding] = useState<Binding | null>(null)
   const [pendingBinding, setPendingBinding] = useState<Binding | null>(null)

banban-mini/src/pages/family-invite/index.config.ts

@@ -1,3 +1,4 @@
 export default definePageConfig({
   navigationBarTitleText: '家庭邀请',
+  enableShareAppMessage: true,
 })

banban-mini/src/pages/family-invite/index.tsx

@@ -7,9 +7,11 @@ import { getToken } from '@/services/auth'
 import { setSelectedBindingDeviceId } from '@/services/binding'
 import { setSelectedChildId } from '@/services/child'
 import { acceptFamilyInvitation, FamilyInvitation, getFamilyInvitation } from '@/services/family'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 export default function FamilyInvite() {
+  usePageShare()
   const router = useRouter()
   const inviteToken = String(router.params?.token || '').trim()
   const redirectUrl = `/pages/family-invite/index?token=${inviteToken}`

banban-mini/src/pages/location/index.config.ts

@@ -1,3 +1,4 @@
 export default definePageConfig({
-  navigationStyle: 'custom'
-})
+  navigationStyle: 'custom',
+  enableShareAppMessage: true,
+})

banban-mini/src/pages/location/index.tsx

@@ -13,6 +13,7 @@ import {
   DeviceTrajectoryPoint,
 } from '@/services/location'
 import { useSystemBanner } from '@/components/system-banner/use-system-banner'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 type TrajectoryMode = 'current' | 'today' | 'recent'
@@ -127,6 +128,7 @@ function getTrajectoryMarkerId(index: number): number {
 }
 
 export default function Location() {
+  usePageShare()
   const systemBanner = useSystemBanner()
   const [loading, setLoading] = useState(true)
   const [deviceLocation, setDeviceLocation] = useState<DeviceLocation | null>(null)

banban-mini/src/pages/login/index.config.ts

@@ -1,3 +1,4 @@
 export default definePageConfig({
   navigationBarTitleText: '微信登录',
+  enableShareAppMessage: true,
 })

banban-mini/src/pages/login/index.tsx

@@ -3,6 +3,7 @@ import { View, Text, Button, Checkbox, CheckboxGroup, Image } from '@tarojs/comp
 import { useEffect, useState } from 'react'
 import Taro from '@tarojs/taro'
 import { getToken, wechatLogin } from '@/services/auth'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 const LOGIN_PAGE_URL = '/pages/login/index'
@@ -39,6 +40,7 @@ function navigateAfterLogin(url: string): void {
 }
 
 export default function Login() {
+  usePageShare()
   const [submitting, setSubmitting] = useState(false)
   const [agreementAccepted, setAgreementAccepted] = useState(false)
 

banban-mini/src/pages/notification/index.config.ts

@@ -0,0 +1,3 @@
+export default definePageConfig({
+  enableShareAppMessage: true,
+})

banban-mini/src/pages/notification/index.tsx

@@ -4,6 +4,7 @@ import { useEffect, useState } from 'react'
 import Taro, { useRouter } from '@tarojs/taro'
 import { getToken } from '@/services/auth'
 import { resolveWechatMpNotificationLink } from '@/services/notification'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 function buildQuery(params: Record<string, string | number | null | undefined>): string {
@@ -18,6 +19,7 @@ function buildCurrentPageUrl(token: string): string {
 }
 
 export default function NotificationEntry() {
+  usePageShare()
   const router = useRouter()
   const token = String(router.params.token || '').trim()
   const [message, setMessage] = useState('正在打开留言...')

banban-mini/src/pages/sleep-schedule/index.config.ts

@@ -2,4 +2,5 @@ export default definePageConfig({
   navigationBarTitleText: '定时休眠',
   navigationBarBackgroundColor: '#F5F7FA',
   navigationBarTextStyle: 'black',
+  enableShareAppMessage: true,
 })

banban-mini/src/pages/sleep-schedule/index.tsx

@@ -7,6 +7,7 @@ import { loadCurrentChildBindingContext, Binding } from '@/services/binding'
 import { Child } from '@/services/child'
 import { DeviceStatus, getDeviceStatus, setDeviceSleepSchedule } from '@/services/device'
 import { useSystemBanner } from '@/components/system-banner/use-system-banner'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 function formatSleepRange(status: DeviceStatus | null): string {
@@ -28,6 +29,7 @@ function formatUpdatedAt(value?: string | null): string {
 }
 
 export default function SleepSchedule() {
+  usePageShare()
   const systemBanner = useSystemBanner()
   const [isLoading, setIsLoading] = useState(true)
   const [isSaving, setIsSaving] = useState(false)

banban-mini/src/pages/sleep/index.config.ts

@@ -1,3 +1,4 @@
 export default definePageConfig({
-  navigationStyle: 'custom'
-})
+  navigationStyle: 'custom',
+  enableShareAppMessage: true,
+})

banban-mini/src/pages/sleep/index.tsx

@@ -26,7 +26,6 @@ import {
   CHILD_NAME_RULE_MESSAGE,
   clearSelectedChildId,
   deleteChild,
-  sanitizeChildNameInput,
   setSelectedChildId,
   updateChild,
   validateChildNameInput,
@@ -1496,8 +1495,8 @@ export default function Sleep() {
                 value={childName}
                 placeholder={CHILD_NAME_RULE_MESSAGE}
                 focus
-                maxlength={5}
-                onInput={(event) => setChildName(sanitizeChildNameInput(event.detail.value))}
+                maxlength={-1}
+                onInput={(event) => setChildName(event.detail.value)}
               />
             </View>
             <View className='modal-actions'>

banban-mini/src/pages/traffic-binding/index.config.ts

@@ -1 +1,4 @@
-export default definePageConfig({ navigationBarTitleText: '流量服务关联' })
+export default definePageConfig({
+  navigationBarTitleText: '流量服务关联',
+  enableShareAppMessage: true,
+})

banban-mini/src/pages/traffic-binding/index.tsx

@@ -4,11 +4,13 @@ import { useRef, useState } from 'react'
 import { getToken } from '@/services/auth'
 import { clearSelectedBindingDeviceId } from '@/services/binding'
 import { getPageStyle } from '@/utils/page-style'
+import { usePageShare } from '@/utils/page-share'
 import { TrafficOperation, getTrafficOperation, resumeTrafficOperation,
   flushTrafficReceipt, finishTrafficUnbind, jumpTraffic } from '@/services/traffic-binding'
 import './index.scss'
 
 export default function TrafficBindingPage() {
+  usePageShare()
   const { params } = useRouter()
   const deviceId = String(params.device_id || '')
   const action = params.action === 'unbind' ? 'unbind' : 'bind'

banban-mini/src/pages/webview/index.config.ts

@@ -1,3 +1,4 @@
 export default definePageConfig({
   navigationBarTitleText: '服务号通知',
+  enableShareAppMessage: true,
 })

banban-mini/src/pages/webview/index.tsx

@@ -1,8 +1,10 @@
 import { Text, View, WebView } from '@tarojs/components'
 import { useRouter } from '@tarojs/taro'
+import { usePageShare } from '@/utils/page-share'
 import './index.scss'
 
 export default function WebviewPage() {
+  usePageShare()
   const router = useRouter()
   const encodedUrl = String(router.params?.url || '').trim()
   const targetUrl = encodedUrl ? decodeURIComponent(encodedUrl) : ''

banban-mini/src/services/child.ts

@@ -17,10 +17,6 @@ export function normalizeChildNameInput(value?: string | null): string {
   return String(value || '').trim()
 }
 
-export function sanitizeChildNameInput(value?: string | null): string {
-  return normalizeChildNameInput(value).replace(/[^\u4e00-\u9fa5]/g, '').slice(0, 5)
-}
-
 export function validateChildNameInput(value?: string | null): string {
   const childName = normalizeChildNameInput(value)
   if (!childName) return '请输入宝贝姓名'

banban-mini/src/utils/firmware-status.ts

@@ -8,6 +8,20 @@ export function isFirmwareUpdateInProgress(status?: string | null): boolean {
   return ['sent', 'accepted', 'updating'].includes(normalizeFirmwareUpdateStatus(status))
 }
 
+export function normalizeFirmwareStatus(status: DeviceFirmwareStatus | null): DeviceFirmwareStatus | null {
+  if (!status) return null
+
+  const numericProgress = Number(status.progress)
+  const progress = Number.isFinite(numericProgress)
+    ? Math.max(0, Math.min(100, numericProgress))
+    : 0
+
+  return {
+    ...status,
+    progress,
+  }
+}
+
 function firmwareVersionParts(value?: string | null): number[] | null {
   const normalized = String(value || '').trim().toLowerCase()
   if (!normalized || ['unknown', 'updating', '0.0.0', '--'].includes(normalized)) return null
@@ -36,15 +50,16 @@ export function isFirmwareCurrent(status: DeviceFirmwareStatus | null): boolean
 }
 
 export function firmwareSummaryView(status: DeviceFirmwareStatus | null) {
-  if (!status) return { action: '暂不可用', subtitle: '暂时无法获取版本，请稍后重试' }
-  if (isFirmwareUpdateInProgress(status.update_status) && !isFirmwareTargetReached(status)) {
+  const normalizedStatus = normalizeFirmwareStatus(status)
+  if (!normalizedStatus) return { action: '暂不可用', subtitle: '暂时无法获取版本，请稍后重试' }
+  if (isFirmwareUpdateInProgress(normalizedStatus.update_status) && !isFirmwareTargetReached(normalizedStatus)) {
     return { action: '查看进度', subtitle: '设备正在更新，请稍候' }
   }
-  if (isFirmwareCurrent(status)) return { action: '已最新', subtitle: '当前已是最新版本' }
-  if (normalizeFirmwareUpdateStatus(status.update_status) === 'failed') {
-    return { action: status.update_available ? '更新' : '版本待确认', subtitle: status.can_update ? '更新失败，可重试' : '上次更新失败，当前版本待确认' }
+  if (isFirmwareCurrent(normalizedStatus)) return { action: '已最新', subtitle: '当前已是最新版本' }
+  if (normalizeFirmwareUpdateStatus(normalizedStatus.update_status) === 'failed') {
+    return { action: normalizedStatus.update_available ? '更新' : '版本待确认', subtitle: normalizedStatus.can_update ? '更新失败，可重试' : '上次更新失败，当前版本待确认' }
   }
-  return status.update_available
+  return normalizedStatus.update_available
     ? { action: '更新', subtitle: '发现可更新版本' }
     : { action: '版本待确认', subtitle: '暂无完整版本信息' }
 }

banban-mini/src/utils/page-share.ts

@@ -0,0 +1,40 @@
+import { useRouter, useShareAppMessage } from '@tarojs/taro'
+
+const DEFAULT_SHARE_TITLES: Record<string, string> = {
+  '/pages/device/index': '伴伴设备管理',
+  '/pages/login/index': '登录伴伴小程序',
+  '/pages/bind/index': '绑定伴伴设备',
+  '/pages/traffic-binding/index': '伴伴流量服务',
+  '/pages/notification/index': '查看伴伴通知',
+  '/pages/chat/index': '伴伴聊天记录',
+  '/pages/chat/detail/index': '查看伴伴聊天',
+  '/pages/location/index': '查看设备位置',
+  '/pages/sleep/index': '伴伴设备管理',
+  '/pages/sleep-schedule/index': '设置设备休眠时间',
+  '/pages/family-invite/index': '家庭共享邀请',
+  '/pages/webview/index': '伴伴服务通知',
+}
+
+function normalizePath(path: string): string {
+  if (!path) return '/pages/device/index'
+  return path.startsWith('/') ? path : `/${path}`
+}
+
+function buildPagePath(path: string, params: Record<string, string | undefined>): string {
+  const query = Object.entries(params)
+    .filter(([, value]) => value !== undefined && value !== '')
+    .map(([key, value]) => `${encodeURIComponent(key)}=${encodeURIComponent(String(value))}`)
+    .join('&')
+
+  return query ? `${path}?${query}` : path
+}
+
+export function usePageShare(title?: string): void {
+  const router = useRouter()
+  const pagePath = normalizePath(router.path)
+
+  useShareAppMessage(() => ({
+    title: title || DEFAULT_SHARE_TITLES[pagePath] || '伴伴小程序',
+    path: buildPagePath(pagePath, router.params || {}),
+  }))
+}

banban-mini/tests/firmware-progress.test.cjs

@@ -32,6 +32,22 @@ test('received progress is used directly, including fractional values', () => {
   assert.equal(view.percent, 65)
   assert.equal(view.complete, false)
 })
+test('MQTT 003 target_version is shown while progress is updating', () => {
+  const view = firmwareProgressView({
+    ...status('updating', 30),
+    latest_version: '2.0.4',
+    target_version: '2.0.4',
+  })
+  assert.equal(view.version, '2.0.4')
+  assert.equal(view.progress, 30)
+  assert.equal(view.percent, 30)
+  assert.match(view.message, /保持设备电量充足/)
+})
+test('active status without target_version falls back to latest_version', () => {
+  const view = firmwareProgressView(status('updating', 30))
+  assert.equal(view.version, '1.8.2')
+  assert.equal(view.progress, 30)
+})
 test('nearly complete progress never rounds up to 100', () => {
   assert.equal(firmwareProgressView(status('updating', 99.9)).percent, 99)
 })

talkingq-url/config.py

@@ -71,6 +71,7 @@ class Settings(BaseSettings):
 
     assets_dir: str = Field(default="assets", validation_alias="ASSETS_DIR")
     session_timeout: int = Field(default=600, validation_alias="SESSION_TIMEOUT")
+    voice_target_timeout_seconds: float = Field(default=120, gt=0, le=600, validation_alias="VOICE_TARGET_TIMEOUT_SECONDS")
     conversation_history_timeout: int = Field(default=1800, validation_alias="CONVERSATION_HISTORY_TIMEOUT")
     max_conversation_history: int = Field(default=5, validation_alias="MAX_CONVERSATION_HISTORY")
     cleanup_interval: int = Field(default=300, validation_alias="CLEANUP_INTERVAL")

talkingq-url/docs/ota_flow.md

@@ -127,6 +127,21 @@ Headers:
 }
 ```
 
+设备通过 MQTT `msg_id=003` 上报实时进度时，使用 JSON `data.target_version` 和 `data.progress`，例如：
+
+```json
+{
+  "msg_id": "003",
+  "status": "updating",
+  "data": {
+    "target_version": "2.0.4",
+    "progress": 30
+  }
+}
+```
+
+服务器保存该进度，小程序继续通过本状态接口轮询显示。`progress=100` 只表示设备报告写入进度完成，设备重启后上报的实际版本仍是最终完成依据。
+
 ### 4. 管理固件配置信息（管理员接口）
 
 **获取固件配置**

talkingq-url/docs/ota_progress_protocol.md

@@ -0,0 +1,31 @@
+# OTA `003` 实时进度协议
+
+设备通过 `device/{device_id}/response` 上报 OTA 状态。服务器下发 OTA 命令仍使用 `msg_id=003`，设备回执使用相同的 `msg_id`。
+
+## 进行中回执
+
+```json
+{
+  "msg_id": "003",
+  "status": "updating",
+  "data": {
+    "target_version": "2.0.4",
+    "progress": 30
+  }
+}
+```
+
+`progress` 表示设备当前已完成的百分比，范围为 `0` 到 `100`。设备应在状态变化时上报，并避免高频重复发送。`target_version` 应与服务器下发任务的目标版本一致。
+
+## 状态约定
+
+| 状态 | 必要字段 | 服务器行为 |
+| --- | --- | --- |
+| `accepted` | `current_version`、`target_version` | 记录设备已接受任务，进度默认为 `0` |
+| `updating` | `target_version`、`progress` | 保存最新进度并供小程序查询 |
+| `success` | `new_version` 或 `target_version` | 记录完成，进度置为 `100`，随后以设备重启后的版本上报作为最终确认 |
+| `failed` | `target_version`、可选 `progress` | 保存失败状态和最后有效进度 |
+
+服务器兼容旧设备上报的 `version`、`new_version` 字段，并优先使用 `target_version` 作为本次任务目标。非法进度值不会覆盖已保存的有效进度。
+
+小程序通过设备固件状态接口轮询 `update_status`、`progress` 和 `target_version`，显示最近一次服务器收到的设备进度。进度达到 `100` 仍显示为等待设备确认，直到设备重启后上报目标版本。

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

@@ -0,0 +1,64 @@
+# 留言状态与后续 AI 对话恢复
+
+留言分为待录音意图和已关联录音两个阶段。刷卡或短按先创建待录音意图，不绑定
+连接表中的旧 WebSocket；录音开始包再将意图关联到唯一的连接和 session。
+新连接接入、尚未录音便断开以及旧连接延迟退出，均不清除待录音意图；
+已关联录音的目标仍在其连接断开或被替换时清理。
+
+目标设置后默认空闲有效期为 120 秒，可用 `VOICE_TARGET_TIMEOUT_SECONDS` 调整；
+录音开始及有效留言音频会刷新空闲期限，单纯连接或重连不会延长有效期。
+每次访问均检查过期状态，后台会定期回收过期目标及缓存。
+
+开始包统一返回一次 `START_ACK`，先登记会话，再根据首个音频包选择流程：
+类型 `0` 清除待处理留言目标并完成 AI 初始化，原始首包随后正常送入 ASR；
+类型 `4` 写入本次留言的独立缓存。录音开始传输音频后不允许切换类型。
+没有有效目标的留言音频不会启动 AI，结束时返回 `END_ACK` 和既有留言保存失败
+提示音；旧录音延迟结束时不向正在进行的新会话插入该失败提示。
+没有音频的普通 START/END 会话仍走原有静音识别响应。
+
+已关联录音的目标及音频缓存，在结束、取消、异常、空音频、超时或连接断开时释放。
+结束时先消费本次目标，再处理上传和通知，避免旧处理完成时清除新设置的目标。
+旧连接断开只清理该连接的会话和任务，不影响替换它的新连接。
+音频缓存键按发送设备及本次目标独立生成，多个发送方对同一接收方留言时互不覆盖。
+
+本次修复保留既有数据包类型、鉴权及设备协议。家长留言的上传、会话保存和通知流程
+沿用原实现。
+
+## 本地验证（2026-10-28）
+
+- 全量后端回归 243/243，包括新增的 28 项留言状态与恢复测试。
+- 本机隔离 MySQL、原有 `/ws` 路由和真实 ASR/LLM/TTS：正常对话及静音 10/10，
+  中断及恢复 4/4；包含两个角色在待处理、过期、取消家长留言后的 AI 对话。
+- 实际 WebSocket 家长留言取消 2 次，均收到取消确认及处理完成通知，随后 AI 对话成功。
+- 共 52 段音频通过 HTTP 200、Range 206、完整 MP3 解码及非零 PCM 检查。
+- 正常留言的上传、保存、通知和失败分支使用本地替身测试，未向真实家长发送通知。
+- 没有访问或部署生产服务，未验证实物设备的麦克风、扬声器及播放中打断。
+
+覆盖边界：留言取消、空音频、处理异常、过期及未结束录音后的 AI 恢复已有回归；
+AI 中断后再次 AI 对话已有真实服务验证。留言取消后再次留言验证了录音隔离，
+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/audio_session_handler.py

@@ -22,8 +22,10 @@ async def handle_websocket_data(
 
     if packet_type == 1:  # 开始包
         try:
-            await websocket.send_text("START_ACK")
-            session_logger.info(device_id, session_id, "已发送开始包确认")
+            if not session.start_ack_sent:
+                await websocket.send_text("START_ACK")
+                session.start_ack_sent = True
+                session_logger.info(device_id, session_id, "已发送开始包确认")
         except Exception as e:
             session_logger.error(device_id, session_id, f"发送开始包确认失败: {str(e)}")
         session.sample_rate = sample_rate

talkingq-url/handlers/mqtt_handler.py

@@ -1,5 +1,6 @@
 import asyncio
 import json
+import math
 import time
 from datetime import datetime
 from typing import Awaitable, Callable, Dict, Optional
@@ -311,26 +312,39 @@ class TalkingQMQTTService:
 
     async def _handle_ota_response(self, device_id: str, payload: dict):
         status = payload.get("status")
-        data = payload.get("data", {})
+        data = payload.get("data") or {}
+        if not isinstance(data, dict):
+            logger.warning(device_id, "", f"[OTA] invalid response data: {payload}")
+            return
+
+        target_version = data.get("target_version") or data.get("version") or data.get("new_version")
         progress = data.get("progress")
         try:
-            progress_value = float(progress) if progress is not None else None
+            parsed_progress = float(progress) if progress is not None else None
+            progress_value = (
+                parsed_progress
+                if parsed_progress is not None and math.isfinite(parsed_progress) and 0 <= parsed_progress <= 100
+                else None
+            )
         except (TypeError, ValueError):
             progress_value = None
 
+        if progress is not None and progress_value is None:
+            logger.warning(device_id, "", f"[OTA] invalid progress value: {progress!r}")
+
         if status == "accepted":
             current = data.get("current_version")
-            target = data.get("target_version")
-            logger.info(device_id, "", f"[OTA] 设备 {device_id} 已接受升级: {current} -> {target}")
+            logger.info(device_id, "", f"[OTA] 设备 {device_id} 已接受升级: {current} -> {target_version}")
         
             await self._schedule_persistence(
                 device_id,
                     "ota",
                     device_firmware_update_manager.update_firmware_update(
                         device_id=device_id,
-                        firmware_version=data.get("target_version") or data.get("version") or data.get("new_version") or "unknown",
+                        firmware_version=target_version or "unknown",
                         update_status="accepted",
                         progress=progress_value if progress_value is not None else 0.0,
+                        target_version=target_version,
                     ),
             )
         elif status == "updating":
@@ -339,13 +353,14 @@ class TalkingQMQTTService:
                 "ota",
                 device_firmware_update_manager.update_firmware_update(
                     device_id=device_id,
-                    firmware_version=data.get("version") or data.get("new_version") or "updating",
+                    firmware_version=target_version or "updating",
                     update_status="updating",
                     progress=progress_value,
+                    target_version=target_version,
                 ),
             )
         elif status == "success":
-            new_version = data.get("new_version") or data.get("version")
+            new_version = data.get("new_version") or data.get("version") or data.get("target_version")
             logger.info(device_id, "", f"[OTA] 设备 {device_id} 升级完成: {new_version}")
             
             await self._schedule_persistence(
@@ -367,6 +382,7 @@ class TalkingQMQTTService:
                     firmware_version=new_version or "unknown",
                     update_status="success",
                     progress=100.0,
+                    target_version=target_version,
                 ),
             )
         elif status == "failed":
@@ -375,9 +391,10 @@ class TalkingQMQTTService:
                 "ota",
                 device_firmware_update_manager.update_firmware_update(
                     device_id=device_id,
-                    firmware_version=data.get("version") or data.get("new_version") or "unknown",
+                    firmware_version=target_version or "unknown",
                     update_status="failed",
                     progress=progress_value,
+                    target_version=target_version,
                 ),
             )
             logger.warning(device_id, "", f"[OTA] command failed: {payload}")

talkingq-url/handlers/session_cleanup_handler.py

@@ -1,6 +1,8 @@
 from services.interrupt_handler import interrupt_handler
 from services.audio_session import audio_session_manager
 from utils.logger import session_logger
+from services.device_target_cache import device_target_cache
+from services.task_manager import task_manager
 
 async def handle_old_session_cleanup(old_session, session_key):
     """
@@ -21,19 +23,22 @@ async def handle_old_session_cleanup(old_session, session_key):
         session_logger.error("unknown", "unknown", f"清理旧会话资源时出错: {e}")
 
 
-async def cleanup_device_sessions(device_id):
+async def cleanup_device_sessions(device_id, websocket=None):
     """
     清理设备所有会话
     
     Args:
         device_id: 设备ID
     """
+    await device_target_cache.remove_target(device_id, websocket=websocket, reason="disconnected")
     sessions_to_clean = []
     for session_key, session in await audio_session_manager.get_all_sessions():
-        if session_key[0] == device_id:
+        if session_key[0] == device_id and (websocket is None or session.websocket is websocket):
             sessions_to_clean.append((session_key, session))
     
     for session_key, session in sessions_to_clean:
+        if websocket is not None and session.websocket is not websocket:
+            continue
         if hasattr(session, "set_interrupted"):
             session_logger.info(
                 device_id, session_key[1], "WebSocket连接断开，中断会话"
@@ -44,8 +49,11 @@ async def cleanup_device_sessions(device_id):
             await interrupt_handler.set_interrupt_state(session_key, True)
             if session.llm_service and hasattr(session.llm_service, "closed"):
                 session.llm_service.closed = True
+        await task_manager.cancel_session_tasks(session_key)
     
-    for session_key, _ in sessions_to_clean:
+    for session_key, session in sessions_to_clean:
+        if websocket is not None and session.websocket is not websocket:
+            continue
         await audio_session_manager.remove_session(session_key)
         await interrupt_handler.remove_session(session_key)
         session_logger.info(

talkingq-url/handlers/websocket_handler.py

@@ -1,7 +1,6 @@
 from fastapi import WebSocket, WebSocketDisconnect
 from utils.logger import session_logger
 from services.connection_manager import connection_manager
-from services.task_manager import task_manager
 from services.offline_audio_cache import offline_audio_cache
 from handlers.websocket_auth_handler import authenticate_websocket
 from handlers.websocket_message_handler import handle_websocket_messages
@@ -59,9 +58,7 @@ async def websocket_endpoint(websocket: WebSocket):
     finally:
         if device_id:
             await connection_manager.remove_connection(device_id, websocket)
-            await cleanup_device_sessions(device_id)
-            # 清理设备相关的所有异步任务
-            await task_manager.cancel_device_tasks(device_id)
+            await cleanup_device_sessions(device_id, websocket)
         
         session_logger.info(
             device_id or "unknown",

talkingq-url/handlers/websocket_message_handler.py

@@ -44,6 +44,7 @@ async def handle_websocket_messages(websocket: WebSocket, device_id: str, serial
         websocket: WebSocket连接
         device_id: 设备ID
     """
+    await device_target_cache.bind_connection(device_id, websocket)
     current_active_session = None
     first_audio_received_time = None
     
@@ -156,54 +157,121 @@ async def handle_text_message(websocket: WebSocket, device_id: str, text_data: s
 
 async def handle_binary_message(websocket: WebSocket, device_id: str, serial_number: str, binary_data, current_active_session, first_audio_received_time):
     """处理二进制音频消息"""
+    active_connection = await connection_manager.get_connection(device_id)
+    if active_connection is not None and active_connection is not websocket:
+        session_logger.warning(device_id, "connection", "忽略已被替换连接的录音包")
+        return current_active_session, first_audio_received_time
     session_key, session, packet_type, audio_data, sample_rate = await parse_packet(binary_data, websocket)
-    
-    if session:
-        session.websocket = websocket
-        session_id = session_key[1]
-        
-        if packet_type == 1:  # 开始包
-            voice_target = await device_target_cache.get_voice_target(device_id)
-            if voice_target:
-                await target_audio_cache.clear_audio_data(voice_target.audio_cache_key)
-                return None, None
-            current_active_session = await handle_start_packet(
-                device_id, session_id, session_key, session, current_active_session
+    if not session:
+        return current_active_session, first_audio_received_time
+    session_id = session_key[1]
+    if session.websocket is not None and session.websocket is not websocket:
+        session_logger.warning(device_id, session_id, "忽略其他连接的录音包")
+        return current_active_session, first_audio_received_time
+    session.websocket = websocket
+    session.last_activity_time = asyncio.get_running_loop().time()
+
+    if packet_type == 1:
+        if session.start_received:
+            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)
+        # START does not identify AI versus message audio. Decide on the first
+        # audio packet, even without a target, so an invalid message cannot
+        # start ASR and later report an unrelated recognition failure.
+        current_active_session = await handle_start_packet(
+            device_id, session_id, session_key, session, current_active_session)
+        await websocket.send_text("START_ACK")
+        session.start_ack_sent = True
+        return current_active_session, None
+
+    if not session.start_received or session.recording_finished:
+        session_logger.warning(device_id, session_id, "忽略未开始或已结束的录音包")
+        return current_active_session, first_audio_received_time
+
+    if packet_type == 0:
+        if session.message_audio_received:
+            session_logger.warning(device_id, session_id, "忽略留言录音中混入的普通对话音频")
+            return current_active_session, first_audio_received_time
+        if not session.ai_started:
+            target = session.voice_target
+            if target:
+                await device_target_cache.remove_target(device_id, expected=target, reason="ai_audio")
+                session.voice_target = None
+            current_active_session = await start_ai_recording(
+                websocket, device_id, session_id, session_key, session, current_active_session, sample_rate
             )
-        elif packet_type == 3:  # 中断包
-            voice_target = await device_target_cache.get_voice_target(device_id)
-            if voice_target:
-                return None, None
-            current_active_session = await handle_interrupt_packet(
-                device_id, session_id, session_key, session, current_active_session
-            )
-        elif packet_type == 2:  # 结束包
-            voice_target = await device_target_cache.get_voice_target(device_id)
-            if voice_target:
-                session_logger.info(device_id, session_id, f"收到结束包，开始处理缓存音频数据")
-                if voice_target.kind == VOICE_TARGET_KIND_PARENT:
-                    await process_parent_leave_message(device_id, voice_target.audio_cache_key)
-                else:
-                    await process_cached_audio(device_id, voice_target.target_device_id, serial_number)
-                await device_target_cache.remove_target(device_id)
-                return None, None
-            if current_active_session == session_key:
-                current_active_session = None
-        elif packet_type == 4:  # 发送给目标设备的音频包
-            await handle_target_audio_packet(device_id, audio_data)
-            return None, None
-        elif packet_type == 0 and first_audio_received_time is None:
+        if first_audio_received_time is None:
             first_audio_received_time = time.perf_counter()
             session.start_time = first_audio_received_time
-            session_logger.info(
-                device_id, session_id, f"收到第一个音频数据包时间: {first_audio_received_time}"
-            )
-        
-        await handle_websocket_data(websocket, session_key, session, packet_type, audio_data, sample_rate)
-    
+            session_logger.info(device_id, session_id, f"收到第一个音频数据包时间: {first_audio_received_time}")
+    elif packet_type == 4:
+        target = session.voice_target
+        if not session.ai_started:
+            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, "忽略普通对话中混入的留言音频")
+        return current_active_session, first_audio_received_time
+    elif packet_type == 3:
+        if not session.ai_started:
+            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 or session.message_audio_received:
+            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:
+                await websocket.send_text("END_ACK")
+                if target:
+                    session_logger.info(device_id, session_id, "收到结束包，开始处理缓存音频数据")
+                    if target.kind == VOICE_TARGET_KIND_PARENT:
+                        await process_parent_leave_message(device_id, target.audio_cache_key)
+                    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
+
+    await handle_websocket_data(websocket, session_key, session, packet_type, audio_data, sample_rate)
+    if packet_type in (2, 3):
+        session.recording_finished = True
     return current_active_session, first_audio_received_time
 
 
+async def start_ai_recording(websocket, device_id, session_id, session_key, session, current_active_session, sample_rate):
+    await handle_websocket_data(websocket, session_key, session, 1, b"", sample_rate)
+    session.ai_started = True
+    return current_active_session
+
+
 async def handle_start_packet(device_id, session_id, session_key, session, current_active_session):
     """处理会话开始包"""
     if current_active_session and current_active_session[1] != session_id:
@@ -255,20 +323,17 @@ async def handle_interrupt_packet(device_id, session_id, session_key, session, c
     return current_active_session
 
 
-async def handle_target_audio_packet(device_id: str, audio_data: bytes):
+async def handle_target_audio_packet(device_id: str, audio_data: bytes, voice_target, websocket, session_id):
     """处理发送给目标设备的音频包"""
     try:
-        voice_target = await device_target_cache.get_voice_target(device_id)
-        
-        if not voice_target:
-            session_logger.warning(device_id, "target", "未设置目标设备，无法发送音频")
-            return 
-        
-        await target_audio_cache.add_audio_data(voice_target.audio_cache_key, audio_data)
+        if not await device_target_cache.append_audio(device_id, voice_target, websocket, session_id, audio_data):
+            session_logger.warning(device_id, session_id, "忽略已过期或不属于本次录音的留言音频")
+            return
         if voice_target.kind == VOICE_TARGET_KIND_PARENT:
             session_logger.info(device_id, "target", "已缓存发给家长的留言音频数据")
         else:
             session_logger.info(device_id, "target", f"已缓存音频数据到目标设备 {voice_target.target_device_id}")
+        return True
         
     except Exception as e:
         session_logger.error(device_id, "target", f"处理目标音频包时出错: {e}", exc_info=True)
@@ -337,10 +402,11 @@ async def process_parent_leave_message(device_id: str, audio_cache_key: str):
         await target_audio_cache.clear_audio_data(audio_cache_key)
 
 
-async def process_cached_audio(device_id: str, target_device_id: str, serial_number: str):
+async def process_cached_audio(device_id: str, target_device_id: str, serial_number: str, *, audio_cache_key=None):
     """处理缓存的音频数据并发送音频URL"""
     try:
-        cached_audio = await target_audio_cache.get_audio_data(target_device_id)
+        cache_key = audio_cache_key or target_device_id
+        cached_audio = await target_audio_cache.get_audio_data(cache_key)
         if not cached_audio:
             session_logger.info(device_id, "target", f"目标设备 {target_device_id} 没有缓存的音频数据")
             return
@@ -398,4 +464,4 @@ async def process_cached_audio(device_id: str, target_device_id: str, serial_num
             session_logger.error(device_id, "target", f"处理缓存音频时出错: {e}", exc_info=True)
     finally:
         # 清除缓存
-        await target_audio_cache.clear_audio_data(target_device_id)
+        await target_audio_cache.clear_audio_data(cache_key)

talkingq-url/services/audio_session.py

@@ -30,6 +30,12 @@ class AudioSession:
         self.session_id = None
         self.sample_rate = sample_rate
         self.audio_queue = None
+        self.voice_target = None
+        self.start_received = False
+        self.start_ack_sent = False
+        self.ai_started = False
+        self.recording_finished = False
+        self.message_audio_received = False
 
     async def set_interrupted(self, interrupted=True):
         

talkingq-url/services/device_target_cache.py

@@ -1,59 +1,140 @@
 import asyncio
+import time
 from dataclasses import dataclass
 from typing import Dict, Optional
+from uuid import uuid4
+
+from config import settings
+from services.connection_manager import connection_manager
+from services.target_audio_cache import target_audio_cache
+from utils.logger import session_logger
 
 
 VOICE_TARGET_KIND_DEVICE = "device"
 VOICE_TARGET_KIND_PARENT = "parent"
 
 
-@dataclass(frozen=True)
+@dataclass
 class VoiceTarget:
     kind: str
     target_device_id: Optional[str] = None
     uuid: Optional[str] = None
     audio_cache_key: Optional[str] = None
+    websocket: object = None
+    session_id: Optional[str] = None
+    expires_at: float = 0
 
 
 class DeviceTargetCache:
-    def __init__(self):
+    def __init__(self, *, timeout_seconds=None, clock=None):
         self.targets: Dict[str, VoiceTarget] = {}
         self.lock = asyncio.Lock()
+        self.timeout_seconds = timeout_seconds if timeout_seconds is not None else settings.voice_target_timeout_seconds
+        self.clock = clock or time.monotonic
+
+    async def _remove_locked(self, device_id, target, reason, *, clear_audio=True):
+        del self.targets[device_id]
+        if clear_audio:
+            await target_audio_cache.clear_audio_data(target.audio_cache_key)
+        session_logger.info(device_id, target.session_id or "target",
+                            f"留言目标已清除: kind={target.kind}, reason={reason}")
+
+    async def _get_locked(self, device_id):
+        target = self.targets.get(device_id)
+        if target and self.clock() >= target.expires_at:
+            await self._remove_locked(device_id, target, "timeout")
+            return None
+        return target
 
     async def get_target(self, device_id: str) -> Optional[str]:
-        async with self.lock:
-            target = self.targets.get(device_id)
-            if not target or target.kind != VOICE_TARGET_KIND_DEVICE:
-                return None
-            return target.target_device_id
+        target = await self.get_voice_target(device_id)
+        return target.target_device_id if target and target.kind == VOICE_TARGET_KIND_DEVICE else None
 
     async def get_voice_target(self, device_id: str) -> Optional[VoiceTarget]:
         async with self.lock:
-            return self.targets.get(device_id)
+            return await self._get_locked(device_id)
+
+    async def _set_target(self, device_id, *, kind, target_device_id=None, uuid=None):
+        async with self.lock:
+            old = self.targets.get(device_id)
+            if old:
+                await self._remove_locked(device_id, old, "replaced")
+            self.targets[device_id] = VoiceTarget(
+                kind=kind, target_device_id=target_device_id, uuid=uuid,
+                audio_cache_key=f"voice:{device_id}:{uuid4().hex}",
+                expires_at=self.clock() + self.timeout_seconds,
+            )
+            session_logger.info(device_id, "target", f"留言目标已设置: kind={kind}, timeout={self.timeout_seconds}s")
 
     async def set_target(self, device_id: str, target_device_id: str):
-        async with self.lock:
-            self.targets[device_id] = VoiceTarget(
-                kind=VOICE_TARGET_KIND_DEVICE,
-                target_device_id=target_device_id,
-                audio_cache_key=target_device_id,
-            )
+        await self._set_target(device_id, kind=VOICE_TARGET_KIND_DEVICE, target_device_id=target_device_id)
 
     async def set_parent_target(self, device_id: str, uuid: str):
-        async with self.lock:
-            self.targets[device_id] = VoiceTarget(
-                kind=VOICE_TARGET_KIND_PARENT,
-                uuid=uuid,
-                audio_cache_key=f"parent:{device_id}",
-            )
+        await self._set_target(device_id, kind=VOICE_TARGET_KIND_PARENT, uuid=uuid)
 
-    async def remove_target(self, device_id: str):
+    async def claim_target(self, device_id, websocket, session_id):
+        """Bind one pending intent to one recording on one connection."""
         async with self.lock:
-            if device_id in self.targets:
-                del self.targets[device_id]
+            active = await connection_manager.get_connection(device_id)
+            if active is not None and active is not websocket:
+                return None
+            target = await self._get_locked(device_id)
+            if target is None:
+                return None
+            if target.websocket is not None and target.websocket is not websocket:
+                await self._remove_locked(device_id, target, "connection_changed")
+                return None
+            if target.session_id is not None and target.session_id != session_id:
+                await self._remove_locked(device_id, target, "new_recording")
+                return None
+            target.websocket, target.session_id = websocket, session_id
+            target.expires_at = self.clock() + self.timeout_seconds
+            session_logger.info(device_id, session_id, f"留言目标已关联录音: kind={target.kind}")
+            return target
+
+    async def bind_connection(self, device_id, websocket):
+        """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.session_id is not None and target.websocket is not websocket:
+                await self._remove_locked(device_id, target, "connection_changed")
+
+    async def append_audio(self, device_id, target, websocket, session_id, audio):
+        async with self.lock:
+            current = await self._get_locked(device_id)
+            if current is not target or target.websocket is not websocket or target.session_id != session_id:
+                return False
+            target.expires_at = self.clock() + self.timeout_seconds
+            await target_audio_cache.add_audio_data(target.audio_cache_key, audio)
+            return True
+
+    async def take_target(self, device_id, target, websocket, session_id):
+        """Consume the exact recording before slow upload/notification work."""
+        async with self.lock:
+            current = await self._get_locked(device_id)
+            if current is not target or target.websocket is not websocket or target.session_id != session_id:
+                return None
+            await self._remove_locked(device_id, target, "recording_ended", clear_audio=False)
+            return target
+
+    async def remove_target(self, device_id: str, *, expected=None, websocket=None, reason="cancelled"):
+        async with self.lock:
+            target = self.targets.get(device_id)
+            if target is None or (expected is not None and target is not expected):
+                return False
+            if websocket is not None and target.websocket is not websocket:
+                return False
+            await self._remove_locked(device_id, target, reason)
+            return True
 
     async def get_all_targets(self) -> list:
         async with self.lock:
+            for device_id in list(self.targets):
+                await self._get_locked(device_id)
             return list(self.targets.items())
 
 

talkingq-url/services/schedule_session_cleanup.py

@@ -3,6 +3,7 @@ from config import settings
 from services.audio_session import audio_session_manager
 from utils.logger import session_logger
 from services.conversation_history import conversation_history_manager
+from services.device_target_cache import device_target_cache
 
 async def run_in_executor(func, *args):
     loop = asyncio.get_running_loop()
@@ -12,6 +13,7 @@ async def background_cleanup_task():
     try:
         while True:
             await asyncio.sleep(settings.cleanup_interval)
+            await device_target_cache.get_all_targets()
             
             current_time = asyncio.get_running_loop().time()
             audio_sessions_to_clean = []
@@ -54,4 +56,4 @@ async def background_cleanup_task():
     except asyncio.CancelledError:
         session_logger.info("system", "cleanup", "后台清理任务已取消")
     except Exception as e:
-        session_logger.error("system", "cleanup", f"后台清理任务出错: {str(e)}")
+        session_logger.error("system", "cleanup", f"后台清理任务出错: {str(e)}")

talkingq-url/tests/test_forced_ota.py

@@ -146,9 +146,15 @@ async def test_simulated_device_force_ota_dispatches_once_and_completes(monkeypa
 
     await mqtt_service._handle_ota_response(
         device_id,
-        {"status": "updating", "data": {"version": "1.8.5", "progress": 65}},
+        {
+            "msg_id": "003",
+            "status": "updating",
+            "data": {"target_version": "1.8.5", "progress": 65},
+        },
     )
     assert record["update_status"] == "updating"
+    assert record["firmware_version"] == "1.8.5"
+    assert record["target_version"] == "1.8.5"
     assert record["progress"] == 65.0
 
     await mqtt_service._handle_ota_response(

talkingq-url/tests/test_voice_target_lifecycle.py

@@ -0,0 +1,567 @@
+"""Regression for stale message intents swallowing the next AI recording."""
+import asyncio
+import struct
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from unittest.mock import Mock
+
+import pytest
+
+from handlers import audio_packet_parser as parser
+from handlers import audio_session_handler as audio_handler
+from handlers import session_cleanup_handler as cleanup
+from handlers import websocket_message_handler as messages
+from handlers import websocket_handler as endpoint
+from services import device_target_cache as targets_module
+from services.audio_session import AudioSessionManager
+from services.connection_manager import ConnectionManager
+from services.device_target_cache import DeviceTargetCache
+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''):
+    return DEVICE.encode()+b'\0'+sid.encode()+b'\0'+struct.pack('<IBI', 1, kind, len(audio))+audio
+
+
+@pytest.fixture
+def state(monkeypatch):
+    now = [100.0]
+    connections, sessions, buffers = ConnectionManager(), AudioSessionManager(), TargetAudioCache()
+    targets = DeviceTargetCache(timeout_seconds=10, clock=lambda: now[0])
+    started, ended = [], []
+    ws = SimpleNamespace(send_text=AsyncMock(), accept=AsyncMock(), client='local',
+                         client_state=SimpleNamespace(name='CONNECTED'))
+    for module in [messages, endpoint, targets_module]:
+        monkeypatch.setattr(module, 'connection_manager', connections)
+    for module in [messages, cleanup]:
+        monkeypatch.setattr(module, 'device_target_cache', targets)
+    for module in [messages, targets_module]:
+        monkeypatch.setattr(module, 'target_audio_cache', buffers)
+    for module in [messages, cleanup, parser]:
+        monkeypatch.setattr(module, 'audio_session_manager', sessions)
+    monkeypatch.setattr(cleanup, 'task_manager', SimpleNamespace(cancel_session_tasks=AsyncMock()))
+    monkeypatch.setattr(parser, 'get_asr', AsyncMock(side_effect=lambda _: SimpleNamespace(connected=False)))
+    monkeypatch.setattr(parser, 'get_llm', AsyncMock(side_effect=lambda _: SimpleNamespace(close=AsyncMock())))
+    monkeypatch.setattr(parser, 'get_tts', AsyncMock(side_effect=lambda _: SimpleNamespace(close=AsyncMock())))
+
+    async def prepare(session, device_id, sid):
+        started.append(sid)
+        session.asr_service.connected = True
+        session.send_queue = asyncio.Queue()
+
+    async def finish(session, key):
+        chunks = []
+        while not session.send_queue.empty():
+            chunks.append(session.send_queue.get_nowait())
+        ended.append((key[1], b''.join(chunks)))
+
+    monkeypatch.setattr(audio_handler, 'handle_start_packet', prepare)
+    monkeypatch.setattr(audio_handler, 'handle_end_packet', finish)
+    monkeypatch.setattr(messages, 'process_parent_leave_message', AsyncMock())
+    monkeypatch.setattr(messages, 'process_cached_audio', AsyncMock())
+    return SimpleNamespace(now=now, connections=connections, sessions=sessions, buffers=buffers,
+                           targets=targets, ws=ws, started=started, ended=ended)
+
+
+async def send(state, sid, kind, audio=b'', *, ws=None, active=None):
+    return await messages.handle_binary_message(ws or state.ws, DEVICE, 'LOCAL-SERIAL',
+                                                packet(sid, kind, audio), active, None)
+
+
+async def set_target(state, kind):
+    if kind == 'parent':
+        await state.targets.set_parent_target(DEVICE, DEVICE)
+    else:
+        await state.targets.set_target(DEVICE, 'TalkingQ_receiver')
+    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):
+    target = await set_target(state, kind)
+    await send(state, 'ai', 1)
+    assert state.started == []
+    await send(state, 'ai', 0, b'first-audio')
+    await send(state, 'ai', 0, b'-tail')
+    await send(state, 'ai', 2)
+    assert state.started == ['ai']
+    assert state.ended == [('ai', b'first-audio-tail')]
+    assert [c.args[0] for c in state.ws.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
+    messages.process_parent_leave_message.assert_not_awaited()
+    messages.process_cached_audio.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('kind', ['parent', 'device'])
+@pytest.mark.parametrize('outcome', ['success', 'empty', 'error', 'cancelled'])
+async def test_message_terminal_paths_release_only_their_recording(state, kind, outcome):
+    target = await set_target(state, kind)
+    await send(state, 'message', 1)
+    if outcome != 'empty':
+        await send(state, 'message', 4, b'message-audio')
+    processor = messages.process_parent_leave_message if kind == 'parent' else messages.process_cached_audio
+    if outcome == 'error':
+        processor.side_effect = RuntimeError('upload failed')
+        with pytest.raises(RuntimeError, match='upload failed'):
+            await send(state, 'message', 2)
+    else:
+        await send(state, 'message', 3 if outcome == 'cancelled' else 2)
+    assert await state.targets.get_voice_target(DEVICE) is None
+    assert await state.buffers.get_audio_data(target.audio_cache_key) is None
+    assert state.started == []
+    if outcome == 'cancelled':
+        processor.assert_not_awaited()
+        assert [c.args[0] for c in state.ws.send_text.call_args_list][-2:] == ['INTERRUPT_ACK', 'INTERRUPT_PROCESSED']
+    else:
+        processor.assert_awaited_once()
+    await send(state, 'next-ai', 1)
+    await send(state, 'next-ai', 0, b'next-question')
+    await send(state, 'next-ai', 2)
+    assert state.ended == [('next-ai', b'next-question')]
+
+
+@pytest.mark.asyncio
+async def test_cancelled_message_cannot_receive_late_audio_or_end(state):
+    await set_target(state, 'parent')
+    await send(state, 'cancelled', 1)
+    await send(state, 'cancelled', 4, b'old')
+    await send(state, 'cancelled', 3)
+    await set_target(state, 'parent')
+    fresh = await state.targets.get_voice_target(DEVICE)
+    await send(state, 'fresh', 1)
+    await send(state, 'fresh', 4, b'new')
+    await send(state, 'cancelled', 4, b'late')
+    await send(state, 'cancelled', 2)
+    assert await state.targets.get_voice_target(DEVICE) is fresh
+    assert await state.buffers.get_audio_data(fresh.audio_cache_key) == b'new'
+    messages.process_parent_leave_message.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('claimed', [False, True])
+async def test_timeout_clears_pending_or_partial_message_and_allows_ai(state, claimed):
+    target = await set_target(state, 'parent')
+    if claimed:
+        await send(state, 'expired', 1)
+        await send(state, 'expired', 4, b'partial')
+    state.now[0] += 11
+    assert await state.targets.get_voice_target(DEVICE) is None
+    assert await state.buffers.get_audio_data(target.audio_cache_key) is None
+    await send(state, 'ai', 1)
+    await send(state, 'ai', 0, b'question')
+    await send(state, 'ai', 2)
+    assert state.ended == [('ai', b'question')]
+
+
+@pytest.mark.asyncio
+async def test_active_message_refreshes_idle_timeout(state):
+    target = await set_target(state, 'parent')
+    await send(state, 'message', 1)
+    for _ in range(4):
+        state.now[0] += 9
+        await send(state, 'message', 4, b'a')
+    assert await state.targets.get_voice_target(DEVICE) is target
+    assert await state.buffers.get_audio_data(target.audio_cache_key) == b'aaaa'
+
+
+@pytest.mark.asyncio
+async def test_replacement_during_upload_survives_old_recording_cleanup(state):
+    old = await set_target(state, 'parent')
+    await send(state, 'old', 1)
+    await send(state, 'old', 4, b'old-audio')
+    replacement = []
+
+    async def upload(*args):
+        await set_target(state, 'parent')
+        replacement.append(await state.targets.get_voice_target(DEVICE))
+
+    messages.process_parent_leave_message.side_effect = upload
+    await send(state, 'old', 2)
+    assert await state.targets.get_voice_target(DEVICE) is replacement[0]
+    assert replacement[0].audio_cache_key != old.audio_cache_key
+    assert await state.buffers.get_audio_data(old.audio_cache_key) is None
+
+
+@pytest.mark.asyncio
+async def test_new_message_command_during_ai_does_not_hijack_its_end(state):
+    await send(state, 'ai', 1)
+    await send(state, 'ai', 0, b'question')
+    target = await set_target(state, 'parent')
+    await send(state, 'ai', 2)
+    assert state.ended == [('ai', b'question')]
+    assert await state.targets.get_voice_target(DEVICE) is target
+    messages.process_parent_leave_message.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_two_senders_to_same_receiver_use_separate_buffers(state):
+    await state.targets.set_target(DEVICE, 'receiver')
+    await state.targets.set_target('other-sender', 'receiver')
+    a = await state.targets.claim_target(DEVICE, state.ws, 'a')
+    b = await state.targets.claim_target('other-sender', state.ws, 'b')
+    await state.targets.append_audio(DEVICE, a, state.ws, 'a', b'one')
+    await state.targets.append_audio('other-sender', b, state.ws, 'b', b'two')
+    await state.targets.remove_target(DEVICE, expected=a)
+    assert await state.buffers.get_audio_data(b.audio_cache_key) == b'two'
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('claimed', [False, True])
+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:
+        await send(state, 'message', 1)
+        await send(state, 'message', 4, b'partial')
+        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 if claimed else target)
+    assert await state.buffers.get_audio_data(target.audio_cache_key) is None
+    assert await state.sessions.get_all_sessions() == []
+
+
+@pytest.mark.asyncio
+async def test_old_socket_cleanup_and_late_packets_preserve_new_connection(state, monkeypatch):
+    old = state.ws
+    new = SimpleNamespace(send_text=AsyncMock())
+
+    async def authenticate(ws):
+        await state.connections.add_connection(DEVICE, old)
+        return True, DEVICE, 'LOCAL'
+
+    async def replacement_connects(*args):
+        await state.connections.add_connection(DEVICE, new)
+        await set_target(state, 'parent')
+        await send(state, 'new', 1, ws=new)
+        await send(state, 'new', 4, b'new-audio', ws=new)
+        await send(state, 'old-late-start', 1, ws=old)
+        await send(state, 'old-late-end', 2, ws=old)
+
+    monkeypatch.setattr(endpoint, 'authenticate_websocket', authenticate)
+    monkeypatch.setattr(endpoint, 'handle_websocket_messages', replacement_connects)
+    await endpoint.websocket_endpoint(old)
+    target = await state.targets.get_voice_target(DEVICE)
+    assert target.websocket is new
+    assert await state.buffers.get_audio_data(target.audio_cache_key) == b'new-audio'
+    assert await state.connections.get_connection(DEVICE) is new
+    assert await state.sessions.get_session((DEVICE, 'new')) is not None
+    cleanup.task_manager.cancel_session_tasks.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+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
+
+
+@pytest.mark.asyncio
+async def test_reconnect_discards_old_partial_message(state):
+    await state.connections.add_connection(DEVICE, state.ws)
+    old = await set_target(state, 'parent')
+    await send(state, 'old', 1)
+    await send(state, 'old', 4, b'old')
+    new = SimpleNamespace(send_text=AsyncMock())
+    await state.connections.add_connection(DEVICE, new)
+    await state.targets.bind_connection(DEVICE, new)
+    assert await state.targets.get_voice_target(DEVICE) is None
+    assert await state.buffers.get_audio_data(old.audio_cache_key) is None
+    await send(state, 'new-ai', 1, ws=new)
+    await send(state, 'new-ai', 0, b'reconnected-question', ws=new)
+    await send(state, 'new-ai', 2, ws=new)
+    assert state.ended == [('new-ai', b'reconnected-question')]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('outcome', ['success', 'empty', 'upload_error'])
+async def test_real_parent_message_processing_keeps_notifications_and_clears_state(state, monkeypatch, outcome):
+    monkeypatch.setattr(messages, 'process_parent_leave_message', PROCESS_PARENT_MESSAGE)
+    upload = AsyncMock(return_value=SimpleNamespace(file_key='local-test/audio.wav'))
+    create = AsyncMock(return_value=(SimpleNamespace(child_name='local-test'), None, []))
+    notify = Mock()
+    if outcome == 'upload_error':
+        upload.side_effect = RuntimeError('simulated upload failure')
+    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)
+    await state.connections.add_connection(DEVICE, state.ws)
+    target = await set_target(state, 'parent')
+    await send(state, 'parent', 1)
+    if outcome != 'empty':
+        await send(state, 'parent', 4, b'\x01\x00'*400)
+    await send(state, 'parent', 2)
+    assert await state.targets.get_voice_target(DEVICE) is None
+    assert await state.buffers.get_audio_data(target.audio_cache_key) is None
+    if outcome == 'success':
+        upload.assert_awaited_once()
+        create.assert_awaited_once()
+        assert create.call_args.kwargs['audio_content'].startswith(b'RIFF')
+        notify.assert_called_once()
+        assert state.ws.send_text.call_args.args[0].startswith('PROMPT_SOUND_URL:')
+    else:
+        create.assert_not_awaited()
+        notify.assert_not_called()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('first_kind', [0, 4])
+async def test_recording_cannot_switch_between_ai_and_message_after_audio(state, first_kind):
+    target = await set_target(state, 'parent')
+    await send(state, 'mixed', 1)
+    await send(state, 'mixed', first_kind, b'first')
+    await send(state, 'mixed', 4 if first_kind == 0 else 0, b'wrong-mode')
+    if first_kind == 0:
+        await send(state, 'mixed', 2)
+        assert state.ended == [('mixed', b'first')]
+        messages.process_parent_leave_message.assert_not_awaited()
+    else:
+        assert await state.buffers.get_audio_data(target.audio_cache_key) == b'first'
+        assert state.started == []
+        await send(state, 'mixed', 2)
+        messages.process_parent_leave_message.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_new_recording_recovers_abandoned_message_without_reusing_audio(state):
+    abandoned = await set_target(state, 'parent')
+    await send(state, 'abandoned', 1)
+    await send(state, 'abandoned', 4, b'abandoned-audio')
+    await send(state, 'new-ai', 1)
+    await send(state, 'new-ai', 0, b'question')
+    await send(state, 'abandoned', 2, active=(DEVICE, 'new-ai'))
+    await send(state, 'new-ai', 2)
+    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)