diff --git a/talkingq-url/config.py b/talkingq-url/config.py index acdf735..6778ff2 100644 --- a/talkingq-url/config.py +++ b/talkingq-url/config.py @@ -57,6 +57,17 @@ class Settings(BaseSettings): aliyun_api_key: str = Field(default="", validation_alias="ALIYUN_API_KEY") aliyun_vocabulary_id: str = Field(default="", validation_alias="ALIYUN_VOCABULARY_ID") + aliyun_asr_model: str = Field(default="qwen-audio-3.1-asr-flash-streaming", validation_alias="ALIYUN_ASR_MODEL") + aliyun_asr_url: str = Field( + default="wss://dashscope.aliyuncs.com/api-ws/v1/inference", + validation_alias="ALIYUN_ASR_URL", + ) + # Separate from the retired Gummy model's vocabulary ID. + aliyun_asr_vocabulary_id: str = Field(default="", validation_alias="ALIYUN_ASR_VOCABULARY_ID") + asr_connect_timeout_seconds: float = Field(default=5.0, gt=0, le=30, validation_alias="ASR_CONNECT_TIMEOUT_SECONDS") + asr_send_timeout_seconds: float = Field(default=5.0, gt=0, le=30, validation_alias="ASR_SEND_TIMEOUT_SECONDS") + asr_final_timeout_seconds: float = Field(default=8.0, gt=0, le=30, validation_alias="ASR_FINAL_TIMEOUT_SECONDS") + asr_queue_drain_timeout_seconds: float = Field(default=5.0, gt=0, le=30, validation_alias="ASR_QUEUE_DRAIN_TIMEOUT_SECONDS") assets_dir: str = Field(default="assets", validation_alias="ASSETS_DIR") session_timeout: int = Field(default=600, validation_alias="SESSION_TIMEOUT") diff --git a/talkingq-url/docs/qwen-asr-rollout.md b/talkingq-url/docs/qwen-asr-rollout.md new file mode 100644 index 0000000..b20d558 --- /dev/null +++ b/talkingq-url/docs/qwen-asr-rollout.md @@ -0,0 +1,23 @@ +# 千问 ASR 切换与部署验证 + +默认模型改为 `qwen-audio-3.1-asr-flash-streaming`，通过异步 DashScope recognition WebSocket 接收现有 16 kHz MP3 音频。`ALIYUN_ASR_MODEL` 可覆盖模型；`ALIYUN_ASR_URL` 可覆盖接口。旧 Gummy 热词表不自动复用，新热词表需显式配置 `ALIYUN_ASR_VOCABULARY_ID`。 + +结束录音时先排空发送队列和尾包，再等待最终结果；队列溢出、识别错误、未完成结果不进入正常回复链路。模型拒绝、启动/发送/结束等待均有超时。LLM/TTS 预热连接注册中断清理，覆盖连接完成后中断、连接期间中断、任务取消和已中断会话。 + +2026-10-12 本地验证： + +- 后端 tests：215 项通过，回归运行阻止非本机网络连接。 +- 真实千问 ASR：10/10 轮通过。 +- 原有 `/ws`、角色/语言选择、真实千问 ASR、火山 LLM、MiniMax TTS、音频 URL 下发、HTTP 下载/Range、MP3 解码及 MySQL 历史落库：10/10 轮通过（8 轮有声对话、2 轮静音提示），共 43 段生成回复和 2 段提示音。 +- 发现并修复录音中断时 LLM/TTS 客户端未关闭的问题；修复后两个角色的中断及后续恢复共 4/4 通过。 + +使用 `tests/fixtures/asr/manifest.json` 中的合成语音复现 ASR： + +```sh +.venv/bin/python scripts/smoke_fun_asr.py tests/fixtures/asr/manifest.json --output output/asr-retest.json +.venv/bin/python -m pytest tests -q -o asyncio_default_fixture_loop_scope=function +``` + +完整对话脚本 `scripts/smoke_ai_dialog.py` 要求独立 MySQL 监听 `127.0.0.1:13316`，并使用测试数据库 `qwen_dialog_test`、临时设备和 `127.0.0.1:18086` 服务。它使用真实云端 API，会产生用量；本地缺少 MiniMax key 时读取已有 `test/minimax.env`，不修改 `.env`。需要 macOS `afconvert` 做音频解码。运行 `--interrupt-only` 复测中断与下一轮恢复。 + +以上不等于实物设备麦克风/扬声器验收，也未验证回复播放中的打断。部署只更新六个运行文件，不迁移数据库，不替换 `.env`、音频或 OTA 固件；线上测试另存执行证据。 diff --git a/talkingq-url/handlers/audio_session_handler.py b/talkingq-url/handlers/audio_session_handler.py index fb8ee94..4a65f45 100644 --- a/talkingq-url/handlers/audio_session_handler.py +++ b/talkingq-url/handlers/audio_session_handler.py @@ -72,6 +72,8 @@ async def handle_start_packet(session: AudioSession, device_id: str, session_id: from handlers.service_connection_handler import prepare_asr_service await prepare_asr_service(session, device_id, session_id) + if not session.asr_service.connected: + return from handlers.service_connection_handler import ( prepare_llm_service, @@ -122,6 +124,7 @@ async def handle_audio_packet(session: AudioSession, audio_data: bytes): if session.send_queue.qsize() < 100: # 设置合理的上限 await session.send_queue.put(audio_data) else: + session.asr_error = "音频队列溢出，录音不完整" session_logger.warning("unknown", "unknown", "音频队列过大，丢弃部分数据") @@ -130,17 +133,12 @@ async def handle_end_packet(session: AudioSession, session_key: tuple): device_id, session_id = session_key session.end_time = time.perf_counter() session_logger.info(device_id, session_id, "会话结束") - if session.send_task: - session.send_task.cancel() - try: - await session.send_task - except asyncio.CancelledError: - pass from handlers.service_connection_handler import end_asr_session from handlers.transcription_handler import handle_transcription await end_asr_session(session) - await handle_transcription(session, device_id, session_id) + if not interrupt_handler.is_interrupted(session_key): + await handle_transcription(session, device_id, session_id) if session.llm_service and hasattr(session.llm_service, "close"): await session.llm_service.close() if session.tts_service: diff --git a/talkingq-url/handlers/service_connection_handler.py b/talkingq-url/handlers/service_connection_handler.py index 409dc41..23732e5 100644 --- a/talkingq-url/handlers/service_connection_handler.py +++ b/talkingq-url/handlers/service_connection_handler.py @@ -4,9 +4,34 @@ from services.task_manager import task_manager from utils.logger import session_logger from services.interrupt_handler import interrupt_handler import time +from config import settings + + +async def _connect_with_interrupt_cleanup(service, session_key): + """Own warmup connections even when interruption races with connect().""" + if interrupt_handler.is_interrupted(session_key): + return False + await interrupt_handler.register_cleanup_handler( + session_key, service.close, + priority=interrupt_handler.PRIORITY["NETWORK_CONNECTION"], + ) + try: + await service.connect() + if interrupt_handler.is_interrupted(session_key): + await service.close() + return False + return True + except asyncio.CancelledError: + await service.close() + raise + except Exception: + await service.close() + raise async def prepare_asr_service(session, device_id, session_id): + session.asr_error = None + session.asr_service.sample_rate = session.sample_rate if hasattr(session.asr_service, "device_id"): session.asr_service.device_id = device_id if hasattr(session.asr_service, "session_id"): @@ -20,7 +45,8 @@ async def prepare_llm_service(session, device_id, session_id, selected_role): if hasattr(session.llm_service, "session_id"): session.llm_service.session_id = session_id if hasattr(session.llm_service, "connect"): - await session.llm_service.connect() + if not await _connect_with_interrupt_cleanup(session.llm_service, (device_id, session_id)): + return from services.conversation_history import conversation_history_manager from config import settings @@ -58,12 +84,29 @@ async def establish_asr_connection(session: AudioSession): ) await session.asr_service.connect() if not session.asr_service.connected: - session_logger.error("unknown", "unknown", "语音识别 WebSocket 连接未建立") - return + raise RuntimeError("语音识别 WebSocket 连接未建立") + if session.device_id and session.session_id: + if interrupt_handler.is_interrupted((session.device_id, session.session_id)): + raise RuntimeError("连接期间会话已中断") + await interrupt_handler.register_cleanup_handler( + (session.device_id, session.session_id), session.asr_service.close, + priority=interrupt_handler.PRIORITY["NETWORK_CONNECTION"], + ) + await interrupt_handler.register_cleanup_handler( + (session.device_id, session.session_id), session.cancel_send_task, + priority=interrupt_handler.PRIORITY["TASK_CANCELLATION"], + ) session_logger.info( "unknown", "unknown", f"语音识别连接已建立，采样率: {session.sample_rate}Hz" ) + except asyncio.CancelledError: + await session.cancel_send_task() + await session.asr_service.close() + raise except Exception as e: + session.asr_error = f"{type(e).__name__}: {e}" + await session.cancel_send_task() + await session.asr_service.close() session_logger.error( "unknown", "unknown", f"建立语音识别 WebSocket 连接失败: {e}", exc_info=True ) @@ -80,7 +123,10 @@ async def establish_tts_connection(session: AudioSession): ) return - await session.tts_service.connect() + if not await _connect_with_interrupt_cleanup( + session.tts_service, (session.device_id, session.session_id) + ): + return if not session.tts_service.connected: session_logger.error( @@ -113,19 +159,24 @@ async def send_audio_task(session: AudioSession): session.device_id, session.session_id, "发送音频任务被中断" ) break - audio_data = await session.send_queue.get() - if audio_data is None: # 结束信号 - break - - await session.asr_service.send_audio(audio_data) + queue = session.send_queue + audio_data = await queue.get() + try: + if audio_data is None: # Drain all preceding audio before finish-task. + break + await session.asr_service.send_audio(audio_data) + finally: + queue.task_done() except asyncio.CancelledError: + session.asr_error = session.asr_error or "发送音频任务被取消" session_logger.info( session.device_id or "unknown", session.session_id or "unknown", "发送音频数据任务被取消", ) except Exception as e: + session.asr_error = str(e) session_logger.error( session.device_id or "unknown", session.session_id or "unknown", @@ -135,51 +186,51 @@ async def send_audio_task(session: AudioSession): async def end_asr_session(session: AudioSession): - - if session.asr_service.connected: - try: - asr_end_start_time = time.perf_counter() - await session.asr_service.send_end() - await session.asr_service.receive_results() - session.transcript = session.asr_service.transcript - asr_end_elapsed = time.perf_counter() - asr_end_start_time - session_logger.info( - session.device_id or "unknown", - session.session_id or "unknown", - f"ASR结束延迟: 从发送结束包到获取最终结果耗时 {asr_end_elapsed:.2f} 秒", + """Drain device audio, collect the complete transcript, always release ASR.""" + session.transcript = "" + try: + session_key = (session.device_id, session.session_id) + if interrupt_handler.is_interrupted(session_key): + raise RuntimeError("识别会话已中断") + if session.send_task and not session.send_task.done(): + await session.send_queue.put(None) + await asyncio.wait_for( + asyncio.shield(session.send_task), + timeout=settings.asr_queue_drain_timeout_seconds, ) - except Exception as e: - session_logger.error( - session.device_id or "unknown", - session.session_id or "unknown", - f"ASR结束处理出错: {e}", - exc_info=True, - ) - finally: - if session.send_task: - session.send_task.cancel() - try: - await asyncio.wait_for(session.send_task, timeout=2.0) - except (asyncio.CancelledError, asyncio.TimeoutError): - pass - except Exception as e: - session_logger.error( - session.device_id or "unknown", - session.session_id or "unknown", - f"取消发送任务时出错: {e}", - ) - finally: - session.send_task = None - - await session.asr_service.close() - session_logger.info( - session.device_id or "unknown", - session.session_id or "unknown", - "语音识别 WebSocket 连接和 ClientSession 已关闭", - ) - else: + if session.asr_error: + raise RuntimeError(session.asr_error) + if session.send_queue and not session.send_queue.empty(): + raise RuntimeError("发送任务提前退出，音频未完整发送") + if interrupt_handler.is_interrupted(session_key): + raise RuntimeError("识别会话已中断") + asr_end_start_time = time.perf_counter() + await session.asr_service.send_end() + if interrupt_handler.is_interrupted(session_key): + raise RuntimeError("识别会话已中断") + session.transcript = await session.asr_service.receive_results() + session_logger.info( + session.device_id or "unknown", session.session_id or "unknown", + f"ASR结束延迟: {time.perf_counter() - asr_end_start_time:.2f} 秒", + ) + except Exception as e: + session.asr_error = f"{type(e).__name__}: {e}" + session.transcript = "" session_logger.error( - session.device_id or "unknown", - session.session_id or "unknown", - "语音识别 WebSocket 连接未建立", + session.device_id or "unknown", session.session_id or "unknown", + f"ASR结束处理出错: {session.asr_error}", + ) + finally: + if session.send_task: + session.send_task.cancel() + try: + await asyncio.wait_for(session.send_task, timeout=2.0) + except (asyncio.CancelledError, asyncio.TimeoutError): + pass + finally: + session.send_task = None + await session.asr_service.close() + session_logger.info( + session.device_id or "unknown", session.session_id or "unknown", + "语音识别 WebSocket 连接和 ClientSession 已关闭", ) diff --git a/talkingq-url/implementations/aliyun_asr.py b/talkingq-url/implementations/aliyun_asr.py index 564f783..221782e 100644 --- a/talkingq-url/implementations/aliyun_asr.py +++ b/talkingq-url/implementations/aliyun_asr.py @@ -1,8 +1,11 @@ +"""Async Qwen-Audio/Fun-ASR adapter using the DashScope recognition protocol.""" + import asyncio -import threading -from dashscope.audio.asr import TranslationRecognizerChat -from dashscope.audio.asr import TranscriptionResult -from dashscope.audio.asr import TranslationRecognizerCallback +import json +import uuid + +import aiohttp + from config import settings from interfaces.asr import ASR from implementations.base_service import BaseService @@ -10,333 +13,248 @@ from services.interrupt_handler import interrupt_handler from utils.logger import session_logger -class GummyCallback(TranslationRecognizerCallback): - def __init__(self, device_id, session_id, asr_instance=None): - self.device_id = device_id - self.session_id = session_id - self.asr_instance = asr_instance - self.transcript = "" - self.transcript_event = asyncio.Event() - self.received_result = False - self.error_message = None - self.is_closed = False - self.session_end = False - self.reconnect_event = threading.Event() - self.is_rate_limit_error = False - - def on_open(self) -> None: - session_logger.info(self.device_id, self.session_id, "Gummy ASR连接建立成功") - - def on_event( - self, - request_id, - transcription_result: TranscriptionResult, - translation_result, - usage, - ) -> None: - if transcription_result: - if transcription_result.is_sentence_end: - self.transcript = transcription_result.text - self.received_result = True - session_logger.info( - self.device_id, - self.session_id, - f"Gummy ASR完整句子: {self.transcript}", - ) - self.transcript_event.set() - else: - self.transcript = transcription_result.text - - def on_complete(self) -> None: - session_logger.info(self.device_id, self.session_id, "Gummy ASR识别完成") - self.session_end = True - self.transcript_event.set() - - def on_error(self, result) -> None: - self.error_message = str(result) - error_str = str(result).lower() - self.is_rate_limit_error = "throttling" in error_str or "rate limit" in error_str - - if self.is_rate_limit_error: - session_logger.warning( - self.device_id, - self.session_id, - f"Gummy ASR限流错误，触发重连信号: {self.error_message}" - ) - self.reconnect_event.set() - if self.asr_instance: - self.asr_instance.need_reconnect = True - else: - session_logger.error( - self.device_id, self.session_id, f"Gummy ASR错误: {self.error_message}" - ) - self.transcript_event.set() - - def on_close(self) -> None: - session_logger.info(self.device_id, self.session_id, "Gummy ASR连接已关闭") - self.is_closed = True - self.transcript_event.set() +class ASRError(RuntimeError): + pass class AliyunASR(ASR, BaseService): def __init__(self, selected_role=None): BaseService.__init__(self, "asr", selected_role) self.api_key = settings.aliyun_api_key - self.vocabulary_id = settings.aliyun_vocabulary_id - self.connected = False - self.transcript = "" + self.model = settings.aliyun_asr_model + self.url = settings.aliyun_asr_url + # Gummy vocabulary IDs are model-specific; never reuse them implicitly. + self.vocabulary_id = settings.aliyun_asr_vocabulary_id + self.sample_rate = 16000 self.device_id = "unknown" self.session_id = "unknown" - self.sample_rate = 16000 + self.connected = False + self.transcript = "" + self.error = None + self.task_id = None self.audio_buffer = bytearray() - self.translator = None - self.callback = None self.buffer_size = 3200 - self.max_end_silence = 2000 - self.max_retries = 3 - self.retry_delay = 1.0 - self.need_reconnect = False + self._client = None + self._websocket = None + self._receiver = None + self._ready = asyncio.Event() + self._finished = asyncio.Event() + self._sentences = {} + self._pending = set() + self._complete = False + self._ending = False + self._closing = False + self._end_lock = asyncio.Lock() + self._close_lock = asyncio.Lock() + + def _fail(self, message): + if self.error is None: + self.error = ASRError(message) + session_logger.error( + self.device_id, self.session_id, + f"ASR失败 model={self.model} task_id={self.task_id}: {message}", + ) + self.connected = False + self.transcript = "" + self._ready.set() + self._finished.set() + + def _check_error(self): + if self.error: + raise self.error + + def _command(self, action, payload): + return { + "header": {"action": action, "task_id": self.task_id, "streaming": "duplex"}, + "payload": payload, + } async def connect(self): + if self.connected: + return + await self.close() + self._closing = False + self._ending = False + self._complete = False + self.error = None + self.transcript = "" + self.audio_buffer.clear() + self._sentences.clear() + self._pending.clear() + self._ready.clear() + self._finished.clear() + self.task_id = uuid.uuid4().hex + parameters = { + "format": "mp3", + "sample_rate": self.sample_rate, + "semantic_punctuation_enabled": False, + "max_sentence_silence": 2000, + "heartbeat": True, + } + if self.vocabulary_id: + parameters["vocabulary_id"] = self.vocabulary_id try: - if self.connected and self.translator: - session_logger.info( - self.device_id, self.session_id, "Gummy ASR连接已存在，复用当前连接" - ) - return - - self.transcript = "" - - self.callback = GummyCallback(self.device_id, self.session_id, self) - - self.translator = TranslationRecognizerChat( - model="gummy-chat-v1", - format="mp3", - sample_rate=self.sample_rate, - callback=self.callback, - transcription_enabled=True, - source_language="auto", - semantic_punctuation_enabled=False, - max_end_silence=self.max_end_silence, - api_key=self.api_key, - vocabulary_id=self.vocabulary_id, + if not self.api_key: + raise ASRError("未配置 ALIYUN_API_KEY") + # Native async I/O allows cancellation and bounded cleanup; the old + # SDK's stop() joins a worker thread without a timeout. + self._client = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=None, connect=settings.asr_connect_timeout_seconds), ) - - self.translator.start() - self.callback.transcript_event.clear() - self.connected = True - self.need_reconnect = False - - except Exception as e: - session_logger.error( - self.device_id, self.session_id, f"建立Gummy ASR连接失败: {str(e)}" + self._websocket = await asyncio.wait_for( + self._client.ws_connect( + self.url, headers={"Authorization": f"Bearer {self.api_key}"}, + heartbeat=20, + ), + timeout=settings.asr_connect_timeout_seconds, ) - await self._cleanup_connection() - - async def _cleanup_connection(self): - if self.translator: - try: - self.translator.stop() - except: - pass - self.translator = None - self.callback = None - self.connected = False - - async def _handle_reconnect(self): - if not self.need_reconnect: - return True - - session_logger.warning( - self.device_id, - self.session_id, - "检测到限流错误，准备重连..." - ) - - await self._cleanup_connection() - - current_delay = self.retry_delay - for attempt in range(self.max_retries): + self._receiver = asyncio.create_task(self._receive(), name=f"asr:{self.task_id}") + await asyncio.wait_for(self._websocket.send_json(self._command("run-task", { + "task_group": "audio", "task": "asr", "function": "recognition", + "model": self.model, "parameters": parameters, "input": {}, + })), timeout=settings.asr_send_timeout_seconds) + await asyncio.wait_for(self._ready.wait(), timeout=settings.asr_connect_timeout_seconds) + self._check_error() session_logger.info( - self.device_id, - self.session_id, - f"重连尝试 {attempt + 1}/{self.max_retries}，等待{current_delay}秒" + self.device_id, self.session_id, + f"ASR任务已就绪 model={self.model} task_id={self.task_id}", ) - await asyncio.sleep(current_delay) - - try: - await self.connect() - if self.connected: - session_logger.info( - self.device_id, - self.session_id, - "重连成功" - ) - return True - except Exception as e: - session_logger.error( - self.device_id, - self.session_id, - f"重连失败: {str(e)}" - ) - - current_delay *= 2 - self.retry_delay = current_delay - - session_logger.error( - self.device_id, - self.session_id, - "重连失败，已达到最大重试次数" - ) - self.need_reconnect = False - return False + except asyncio.CancelledError: + await self.close() + raise + except Exception as exc: + self._fail(f"启动失败: {type(exc).__name__}: {exc}") + await self.close() + raise self.error from exc + + async def _receive(self): + try: + async for message in self._websocket: + if message.type == aiohttp.WSMsgType.ERROR: + raise ASRError("识别 WebSocket 传输错误") + if message.type != aiohttp.WSMsgType.TEXT: + continue + event = json.loads(message.data) + header = event.get("header", {}) + if header.get("task_id") != self.task_id: + continue + kind = header.get("event") + if kind == "task-started": + self.connected = True + self._ready.set() + elif kind == "result-generated": + sentence = event.get("payload", {}).get("output", {}).get("sentence", {}) + if not sentence or sentence.get("heartbeat"): + continue + key = sentence.get("sentence_id", sentence.get("begin_time")) + if key is None: + raise ASRError("识别结果缺少句子编号和起始时间") + if sentence.get("sentence_end"): + self._sentences[key] = sentence.get("text", "") + self._pending.discard(key) + elif sentence.get("text") and key not in self._sentences: + self._pending.add(key) + elif kind == "task-finished": + if not self._ending: + self._fail("识别任务提前结束，设备音频尚未发送完毕") + return + self._complete = True + self.connected = False + self._finished.set() + return + elif kind == "task-failed": + self._fail(f"{header.get('error_code', 'Unknown')}: {header.get('error_message', '')}") + return + except asyncio.CancelledError: + raise + except Exception as exc: + self._fail(f"接收失败: {type(exc).__name__}: {exc}") + finally: + if not self._complete and not self._closing and not self.error: + self._fail("识别连接提前关闭，未收到 task-finished") + + async def _send_bytes(self, data): + self._check_error() + try: + await asyncio.wait_for( + self._websocket.send_bytes(data), timeout=settings.asr_send_timeout_seconds, + ) + except Exception as exc: + self._fail(f"发送音频失败: {type(exc).__name__}: {exc}") + raise self.error from exc async def send_audio(self, audio_data: bytes): - if self.need_reconnect: - reconnected = await self._handle_reconnect() - if not reconnected: - session_logger.error(self.device_id, self.session_id, "Gummy ASR重连失败，无法发送音频") - return - - if not self.connected or not self.translator: - session_logger.error(self.device_id, self.session_id, "Gummy ASR连接未建立") - return - - try: - self.audio_buffer.extend(audio_data) - - while len(self.audio_buffer) >= self.buffer_size: - if self.need_reconnect: - session_logger.info( - self.device_id, - self.session_id, - "检测到限流错误，暂停发送音频，准备重连" - ) - reconnected = await self._handle_reconnect() - if not reconnected: - return - if not self.connected or not self.translator: - return - continue - - chunk = bytes(self.audio_buffer[: self.buffer_size]) - self.audio_buffer = self.audio_buffer[self.buffer_size :] - - if self.device_id and self.session_id: - session_key = (self.device_id, self.session_id) - if interrupt_handler.is_interrupted(session_key): - session_logger.info( - self.device_id, - self.session_id, - "检测到中断，停止发送音频数据", - ) - return - - try: - self.translator.send_audio_frame(chunk) - except Exception as e: - session_logger.error( - self.device_id, - self.session_id, - f"发送音频帧到Gummy ASR出错: {str(e)}", - ) - - except Exception as e: - session_logger.error( - self.device_id, self.session_id, f"发送音频到Gummy ASR出错: {str(e)}" - ) + self._check_error() + if interrupt_handler.is_interrupted((self.device_id, self.session_id)): + raise ASRError("会话已中断") + if not self.connected or self._ending: + raise ASRError("识别任务未就绪或已结束") + self.audio_buffer.extend(audio_data) + while len(self.audio_buffer) >= self.buffer_size: + chunk = bytes(self.audio_buffer[:self.buffer_size]) + await self._send_bytes(chunk) + del self.audio_buffer[:self.buffer_size] async def send_end(self): - if not self.connected or not self.translator: - session_logger.error( - self.device_id, self.session_id, "Gummy ASR连接未建立，无法发送结束信号" - ) - return - - try: - if len(self.audio_buffer) > 0: - chunk = bytes(self.audio_buffer) - self.audio_buffer = bytearray() - self.translator.send_audio_frame(chunk) - - session_logger.info( - self.device_id, - self.session_id, - "客户端发送结束包，等待Gummy ASR处理完毕", - ) - - loop = asyncio.get_running_loop() - await loop.run_in_executor(None, lambda: self._safe_stop_translator()) - - except Exception as e: - session_logger.error( - self.device_id, - self.session_id, - f"发送结束信号到Gummy ASR出错: {str(e)}", - ) + async with self._end_lock: + self._check_error() + if self._complete: + return + if not self.connected: + raise ASRError("识别任务未就绪") + self._ending = True + try: + if self.audio_buffer: + await self._send_bytes(bytes(self.audio_buffer)) + self.audio_buffer.clear() + await asyncio.wait_for( + self._websocket.send_json(self._command("finish-task", {"input": {}})), + timeout=settings.asr_send_timeout_seconds, + ) + await asyncio.wait_for(self._finished.wait(), timeout=settings.asr_final_timeout_seconds) + self._check_error() + except asyncio.CancelledError: + await self.close() + raise + except Exception as exc: + self._fail(f"结束识别失败: {type(exc).__name__}: {exc}") + raise self.error from exc async def receive_results(self): - if not self.connected: - session_logger.error( - self.device_id, self.session_id, "Gummy ASR连接未建立，无法接收结果" - ) - return - - try: - try: - await asyncio.wait_for( - self.callback.transcript_event.wait(), timeout=0.5 - ) - except asyncio.TimeoutError: - session_logger.warning( - self.device_id, - self.session_id, - "接收Gummy ASR结果超时，使用最后的中间结果", - ) - - if self.callback.received_result or self.callback.transcript: - self.transcript = self.callback.transcript - - except Exception as e: - session_logger.error( - self.device_id, self.session_id, f"接收Gummy ASR结果出错: {str(e)}" - ) - + self._check_error() + if not self._complete or self._pending: + raise ASRError("识别未完整结束，拒绝使用不完整转录") + self.transcript = "".join(self._sentences[key] for key in sorted(self._sentences)) + session_logger.info( + self.device_id, self.session_id, + f"ASR最终结果 model={self.model} task_id={self.task_id}: {self.transcript}", + ) return self.transcript async def close(self): - if self.connected: - try: - if self.translator is not None: - try: - self.translator.stop() - except Exception as e: - if "has stopped" not in str(e): - session_logger.warning( - self.device_id, - self.session_id, - f"关闭Gummy translator出错: {str(e)}" - ) - self.translator = None - self.connected = False - session_logger.info( - self.device_id, self.session_id, "Gummy ASR连接已关闭" - ) - except Exception as e: - session_logger.error( - self.device_id, self.session_id, f"关闭Gummy ASR连接出错: {str(e)}" - ) + async with self._close_lock: + await self._close_connection() - def _safe_stop_translator(self): - """安全停止translator，忽略已停止的异常""" + async def _close_connection(self): + self._closing = True + self.connected = False + # Wake waiters if the session is interrupted during start or finish. + if self.task_id and not self._complete and not self.error: + self.error = ASRError("识别会话已关闭") + self._ready.set() + self._finished.set() + if self._receiver: + self._receiver.cancel() + await asyncio.gather(self._receiver, return_exceptions=True) + self._receiver = None try: - if self.translator: - self.translator.stop() - except Exception as e: - if "has stopped" not in str(e): - session_logger.warning( - self.device_id, - self.session_id, - f"停止translator时出现非标准错误: {str(e)}", - ) - return None + if self._websocket and not self._websocket.closed: + await asyncio.wait_for(self._websocket.close(), timeout=1.0) + except (asyncio.TimeoutError, aiohttp.ClientError): + pass + finally: + if self._client: + await self._client.close() + self._client = None + self._websocket = None + self.audio_buffer.clear() diff --git a/talkingq-url/scripts/smoke_ai_dialog.py b/talkingq-url/scripts/smoke_ai_dialog.py new file mode 100644 index 0000000..01a68da --- /dev/null +++ b/talkingq-url/scripts/smoke_ai_dialog.py @@ -0,0 +1,232 @@ +"""Local real-provider dialogue acceptance using isolated MySQL and actual /ws. + +Requires an isolated MySQL on 127.0.0.1:13316 (root without password), +macOS afconvert, and existing local ASR/LLM/TTS test credentials. +Starts only the production WebSocket router and asset serving on loopback. +Never starts production MQTT or changes .env. Writes evidence under output/. +""" +import asyncio +import argparse +import hashlib +import json +import shutil +import struct +import sys +import time +import uuid +import wave +from pathlib import Path + +import aiohttp +import pymysql +import uvicorn +import websockets +import yaml +from dotenv import dotenv_values +from fastapi import FastAPI + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +from config import settings + +OUT = ROOT / 'output/qwen-dialog-20261012' +OUT.mkdir(parents=True, exist_ok=True) +DEVICE_PREFIX = 'TalkingQ_E2E' + uuid.uuid4().hex[:6] +settings.db_host, settings.db_port = '127.0.0.1', 13316 +settings.db_user, settings.db_password, settings.db_name = 'root', '', 'qwen_dialog_test' +settings.server_host, settings.server_port = '127.0.0.1', 18086 +settings.assets_dir = str(OUT / 'assets') +settings.device_audio_cache_dir = str(OUT / 'device-audio') +settings.talkingq_mqtt_broker = '' +settings.selected_role_key = 'Banban' +tts_env = dotenv_values(ROOT / 'test/minimax.env') +settings.minimax_api_key = settings.minimax_api_key or tts_env.get('MINIMAX_API_KEY', '') +assert settings.minimax_api_key and settings.volcano_api_key and settings.aliyun_api_key +assert settings.aliyun_asr_model == 'qwen-audio-3.1-asr-flash-streaming' + +from database.models import Base, Role, RoleLanguage, DeviceAuth, DeviceConfig, ConversationHistory, ConversationMessage +from database.connection import get_db_manager +from services.role_manager import role_manager +from services.audio_session import audio_session_manager +from services.task_manager import task_manager +from api.websocket import router +from api.assets import configure_static_assets +from sqlalchemy import select + + +def packet(device, sid, seq, kind, audio=b''): + return device.encode() + b'\0' + sid.encode() + b'\0' + struct.pack(' 0 and any(pcm), 'empty/silent decoded reply' + async with http.get(url, headers={'Range': 'bytes=0-1023'}) as response: + assert response.status == 206 and await response.read() == body[:1024] + return {'url': url, 'file': str(local.relative_to(ROOT)), 'bytes': len(body), 'sha256': hashlib.sha256(body).hexdigest(), 'decoded_seconds': round(seconds, 3), 'http': 200, 'range': 206} + + +async def run_turn(ws, http, manager, device, role, case, interrupt=False): + sid = uuid.uuid4().hex + result = {'role': role, 'case': case['name'], 'session_id': sid, 'model': settings.aliyun_asr_model, + 'messages': [], 'audio': [], 'passed': False, 'interrupted': interrupt} + started = time.perf_counter() + try: + await ws.send(packet(device, sid, 0, 1)) + assert await asyncio.wait_for(ws.recv(), 15) == 'START_ACK' + result['messages'].append('START_ACK') + # Device START_ACK precedes ASR readiness. Real handler queues subsequent + # WebSocket frames while establishing the upstream connection. + audio = (ROOT / case['audio']).read_bytes() + for seq, offset in enumerate(range(0, len(audio), 640), 1): + await ws.send(packet(device, sid, seq, 0, audio[offset:offset + 640])) + await asyncio.sleep(len(audio[offset:offset + 640]) / 2000) + if interrupt and offset >= 1280: + break + ending = time.perf_counter() + await ws.send(packet(device, sid, seq + 1, 3 if interrupt else 2)) + while True: + message = await asyncio.wait_for(ws.recv(), 90) + result['messages'].append(message) + if message.startswith(('TTS_URL:', 'PROMPT_SOUND_URL:')): + if 'first_audio_seconds' not in result: + result['first_audio_seconds'] = round(time.perf_counter() - ending, 3) + result['audio'].append(await download(http, message.split(':', 1)[1], sid, len(result['audio']))) + if message.startswith('PROMPT_SOUND_URL:'): + break + if message == 'TTS_END' or (interrupt and message == 'INTERRUPT_ACK'): + break + if message == 'TTS_ERROR': + raise AssertionError('TTS_ERROR received') + # END may be delivered before the history transaction completes. + for _ in range(100): + session = await audio_session_manager.get_session((device, sid)) + if session and session.asr_service._client is None and (not session.llm_service.connected): + break + await asyncio.sleep(.05) + if session: + result['transcript'] = session.transcript + result['asr_error'] = session.asr_error + result['resource_state'] = { + 'asr_closed': session.asr_service._client is None and session.asr_service._receiver is None, + 'llm_closed': not session.llm_service.connected and (session.llm_service.session is None or session.llm_service.session.closed), + 'tts_closed': session.tts_service._client_session is None or session.tts_service._client_session.closed, + } + result['resources_released'] = all(result['resource_state'].values()) + async with await manager.get_session() as db: + query = select(ConversationMessage).join(ConversationHistory).where(ConversationHistory.device_id == device, + ConversationHistory.role_key == role).order_by(ConversationMessage.timestamp) + messages = (await db.execute(query)).scalars().all() + result['history_messages'] = [{'is_user': m.is_user, 'text': m.content} for m in messages] + if interrupt: + assert 'INTERRUPT_ACK' in result['messages'] and not result['audio'] + elif not case['expected_keywords']: + assert not result.get('transcript') and any(m.startswith('PROMPT_SOUND_URL:') for m in result['messages']) + else: + assert not result.get('asr_error') + assert all(word.lower() in result.get('transcript', '').lower() for word in case['expected_keywords']) + assert 'TTS_START' in result['messages'] and 'TTS_END' in result['messages'] and result['audio'] + assert messages and not messages[-1].is_user and messages[-1].content.strip(), 'assistant history missing' + assert len(messages) >= 2 and messages[-2].content == result['transcript'], 'latest history does not match current transcript' + result['reply'] = messages[-1].content + assert result.get('resources_released'), 'provider resources not released' + result['passed'] = True + except Exception as exc: + result['error'] = f'{type(exc).__name__}: {exc}' + result['total_seconds'] = round(time.perf_counter() - started, 3) + return result + + +async def main(interrupt_only=False): + manager = await setup(interrupt_only) + app = FastAPI() + configure_static_assets(app) + app.include_router(router) + server = uvicorn.Server(uvicorn.Config(app, host='127.0.0.1', port=18086, log_level='warning', lifespan='off')) + server_task = asyncio.create_task(server.serve()) + results = [] + try: + for _ in range(100): + if server.started: break + await asyncio.sleep(.05) + assert server.started + cases = json.loads((ROOT / 'tests/fixtures/asr/manifest.json').read_text()) + async with aiohttp.ClientSession() as http: + for index, role in enumerate(['Banban', 'Baikeboshi']): + device = f'{DEVICE_PREFIX}{index:02d}' + async with websockets.connect('ws://127.0.0.1:18086/ws', ping_interval=None) as ws: + await ws.send(json.dumps({'device_id': device, 'serial_number': f'LOCAL-E2E-{index}'})) + assert json.loads(await ws.recv())['status'] == 'authenticated' + plan = [(case, False) for case in cases] + if interrupt_only: + plan = [(cases[1], True), (cases[0], False)] + for case, interrupted in plan: + result = await asyncio.wait_for(run_turn(ws, http, manager, device, role, case, interrupt=interrupted), 150) + results.append(result) + (OUT / ('results-interrupt.json' if interrupt_only else 'results.json')).write_text(json.dumps(results, ensure_ascii=False, indent=2)) + print('E2E_RESULT', json.dumps(result, ensure_ascii=False), flush=True) + if not result['passed'] and not interrupt_only: + return False + return all(r['passed'] for r in results) + finally: + server.should_exit = True + await server_task + await task_manager.cleanup_all_tasks() + await manager.close() + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--interrupt-only', action='store_true') + args = parser.parse_args() + sys.exit(0 if asyncio.run(main(args.interrupt_only)) else 1) diff --git a/talkingq-url/scripts/smoke_fun_asr.py b/talkingq-url/scripts/smoke_fun_asr.py new file mode 100644 index 0000000..5f7e622 --- /dev/null +++ b/talkingq-url/scripts/smoke_fun_asr.py @@ -0,0 +1,78 @@ +"""Stream local MP3 fixtures through the backend path to the configured ASR model. + +Run from talkingq-url: .venv/bin/python scripts/smoke_fun_asr.py MANIFEST --output RESULT +Manifest: [{"name": "short", "audio": "path.mp3", "expected_keywords": ["你好"]}] +Uses the local ALIYUN_API_KEY. Does not start the app, connect to DB/MQTT, +invoke LLM/TTS, or send anything to a physical device. Audio must be 16 kbps MP3 +unless --bytes-per-second is supplied. Empty keywords means silence is expected. +""" +import argparse +import asyncio +import json +from pathlib import Path +import sys +import time + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from implementations.aliyun_asr import AliyunASR +from services.audio_session import AudioSession +from handlers.audio_session_handler import handle_audio_packet +from handlers.service_connection_handler import send_audio_task, end_asr_session + + +async def run(args): + cases = json.loads(args.manifest.read_text()) + results = [] + for role in ['Banban', 'Baikeboshi']: + for case in cases: + audio = Path(case['audio']).read_bytes() + asr = AliyunASR(selected_role={'role_key': role}) + asr.device_id = 'LOCAL_ASR_TEST' + asr.session_id = f"{role}-{case['name']}" + session = AudioSession(asr_service=asr) + session.device_id, session.session_id = asr.device_id, asr.session_id + session.send_queue = asyncio.Queue() + started = time.perf_counter() + result = {'role': role, 'case': case['name'], 'model': asr.model, 'bytes': len(audio)} + try: + await asr.connect() + result['ready_seconds'] = round(time.perf_counter() - started, 3) + session.send_task = asyncio.create_task(send_audio_task(session)) + for offset in range(0, len(audio), 640): + chunk = audio[offset:offset + 640] + await handle_audio_packet(session, chunk) + await asyncio.sleep(len(chunk) / args.bytes_per_second) + ending = time.perf_counter() + await end_asr_session(session) + result['finish_seconds'] = round(time.perf_counter() - ending, 3) + result['transcript'] = session.transcript + result['error'] = session.asr_error + expected = case['expected_keywords'] + result['passed'] = not session.asr_error and ( + all(word.casefold() in session.transcript.casefold() for word in expected) + if expected else session.transcript == '' + ) + except Exception as exc: + result.update(error=f'{type(exc).__name__}: {exc}', passed=False) + finally: + if session.send_task: + session.send_task.cancel() + await asyncio.gather(session.send_task, return_exceptions=True) + await asr.close() + result['task_id'] = asr.task_id + result['resources_released'] = asr._client is None and asr._receiver is None + results.append(result) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(results, ensure_ascii=False, indent=2)) + print(json.dumps(result, ensure_ascii=False), flush=True) + return all(item['passed'] and item['resources_released'] for item in results) + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('manifest', type=Path) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--bytes-per-second', type=float, default=2000) + args = parser.parse_args() + sys.exit(0 if asyncio.run(run(args)) else 1) diff --git a/talkingq-url/services/audio_session.py b/talkingq-url/services/audio_session.py index 1d7b584..2707e32 100644 --- a/talkingq-url/services/audio_session.py +++ b/talkingq-url/services/audio_session.py @@ -21,6 +21,7 @@ class AudioSession: self.send_queue = None self.send_task = None self.transcript = "" + self.asr_error = None self.last_activity_time = asyncio.get_running_loop().time() self.websocket = None self.llm_service = llm_service diff --git a/talkingq-url/services/factory.py b/talkingq-url/services/factory.py index c881607..5397f36 100644 --- a/talkingq-url/services/factory.py +++ b/talkingq-url/services/factory.py @@ -14,7 +14,7 @@ async def get_asr(selected_role: dict = None) -> ASR: session_logger.info( "system", "factory", - f"ASR提供商: Aliyun (固定)" + f"ASR提供商: Aliyun, 模型: {settings.aliyun_asr_model}" ) return AliyunASR(selected_role=selected_role) diff --git a/talkingq-url/tests/fixtures/asr/english.mp3 b/talkingq-url/tests/fixtures/asr/english.mp3 new file mode 100644 index 0000000..c130a07 Binary files /dev/null and b/talkingq-url/tests/fixtures/asr/english.mp3 differ diff --git a/talkingq-url/tests/fixtures/asr/manifest.json b/talkingq-url/tests/fixtures/asr/manifest.json new file mode 100644 index 0000000..9f13fce --- /dev/null +++ b/talkingq-url/tests/fixtures/asr/manifest.json @@ -0,0 +1,46 @@ +[ + { + "name": "short", + "audio": "tests/fixtures/asr/short.mp3", + "source_text": "你好，请讲一个故事。", + "expected_keywords": [ + "你好", + "故事" + ] + }, + { + "name": "multi", + "audio": "tests/fixtures/asr/multi.mp3", + "source_text": "今天天气很好。[[slnc 2500]]我想去公园散步。", + "expected_keywords": [ + "天气", + "公园", + "散步" + ] + }, + { + "name": "slow", + "audio": "tests/fixtures/asr/slow.mp3", + "source_text": "我今天有一点累，想听一首歌。", + "expected_keywords": [ + "今天", + "累", + "歌" + ] + }, + { + "name": "english", + "audio": "tests/fixtures/asr/english.mp3", + "source_text": "Hello, please tell me a story about a little rabbit.", + "expected_keywords": [ + "story", + "rabbit" + ] + }, + { + "name": "silence", + "audio": "tests/fixtures/asr/silence.mp3", + "source_text": "", + "expected_keywords": [] + } +] diff --git a/talkingq-url/tests/fixtures/asr/multi.mp3 b/talkingq-url/tests/fixtures/asr/multi.mp3 new file mode 100644 index 0000000..2171e45 Binary files /dev/null and b/talkingq-url/tests/fixtures/asr/multi.mp3 differ diff --git a/talkingq-url/tests/fixtures/asr/short.mp3 b/talkingq-url/tests/fixtures/asr/short.mp3 new file mode 100644 index 0000000..38ff05f Binary files /dev/null and b/talkingq-url/tests/fixtures/asr/short.mp3 differ diff --git a/talkingq-url/tests/fixtures/asr/silence.mp3 b/talkingq-url/tests/fixtures/asr/silence.mp3 new file mode 100644 index 0000000..1ed3ca8 Binary files /dev/null and b/talkingq-url/tests/fixtures/asr/silence.mp3 differ diff --git a/talkingq-url/tests/fixtures/asr/slow.mp3 b/talkingq-url/tests/fixtures/asr/slow.mp3 new file mode 100644 index 0000000..42729b2 Binary files /dev/null and b/talkingq-url/tests/fixtures/asr/slow.mp3 differ diff --git a/talkingq-url/tests/test_fun_asr.py b/talkingq-url/tests/test_fun_asr.py new file mode 100644 index 0000000..8633756 --- /dev/null +++ b/talkingq-url/tests/test_fun_asr.py @@ -0,0 +1,456 @@ +"""Test real adapter I/O against a local DashScope-protocol WebSocket server.""" +import asyncio +from types import SimpleNamespace + +import pytest +import pytest_asyncio +from aiohttp import web, WSMsgType + +from config import settings +from implementations.aliyun_asr import AliyunASR, ASRError +from services.audio_session import AudioSession +from services.interrupt_handler import interrupt_handler +from handlers.audio_session_handler import handle_audio_packet +from handlers.service_connection_handler import send_audio_task, end_asr_session + + +@pytest_asyncio.fixture(params=['qwen-audio-3.1-asr-flash-streaming', 'fun-asr-realtime']) +async def provider(monkeypatch, unused_tcp_port, request): + state = SimpleNamespace( + start_error=None, finish_error=None, start_delay=0, finish_delay=0, + no_finish=False, early_close=False, events=[], chunks=[], commands=[], + authorized=False, finish_seen=asyncio.Event(), started=asyncio.Event(), clients=[], + model=request.param, + ) + + async def handler(request): + state.authorized = request.headers.get('Authorization') == 'Bearer local-test-key' + ws = web.WebSocketResponse() + await ws.prepare(request) + task_id = None + + async def emit(kind, payload=None, **header): + await ws.send_json({'header': {'event': kind, 'task_id': task_id, **header}, + 'payload': payload or {}}) + + async for message in ws: + if message.type == WSMsgType.BINARY: + state.chunks.append(message.data) + if state.early_close: + await ws.close() + elif message.type == WSMsgType.TEXT: + command = message.json() + state.commands.append(command) + task_id = command['header']['task_id'] + if command['header']['action'] == 'run-task': + state.started.set() + await asyncio.sleep(state.start_delay) + if ws.closed: + break + if state.start_error: + await emit('task-failed', error_code=state.start_error, error_message='test rejection') + else: + await emit('task-started') + elif command['header']['action'] == 'finish-task': + state.finish_seen.set() + await asyncio.sleep(state.finish_delay) + if ws.closed: + break + if state.finish_error: + await emit('task-failed', error_code=state.finish_error, error_message='test failure') + else: + await emit('result-generated', {'output': {'sentence': sentence(99, 'wrong task')}}, + task_id='stale-task') + for item in state.events: + await emit('result-generated', {'output': {'sentence': item}}) + if not state.no_finish: + await emit('task-finished') + return ws + + app = web.Application() + app.router.add_get('/inference', handler) + runner = web.AppRunner(app) + await runner.setup() + await web.TCPSite(runner, '127.0.0.1', unused_tcp_port).start() + for key, value in { + 'aliyun_api_key': 'local-test-key', + 'aliyun_asr_url': f'http://127.0.0.1:{unused_tcp_port}/inference', + 'aliyun_asr_model': state.model, + 'aliyun_vocabulary_id': 'old-gummy-vocabulary', 'aliyun_asr_vocabulary_id': '', + }.items(): + monkeypatch.setattr(settings, key, value) + + def client(): + asr = AliyunASR() + state.clients.append(asr) + return asr + + state.client = client + yield state + for asr in state.clients: + await asr.close() + assert asr._client is None and asr._receiver is None + await runner.cleanup() + + +def sentence(number, text, final=True): + return {'sentence_id': number, 'text': text, 'sentence_end': final} + + +@pytest.mark.asyncio +async def test_full_stream_keeps_all_sentences_and_tail(provider): + provider.events = [sentence(1, '今天', False), sentence(1, '今天心情很好。'), + sentence(1, '今天心情很好。'), {'heartbeat': True, 'sentence_id': 0}, + sentence(2, '请讲个', False), sentence(2, '请讲个故事。')] + asr = provider.client() + await asr.connect() + audio = bytes(range(256)) * 29 + b'tail' + for offset in range(0, len(audio), 257): + await asr.send_audio(audio[offset:offset + 257]) + await asr.send_end() + await asr.send_end() + assert await asr.receive_results() == '今天心情很好。请讲个故事。' + assert b''.join(provider.chunks) == audio + assert provider.authorized + request = provider.commands[0]['payload'] + assert request['model'] == provider.model and request['function'] == 'recognition' + assert request['parameters']['format'] == 'mp3' + assert request['parameters']['sample_rate'] == 16000 + assert 'vocabulary_id' not in request['parameters'] + assert sum(c['header']['action'] == 'finish-task' for c in provider.commands) == 1 + + +@pytest.mark.asyncio +async def test_explicit_new_model_vocabulary(provider, monkeypatch): + monkeypatch.setattr(settings, 'aliyun_asr_vocabulary_id', 'fun-asr-hotwords') + asr = provider.client() + await asr.connect() + assert provider.commands[0]['payload']['parameters']['vocabulary_id'] == 'fun-asr-hotwords' + + +@pytest.mark.asyncio +async def test_waits_for_provider_task_started(provider): + provider.start_delay = 0.05 + asr = provider.client() + connecting = asyncio.create_task(asr.connect()) + await provider.started.wait() + assert not asr.connected + await connecting + assert asr.connected + + +@pytest.mark.asyncio +@pytest.mark.parametrize('code', ['InvalidParameter', 'InvalidApiKey', 'Throttling']) +async def test_start_rejection_does_not_report_connected(provider, code): + provider.start_error = code + asr = provider.client() + with pytest.raises(ASRError, match=code): + await asr.connect() + assert not asr.connected and asr._client is None and asr.transcript == '' + + +@pytest.mark.asyncio +async def test_connect_timeout_releases_resources(provider, monkeypatch): + provider.start_delay = 0.1 + monkeypatch.setattr(settings, 'asr_connect_timeout_seconds', 0.02) + asr = provider.client() + with pytest.raises(ASRError, match='TimeoutError'): + await asr.connect() + assert asr._client is None + + +@pytest.mark.asyncio +async def test_final_after_old_half_second_timeout_is_collected(provider): + provider.finish_delay = 0.6 + provider.events = [sentence(1, '请完整保留最后一句。')] + asr = provider.client() + await asr.connect() + await asr.send_audio(b'short') + await asr.send_end() + assert await asr.receive_results() == '请完整保留最后一句。' + assert provider.chunks == [b'short'] + + +@pytest.mark.asyncio +async def test_final_timeout_rejects_partial_transcript(provider, monkeypatch): + provider.no_finish = True + provider.events = [sentence(1, '半句话', False)] + monkeypatch.setattr(settings, 'asr_final_timeout_seconds', 0.03) + asr = provider.client() + await asr.connect() + with pytest.raises(ASRError, match='TimeoutError'): + await asr.send_end() + with pytest.raises(ASRError): + await asr.receive_results() + assert asr.transcript == '' + + +@pytest.mark.asyncio +async def test_finish_without_final_sentence_is_rejected(provider): + provider.events = [sentence(1, '未定稿', False)] + asr = provider.client() + await asr.connect() + await asr.send_end() + with pytest.raises(ASRError, match='不完整'): + await asr.receive_results() + + +@pytest.mark.asyncio +async def test_provider_failure_during_finish(provider): + provider.finish_error = 'InternalError' + asr = provider.client() + await asr.connect() + with pytest.raises(ASRError, match='InternalError'): + await asr.send_end() + assert not asr.connected + + +@pytest.mark.asyncio +async def test_early_disconnect_is_not_success(provider): + provider.early_close = True + asr = provider.client() + await asr.connect() + await asr.send_audio(b'x' * 3200) + await asyncio.wait_for(asr._finished.wait(), 1) + with pytest.raises(ASRError, match='提前关闭'): + await asr.receive_results() + + +@pytest.mark.asyncio +async def test_silence_has_empty_transcript(provider): + asr = provider.client() + await asr.connect() + await asr.send_end() + assert await asr.receive_results() == '' + + +@pytest.mark.asyncio +async def test_cancel_finish_releases_socket(provider): + provider.no_finish = True + asr = provider.client() + await asr.connect() + ending = asyncio.create_task(asr.send_end()) + await provider.finish_seen.wait() + ending.cancel() + with pytest.raises(asyncio.CancelledError): + await ending + assert asr._client is None and not asr.connected + + +@pytest.mark.asyncio +async def test_reconnect_does_not_reuse_previous_transcript(provider): + asr = provider.client() + provider.events = [sentence(1, '第一轮。')] + await asr.connect() + await asr.send_end() + assert await asr.receive_results() == '第一轮。' + old_task = asr.task_id + await asr.close() + provider.events = [sentence(1, '第二轮。')] + await asr.connect() + await asr.send_end() + assert await asr.receive_results() == '第二轮。' + assert old_task != asr.task_id + + +def audio_session(asr): + session = AudioSession(asr_service=asr) + session.device_id, session.session_id = 'local-test', 'round' + session.send_queue = asyncio.Queue() + session.send_task = asyncio.create_task(send_audio_task(session)) + return session + + +@pytest.mark.asyncio +async def test_end_drains_queued_audio_before_finish(provider): + provider.events = [sentence(1, '完整识别。')] + asr = provider.client() + await asr.connect() + session = audio_session(asr) + # No yield to the consumer: this reproduced the previous cancellation loss. + for part in [b'a' * 3200, b'b' * 3200, b'last']: + await handle_audio_packet(session, part) + await end_asr_session(session) + assert b''.join(provider.chunks) == b'a' * 3200 + b'b' * 3200 + b'last' + assert session.transcript == '完整识别。' and session.asr_error is None + assert session.send_task is None and session.send_queue.empty() + await asyncio.wait_for(session.send_queue.join(), 0.1) + assert asr._client is None + + +@pytest.mark.asyncio +async def test_queue_timeout_closes_asr_without_partial_result(provider, monkeypatch): + monkeypatch.setattr(settings, 'asr_queue_drain_timeout_seconds', 0.02) + asr = provider.client() + await asr.connect() + session = audio_session(asr) + + async def stalled(data): + await asyncio.Event().wait() + + monkeypatch.setattr(asr, 'send_audio', stalled) + await handle_audio_packet(session, b'pending') + await end_asr_session(session) + assert 'TimeoutError' in session.asr_error + assert session.transcript == '' and not provider.finish_seen.is_set() + assert asr._client is None + + +@pytest.mark.asyncio +async def test_queue_overflow_does_not_return_incomplete_recording(provider): + asr = provider.client() + await asr.connect() + session = audio_session(asr) + for _ in range(101): + await handle_audio_packet(session, b'x') + await end_asr_session(session) + assert '溢出' in session.asr_error and session.transcript == '' + assert asr._client is None + + +@pytest.mark.asyncio +async def test_sender_failure_still_closes_connection(provider, monkeypatch): + asr = provider.client() + await asr.connect() + session = audio_session(asr) + + async def fail(data): + raise RuntimeError('broken send') + + monkeypatch.setattr(asr, 'send_audio', fail) + await handle_audio_packet(session, b'audio') + await end_asr_session(session) + assert 'broken send' in session.asr_error and session.transcript == '' + assert asr._client is None + + +@pytest.mark.asyncio +async def test_interruption_closes_asr_without_transcript(provider): + asr = provider.client() + await asr.connect() + session = audio_session(asr) + key = (session.device_id, session.session_id) + await interrupt_handler.set_interrupt_state(key, True) + try: + await end_asr_session(session) + assert '中断' in session.asr_error and session.transcript == '' + assert asr._client is None + finally: + await interrupt_handler.remove_session(key) + + +@pytest.mark.asyncio +async def test_cancel_connect_releases_socket(provider): + provider.start_delay = 0.1 + asr = provider.client() + connecting = asyncio.create_task(asr.connect()) + await provider.started.wait() + connecting.cancel() + with pytest.raises(asyncio.CancelledError): + await connecting + assert asr._client is None and asr._receiver is None + + +@pytest.mark.asyncio +async def test_concurrent_cleanup_is_idempotent(provider): + asr = provider.client() + await asr.connect() + await asyncio.gather(asr.close(), asr.close()) + assert asr._client is None and asr._receiver is None + + +@pytest.mark.asyncio +async def test_send_timeout_is_reported(provider, monkeypatch): + monkeypatch.setattr(settings, 'asr_send_timeout_seconds', 0.02) + asr = provider.client() + await asr.connect() + + async def blocked(data): + await asyncio.Event().wait() + + monkeypatch.setattr(asr._websocket, 'send_bytes', blocked) + with pytest.raises(ASRError, match='TimeoutError'): + await asr.send_audio(b'a' * 3200) + assert not asr.connected + + +@pytest.mark.asyncio +async def test_active_interrupt_releases_asr_and_sender(provider): + from handlers.service_connection_handler import prepare_asr_service + asr = provider.client() + session = AudioSession(asr_service=asr) + session.device_id, session.session_id = 'local-test', 'active-interrupt' + session.send_queue = asyncio.Queue() + key = (session.device_id, session.session_id) + try: + await prepare_asr_service(session, *key) + assert asr.connected and session.send_task + await interrupt_handler.handle_interrupt(key) + assert asr._client is None and asr._receiver is None + assert session.send_task is None + finally: + await interrupt_handler.remove_session(key) + + +@pytest.mark.asyncio +async def test_start_failure_does_not_leave_waiting_sender(provider): + from handlers.service_connection_handler import prepare_asr_service + provider.start_error = 'InvalidParameter' + asr = provider.client() + session = AudioSession(asr_service=asr) + session.device_id, session.session_id = 'local-test', 'failed-start' + session.send_queue = asyncio.Queue() + await prepare_asr_service(session, session.device_id, session.session_id) + assert 'InvalidParameter' in session.asr_error + assert session.send_task is None and asr._client is None + + +@pytest.mark.asyncio +async def test_device_end_handler_passes_full_text_to_response_stage(provider, monkeypatch): + from unittest.mock import AsyncMock + from handlers.audio_session_handler import handle_end_packet + import handlers.transcription_handler as transcription_handler + provider.events = [sentence(1, '第一句。'), sentence(2, '第二句。')] + asr = provider.client() + await asr.connect() + session = audio_session(asr) + handle_transcription = AsyncMock() + monkeypatch.setattr(transcription_handler, 'handle_transcription', handle_transcription) + await handle_audio_packet(session, b'last bytes') + await handle_end_packet(session, (session.device_id, session.session_id)) + handle_transcription.assert_awaited_once_with(session, session.device_id, session.session_id) + assert session.transcript == '第一句。第二句。' + assert b''.join(provider.chunks) == b'last bytes' + + +@pytest.mark.asyncio +async def test_empty_recognition_uses_existing_prompt_fallback(provider, monkeypatch): + from unittest.mock import AsyncMock + from handlers.audio_session_handler import handle_end_packet + import handlers.prompt_sound_handler as prompts + from services.tts_error_manager import tts_error_manager + asr = provider.client() + await asr.connect() + session = audio_session(asr) + prompt = AsyncMock() + monkeypatch.setattr(prompts, 'handle_prompt_sound_request', prompt) + monkeypatch.setattr(tts_error_manager, 'end_tts_session', AsyncMock()) + await handle_end_packet(session, (session.device_id, session.session_id)) + prompt.assert_awaited_once_with(session.device_id, 'tts_error') + assert session.transcript == '' and asr._client is None + + +@pytest.mark.asyncio +async def test_cancel_session_start_cleans_up_sender(provider): + from handlers.service_connection_handler import prepare_asr_service + provider.start_delay = 0.1 + asr = provider.client() + session = AudioSession(asr_service=asr) + session.device_id, session.session_id = 'local-test', 'cancel-start' + session.send_queue = asyncio.Queue() + task = asyncio.create_task(prepare_asr_service(session, session.device_id, session.session_id)) + await provider.started.wait() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert session.send_task is None and asr._client is None diff --git a/talkingq-url/tests/test_warmup_cleanup.py b/talkingq-url/tests/test_warmup_cleanup.py new file mode 100644 index 0000000..b31a8e5 --- /dev/null +++ b/talkingq-url/tests/test_warmup_cleanup.py @@ -0,0 +1,68 @@ +"""Verify real aiohttp sessions close on interruption and warmup races.""" +import asyncio +import uuid +from unittest.mock import AsyncMock + +import pytest + +from implementations.minimax_tts import MiniMaxTTS +from implementations.volcano_llm import VolcanoLLM +from services.audio_session import AudioSession +from services.interrupt_handler import interrupt_handler +from services.conversation_history import conversation_history_manager +from handlers.service_connection_handler import prepare_llm_service, prepare_tts_service + + +@pytest.mark.asyncio +@pytest.mark.parametrize('kind', ['llm', 'tts']) +@pytest.mark.parametrize('phase', ['ready', 'connecting', 'cancelled', 'already_interrupted']) +async def test_warmup_releases_provider_client(monkeypatch, kind, phase): + key = ('warmup-test', uuid.uuid4().hex) + service = VolcanoLLM() if kind == 'llm' else MiniMaxTTS() + session = AudioSession(llm_service=service if kind == 'llm' else None, + tts_service=service if kind == 'tts' else None) + session.device_id, session.session_id = key + monkeypatch.setattr(conversation_history_manager, 'get_history', AsyncMock(return_value=None)) + prepare = prepare_llm_service if kind == 'llm' else prepare_tts_service + original_connect = service.connect + entered, resume = asyncio.Event(), asyncio.Event() + + async def delayed_connect(): + if phase == 'cancelled': + await original_connect() + entered.set() + await resume.wait() + if phase != 'cancelled': + await original_connect() + + task = None + try: + if phase == 'already_interrupted': + await interrupt_handler.set_interrupt_state(key, True) + if phase in ('connecting', 'cancelled'): + monkeypatch.setattr(service, 'connect', delayed_connect) + task = asyncio.create_task(prepare(session, *key, {})) + if phase in ('connecting', 'cancelled'): + await entered.wait() + if phase == 'cancelled': + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + else: + await interrupt_handler.handle_interrupt(key) + resume.set() + await task + else: + await task + if phase == 'ready': + assert service.connected + await interrupt_handler.handle_interrupt(key) + client = service.session if kind == 'llm' else service._client_session + assert client is None or client.closed + assert not service.connected + finally: + if task and not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await service.close() + await interrupt_handler.remove_session(key)