fix: 未等待服务端残留状态问题

main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py

@@ -149,6 +149,7 @@ class TTSProvider(TTSProviderBase):
         self.access_token = config.get("access_token")
         self.cluster = config.get("cluster")
         self.resource_id = config.get("resource_id")
+        self.activate_session = False
         if config.get("private_voice"):
             self.voice = config.get("private_voice")
         else:
@@ -180,7 +181,7 @@ class TTSProvider(TTSProviderBase):
             raise
 
     async def _ensure_connection(self):
-        """建立新的WebSocket连接"""
+        """建立新的WebSocket连接，并启动监听任务（仅第一次）"""
         try:
             if self.ws:
                 logger.bind(tag=TAG).info(f"使用已有链接...")
@@ -196,6 +197,12 @@ class TTSProvider(TTSProviderBase):
                 self.ws_url, additional_headers=ws_header, max_size=1000000000
             )
             logger.bind(tag=TAG).info("WebSocket连接建立成功")
+            
+            # 连接建立成功后，启动监听任务
+            if self._monitor_task is None or self._monitor_task.done():
+                logger.bind(tag=TAG).info("启动监听任务...")
+                self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
+            
             return self.ws
         except Exception as e:
             logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}")
@@ -315,20 +322,22 @@ class TTSProvider(TTSProviderBase):
     async def start_session(self, session_id):
         logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
         try:       
-            # 会话开始时检测上个会话的监听状态
+            # 等待上一个会话结束，最多等待3次
-            if (
+            for _ in range(3):
-                self._monitor_task is not None
+                if not self.activate_session:
-                and isinstance(self._monitor_task, Task)
+                    break
-                and not self._monitor_task.done()
+                logger.bind(tag=TAG).debug(f"等待上一个会话结束...")
-            ):
+                await asyncio.sleep(0.1)
-                logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务和连接...")
+            else:
+                # 等待超时，强制清除连接状态
+                logger.bind(tag=TAG).debug("等待上一个会话超时，清除连接状态...")
                 await self.close()
             
-            # 建立新连接
+            # 设置会话激活标志
-            await self._ensure_connection()
+            self.activate_session = True
             
-            # 启动监听任务
+            # 确保连接建立
-            self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
+            await self._ensure_connection()
 
             header = Header(
                 message_type=FULL_CLIENT_REQUEST,
@@ -365,17 +374,6 @@ class TTSProvider(TTSProviderBase):
                 await self.send_event(self.ws, header, optional, payload)
                 logger.bind(tag=TAG).info("会话结束请求已发送")
 
-                # 等待监听任务完成
-                if self._monitor_task:
-                    try:
-                        await self._monitor_task
-                    except Exception as e:
-                        logger.bind(tag=TAG).error(
-                            f"等待监听任务完成时发生错误: {str(e)}"
-                        )
-                    finally:
-                        self._monitor_task = None
-
         except Exception as e:
             logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
             # 确保清理资源
@@ -405,6 +403,7 @@ class TTSProvider(TTSProviderBase):
 
     async def close(self):
         """资源清理方法"""
+        self.activate_session = False
         # 取消监听任务
         if self._monitor_task:
             try:
@@ -424,9 +423,8 @@ class TTSProvider(TTSProviderBase):
             self.ws = None
 
     async def _start_monitor_tts_response(self):
-        """监听TTS响应"""
+        """监听TTS响应 - 长期运行"""
         try:
-            session_finished = False  # 标记会话是否正常结束
             while not self.conn.stop_event.is_set():
                 try:
                     # 确保 `recv()` 运行在同一个 event loop
@@ -434,10 +432,17 @@ class TTSProvider(TTSProviderBase):
                     res = self.parser_response(msg)
                     self.print_response(res, "send_text res:")
 
+                    # 只处理当前活跃会话的响应
+                    if res.optional.sessionId and self.conn.sentence_id != res.optional.sessionId:
+                        # 如果是会话结束相关事件，即使会话ID不匹配也要重置状态
+                        if res.optional.event in [EVENT_SessionCanceled, EVENT_SessionFailed, EVENT_SessionFinished]:
+                            logger.bind(tag=TAG).debug(f"收到残余下行结束响应重置会话状态～～")
+                            self.activate_session = False
+                        continue
+
                     if res.optional.event == EVENT_SessionCanceled:
                         logger.bind(tag=TAG).debug(f"释放服务端资源成功～～")
-                        session_finished = True
+                        self.activate_session = False
-                        break
                     elif res.optional.event == EVENT_TTSSentenceStart:
                         json_data = json.loads(res.payload.decode("utf-8"))
                         self.tts_text = json_data.get("text", "")
@@ -454,9 +459,8 @@ class TTSProvider(TTSProviderBase):
                         logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")
                     elif res.optional.event == EVENT_SessionFinished:
                         logger.bind(tag=TAG).debug(f"会话结束～～")
+                        self.activate_session = False
                         self._process_before_stop_play_files()
-                        session_finished = True
-                        break
                 except websockets.ConnectionClosed:
                     logger.bind(tag=TAG).warning("WebSocket连接已关闭")
                     break
@@ -466,8 +470,8 @@ class TTSProvider(TTSProviderBase):
                     )
                     traceback.print_exc()
                     break
-            # 仅在连接异常时才关闭
+            # 连接异常时关闭WebSocket
-            if not session_finished and self.ws:
+            if self.ws:
                 try:
                     await self.ws.close()
                 except:
@@ -513,7 +517,7 @@ class TTSProvider(TTSProviderBase):
     def read_res_content(self, res: bytes, offset: int):
         content_size = int.from_bytes(res[offset : offset + 4], "big", signed=True)
         offset += 4
-        content = str(res[offset : offset + content_size])
+        content = res[offset : offset + content_size].decode('utf-8')
         offset += content_size
         return content, offset