diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index 249b3b0cb..ffc5f295a 100644 --- a/app/agent/orchestrator.py +++ b/app/agent/orchestrator.py @@ -9,7 +9,7 @@ import uuid import warnings from dataclasses import dataclass from datetime import datetime, timedelta -from typing import Any, Callable, Dict, List, Optional +from typing import Any, Awaitable, Callable, Dict, List, Optional from fastapi.concurrency import run_in_threadpool from langchain.agents import create_agent @@ -2572,7 +2572,7 @@ class _MessageTask: allow_message_tools: bool = True output_callback: Optional[Callable[[str], None]] = None protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None - message_callback: Optional[Callable[[Any], None]] = None + message_callback: Optional[Callable[[Any], Awaitable[None] | None]] = None agent_factory: Optional[Callable[..., MoviePilotAgent]] = None agent_setup: Optional[Callable[[MoviePilotAgent], None]] = None completion_future: Optional[asyncio.Future] = None @@ -2834,7 +2834,7 @@ class AgentManager: allow_message_tools: bool = True, output_callback: Optional[Callable[[str], None]] = None, protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None, - message_callback: Optional[Callable[[Any], None]] = None, + message_callback: Optional[Callable[[Any], Awaitable[None] | None]] = None, agent_factory: Optional[Callable[..., MoviePilotAgent]] = None, agent_setup: Optional[Callable[[MoviePilotAgent], None]] = None, wait_for_completion: bool = False, diff --git a/app/agent/tools/base.py b/app/agent/tools/base.py index cbefa569a..0c0a9c988 100644 --- a/app/agent/tools/base.py +++ b/app/agent/tools/base.py @@ -1,4 +1,5 @@ import asyncio +import inspect import json import threading from abc import ABCMeta, abstractmethod @@ -646,7 +647,9 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta): self._channel == NotificationChannel.WebAgent.value and callable(callback) ): - callback(message) + callback_result = callback(message) + if inspect.isawaitable(callback_result): + await callback_result return if not self._channel or not self._source: diff --git a/app/agent/tools/impl/delete_transfer_history.py b/app/agent/tools/impl/delete_transfer_history.py index e001dde3c..b8506ea6d 100644 --- a/app/agent/tools/impl/delete_transfer_history.py +++ b/app/agent/tools/impl/delete_transfer_history.py @@ -1,5 +1,6 @@ """删除整理历史记录工具""" +import asyncio from typing import Optional, Type from pydantic import BaseModel, Field @@ -20,6 +21,14 @@ class DeleteTransferHistoryInput(BaseModel): ) +def _delete_history_destination_file(fileitem: FileItem) -> tuple[bool, bool]: + """在存储 worker 内完成旧目标检查和删除，保持历史删除前的顺序。""" + storage_chain = StorageChain() + if not storage_chain.exists(fileitem): + return False, False + return True, bool(storage_chain.delete_media_file(fileitem)) + + class DeleteTransferHistoryTool(MoviePilotTool): name: str = "delete_transfer_history" tags: list[str] = [ @@ -55,9 +64,21 @@ class DeleteTransferHistoryTool(MoviePilotTool): deleted_dest = False if history.dest_fileitem and not (history.status and history.mode == "move"): dest_fileitem = FileItem(**history.dest_fileitem) - storage_chain = StorageChain() - if storage_chain.exists(dest_fileitem): - if not storage_chain.delete_media_file(dest_fileitem): + try: + destination_exists, destination_deleted = await self.run_blocking( + "storage", + _delete_history_destination_file, + dest_fileitem, + ) + except asyncio.CancelledError: + logger.warning( + "删除整理历史的旧媒体文件等待已取消，底层文件操作可能仍在继续，" + "请确认实际状态后再重试，历史记录尚未删除，path=%s", + dest_fileitem.path, + ) + raise + if destination_exists: + if not destination_deleted: return f"错误：旧媒体库文件删除失败，路径={dest_fileitem.path}" deleted_dest = True await transferhis.async_delete(history_id) diff --git a/app/api/endpoints/agent.py b/app/api/endpoints/agent.py index c9f5aaa31..979cd20a7 100644 --- a/app/api/endpoints/agent.py +++ b/app/api/endpoints/agent.py @@ -4,14 +4,13 @@ import hashlib import json import mimetypes import shutil -import subprocess import time import uuid from collections import deque from queue import Empty, Queue from pathlib import Path from threading import Lock -from typing import Any, AsyncIterator, Callable, Optional, Union +from typing import Any, AsyncIterator, Awaitable, Callable, Optional, Union import aiofiles from fastapi import Depends, File, Form, HTTPException, Request, UploadFile, status @@ -85,6 +84,7 @@ WEB_AGENT_FILE_TTL_SECONDS = 6 * 60 * 60 WEB_AGENT_FILE_MAX_ITEMS = 256 WEB_AGENT_UPLOAD_MAX_BYTES = 32 * 1024 * 1024 WEB_AGENT_UPLOAD_CHUNK_SIZE = 1024 * 1024 +WEB_AGENT_AUDIO_CONVERSION_TIMEOUT_SECONDS = 60.0 WEB_AGENT_BROWSER_AUDIO_SUFFIXES = {".aac", ".m4a", ".mp3", ".mp4", ".wav", ".wave"} WEB_AGENT_TRADITIONAL_IDLE_TIMEOUT_SECONDS = 2.0 WEB_AGENT_TRADITIONAL_MAX_WAIT_SECONDS = 60.0 @@ -401,7 +401,9 @@ class _WebAgentMoviePilotAgentMixin: def __init__( self, *args: Any, - message_callback: Optional[Callable[[_SchemaMessage], None]] = None, + message_callback: Optional[ + Callable[[_SchemaMessage], Awaitable[None] | None] + ] = None, **kwargs: Any, ) -> None: super().__init__(*args, **kwargs) @@ -418,7 +420,9 @@ class _WebAgentMoviePilotAgentMixin: def set_message_callback( self, - message_callback: Optional[Callable[[_SchemaMessage], None]], + message_callback: Optional[ + Callable[[_SchemaMessage], Awaitable[None] | None] + ], ) -> None: """ 更新 Web SSE 通知回调，复用 Agent 实例时指向当前请求队列。 @@ -998,48 +1002,115 @@ def _get_web_agent_audio_mime_type(audio_path: Path) -> Optional[str]: return mimetypes.guess_type(audio_path.name)[0] -def _prepare_web_agent_audio_attachment_path(voice_path: str) -> Path: - """ - 将 Agent 语音回复准备成 Web 面板可稳定播放的音频文件。 - - 部分 TTS provider 会生成 Opus/Ogg，桌面 Chromium 通常可播放，但 iOS/Safari - 兼容性不稳定；WebAgent 只在浏览器内播放，因此这里单独转成 WAV。 - """ +def _resolve_web_agent_audio_source_path(voice_path: str) -> Path: + """解析语音源文件；文件暂时不可用时保留原始路径以维持回退语义。""" try: - source_path = Path(voice_path).expanduser().resolve(strict=True) + return Path(voice_path).expanduser().resolve(strict=True) except OSError: return Path(voice_path) + + +def _remove_web_agent_audio_output(path: Path) -> None: + """清理未完成的转码产物。""" + try: + path.unlink() + except FileNotFoundError: + pass + except OSError as err: + logger.debug("WebAgent 清理未完成语音转码产物失败: path=%s, error=%s", path, err) + + +async def _terminate_web_agent_audio_process( + process: Optional[asyncio.subprocess.Process], +) -> None: + """终止并回收转码进程，避免取消或超时留下孤儿 ffmpeg。""" + if process is None or process.returncode is not None: + return + try: + process.kill() + except ProcessLookupError: + return + try: + await process.communicate() + except (OSError, ProcessLookupError): + pass + + +async def _prepare_web_agent_audio_attachment_path_async(voice_path: str) -> Path: + """异步准备 WebAgent 语音附件，转码等待可取消且有超时。""" + source_path = await run_in_threadpool( + _resolve_web_agent_audio_source_path, + voice_path, + ) if source_path.suffix.lower() in WEB_AGENT_BROWSER_AUDIO_SUFFIXES: return source_path - if not shutil.which("ffmpeg"): + + ffmpeg_path = await run_in_threadpool(shutil.which, "ffmpeg") + if not ffmpeg_path: logger.warning("WebAgent 语音转 WAV 跳过：ffmpeg 不可用，path=%s", source_path) return source_path voice_dir = get_api_runtime_config_snapshot().temp_path / "voice" - voice_dir.mkdir(parents=True, exist_ok=True) + try: + await run_in_threadpool(voice_dir.mkdir, parents=True, exist_ok=True) + except OSError as err: + logger.warning("WebAgent 语音转 WAV 目录不可用，将回退原文件: path=%s, error=%s", voice_dir, err) + return source_path + output_path = voice_dir / f"{source_path.stem}_web_{uuid.uuid4().hex[:8]}.wav" - cmd = [ - "ffmpeg", - "-y", - "-i", - str(source_path), - "-ar", - "24000", - "-ac", - "1", - "-f", - "wav", - str(output_path), - ] - result = subprocess.run(cmd, capture_output=True, text=True, check=False) - if result.returncode != 0 or not output_path.exists(): + process: Optional[asyncio.subprocess.Process] = None + + async def cleanup_conversion_output() -> None: + """清理失败或取消的转码进程及临时产物。""" + await _terminate_web_agent_audio_process(process) + await run_in_threadpool(_remove_web_agent_audio_output, output_path) + + try: + process = await asyncio.create_subprocess_exec( + ffmpeg_path, + "-y", + "-i", + str(source_path), + "-ar", + "24000", + "-ac", + "1", + "-f", + "wav", + str(output_path), + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.PIPE, + ) + _, stderr = await asyncio.wait_for( + process.communicate(), + timeout=WEB_AGENT_AUDIO_CONVERSION_TIMEOUT_SECONDS, + ) + output_exists = await run_in_threadpool(output_path.exists) + if process.returncode != 0 or not output_exists: + await cleanup_conversion_output() + logger.warning( + "WebAgent 语音转 WAV 失败，将回退原文件: returncode=%s, stderr=%s", + process.returncode, + (stderr or b"").decode("utf-8", errors="replace").strip()[:500], + ) + return source_path + return output_path + except asyncio.TimeoutError: + await cleanup_conversion_output() logger.warning( - "WebAgent 语音转 WAV 失败，将回退原文件: returncode=%s, stderr=%s", - result.returncode, - (result.stderr or "").strip()[:500], + "WebAgent 语音转 WAV 超时，将回退原文件: timeout=%ss, path=%s", + WEB_AGENT_AUDIO_CONVERSION_TIMEOUT_SECONDS, + source_path, ) return source_path - return output_path + except asyncio.CancelledError: + await cleanup_conversion_output() + logger.warning("WebAgent 语音转 WAV 已取消，path=%s", source_path) + raise + except OSError as err: + await cleanup_conversion_output() + logger.warning("WebAgent 语音转 WAV 启动失败，将回退原文件: path=%s, error=%s", source_path, err) + return source_path def _get_web_agent_registered_file(ref: str) -> Optional[dict[str, Any]]: @@ -1213,6 +1284,8 @@ def _resolve_web_agent_choice_payload(callback_data: str, user_id: str) -> Optio def _build_web_agent_message_events( message: _SchemaMessage, + *, + prepared_audio_path: Optional[Path] = None, ) -> list[dict]: """ 将 Agent 工具通知转换为 Web SSE 事件。 @@ -1250,7 +1323,7 @@ def _build_web_agent_message_events( events.append({"type": "attachment", "attachment": attachment}) if message.voice_path: - audio_path = _prepare_web_agent_audio_attachment_path(message.voice_path) + audio_path = prepared_audio_path or Path(message.voice_path) attachment = _register_web_agent_file( str(audio_path), file_name=audio_path.name, @@ -1271,6 +1344,21 @@ def _build_web_agent_message_events( return events +async def _build_web_agent_message_events_async( + message: _SchemaMessage, +) -> list[dict]: + """异步构造 WebAgent 通知，确保语音转码不占用事件循环。""" + prepared_audio_path = None + if message.voice_path: + prepared_audio_path = await _prepare_web_agent_audio_attachment_path_async( + message.voice_path + ) + return _build_web_agent_message_events( + message, + prepared_audio_path=prepared_audio_path, + ) + + def _build_web_agent_display_message_from_events( events: list[dict], ) -> dict: @@ -1581,7 +1669,7 @@ async def _collect_web_agent_traditional_events( if not _is_web_agent_message_for_user(message, user_id): continue - events.extend(_build_web_agent_message_events(message)) + events.extend(await _build_web_agent_message_events_async(message)) idle_deadline = time.monotonic() + WEB_AGENT_TRADITIONAL_IDLE_TIMEOUT_SECONDS return events finally: @@ -2231,11 +2319,11 @@ async def _web_agent_stream_impl( _apply_web_agent_display_event(item, assistant_display_message) event_publisher.publish(item) - def message_callback(message: _SchemaMessage) -> None: + async def message_callback(message: _SchemaMessage) -> None: """ 接收 Agent 工具主动发送的 Web 通知。 """ - for item in _build_web_agent_message_events(message): + for item in await _build_web_agent_message_events_async(message): _apply_web_agent_display_event(item, assistant_display_message) event_publisher.publish(item) diff --git a/tests/test_agent_tool_streaming.py b/tests/test_agent_tool_streaming.py index 74ab7d136..89dcfc2b9 100644 --- a/tests/test_agent_tool_streaming.py +++ b/tests/test_agent_tool_streaming.py @@ -14,7 +14,7 @@ from app.agent.tools.base import MoviePilotTool from app.agent.tools.impl.send_voice_message import SendVoiceMessageTool from app.api.endpoints.openai import _get_openai_streaming_handler_type from app.runtime.config import settings -from app.schemas.message import MessageResponse +from app.schemas.message import Message, MessageResponse from app.schemas.types import NotificationChannel, MessageType @@ -77,6 +77,31 @@ class AdminOnlyDummyTool(MoviePilotTool): class TestAgentToolStreaming: """Agent 工具流式输出测试。""" + def test_web_message_callback_can_await_async_delivery(self): + """WebAgent 通知回调支持异步附件准备并保持发送顺序。""" + received = [] + + async def scenario(): + tool = DummyTool(session_id="session-1", user_id="10001") + tool.set_message_attr("WebAgent", "web-agent", "admin") + + async def callback(message): + await asyncio.sleep(0) + received.append(message.text) + + tool.set_agent_context({"message_callback": callback}) + await tool.send_message( + Message( + text="异步通知", + channel=NotificationChannel.WebAgent, + mtype=MessageType.Agent, + ) + ) + + asyncio.run(scenario()) + + assert received == ["异步通知"] + async def _run_tool(self, initial_buffer: str) -> tuple[str, str]: """运行测试工具并返回工具结果与缓冲内容。""" tool = DummyTool(session_id="session-1", user_id="10001") diff --git a/tests/test_delete_transfer_history_tool.py b/tests/test_delete_transfer_history_tool.py index 0f550bdeb..d656480af 100644 --- a/tests/test_delete_transfer_history_tool.py +++ b/tests/test_delete_transfer_history_tool.py @@ -1,4 +1,5 @@ import asyncio +import threading from types import SimpleNamespace from app.agent.tools.impl.delete_transfer_history import DeleteTransferHistoryTool @@ -277,6 +278,130 @@ def test_delete_transfer_history_tool_only_treats_exact_move_as_reorganize_sourc ] +def test_delete_transfer_history_storage_work_runs_outside_event_loop(monkeypatch): + """整理历史的本地存储操作应在 storage worker 中执行。""" + caller_thread = threading.get_ident() + storage_threads = [] + history = SimpleNamespace( + id=15, + title="奔跑吧", + src="/downloads/Keep.Running.mkv", + status=True, + mode="copy", + dest_fileitem={ + "storage": "local", + "path": "/library/奔跑吧 (2014)/Keep.Running.mkv", + "name": "Keep.Running.mkv", + "type": "file", + }, + ) + + class FakeTransferHistoryOper: + async def async_get(self, history_id): + return history + + async def async_delete(self, history_id): + return None + + class FakeStorageChain: + def exists(self, fileitem): + storage_threads.append(threading.get_ident()) + return True + + def delete_media_file(self, fileitem): + storage_threads.append(threading.get_ident()) + return True + + monkeypatch.setattr( + "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + FakeTransferHistoryOper, + ) + monkeypatch.setattr( + "app.agent.tools.impl.delete_transfer_history.StorageChain", + FakeStorageChain, + ) + + result = asyncio.run( + DeleteTransferHistoryTool( + session_id="redo-session", + user_id="10001", + ).run(history_id=15) + ) + + assert "已删除整理历史记录" in result + assert storage_threads + assert all(thread_id != caller_thread for thread_id in storage_threads) + + +def test_delete_transfer_history_cancellation_keeps_history_record(monkeypatch): + """取消等待存储清理时不得继续提交整理历史删除。""" + started = threading.Event() + release = threading.Event() + finished = threading.Event() + history = SimpleNamespace( + id=16, + title="奔跑吧", + src="/downloads/Keep.Running.mkv", + status=True, + mode="copy", + dest_fileitem={ + "storage": "local", + "path": "/library/奔跑吧 (2014)/Keep.Running.mkv", + "name": "Keep.Running.mkv", + "type": "file", + }, + ) + delete_history_calls = [] + + class FakeTransferHistoryOper: + async def async_get(self, history_id): + return history + + async def async_delete(self, history_id): + delete_history_calls.append(history_id) + + class FakeStorageChain: + def exists(self, fileitem): + started.set() + release.wait(timeout=1) + return True + + def delete_media_file(self, fileitem): + finished.set() + return True + + monkeypatch.setattr( + "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper", + FakeTransferHistoryOper, + ) + monkeypatch.setattr( + "app.agent.tools.impl.delete_transfer_history.StorageChain", + FakeStorageChain, + ) + + async def scenario(): + task = asyncio.create_task( + DeleteTransferHistoryTool( + session_id="redo-session", + user_id="10001", + ).run(history_id=16) + ) + assert await asyncio.to_thread(started.wait, 1) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + try: + asyncio.run(scenario()) + finally: + release.set() + + assert delete_history_calls == [] + assert finished.wait(timeout=1) + + def test_manual_redo_context_uses_dest_path_for_successful_move_record(): """成功 move 记录重新整理时，旧目标文件才是可继续整理的输入路径。""" history = SimpleNamespace( diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 97ceca095..687349532 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -27,7 +27,7 @@ from app.api.endpoints.agent import ( _extract_web_agent_message_from_event_data, _get_web_agent_type, _has_web_agent_traditional_interaction, - _prepare_web_agent_audio_attachment_path, + _prepare_web_agent_audio_attachment_path_async, _resolve_web_agent_audio_refs, _transcribe_web_agent_audio_files, web_agent_stream, @@ -661,31 +661,271 @@ def test_build_web_agent_message_events_registers_voice_attachment(tmp_path): assert attachment["url"].startswith("message/agent/file/") -def test_prepare_web_agent_audio_attachment_converts_unsupported_audio(tmp_path): - """WebAgent 会把浏览器不稳定支持的语音格式转为 WAV 供面板播放。""" +def test_prepare_web_agent_audio_attachment_async_keeps_loop_responsive(tmp_path): + """异步转码等待期间事件循环仍应可调度其它任务。""" source_path = tmp_path / "reply.opus" source_path.write_bytes(b"opus-bytes") converted_path = tmp_path / "voice" / "reply_web_abcdef12.wav" + started = asyncio.Event() + release = asyncio.Event() - with patch("app.api.endpoints.agent.shutil.which", return_value="/usr/bin/ffmpeg"), patch( - "app.api.endpoints.agent.uuid.uuid4", - return_value=SimpleNamespace(hex="abcdef1234567890"), - ), patch("app.api.endpoints.agent.subprocess.run") as run: - def write_converted_file(*args, **kwargs): + class FakeProcess: + returncode = 0 + + async def communicate(self): + started.set() + await release.wait() converted_path.write_bytes(b"wav-bytes") - return SimpleNamespace(returncode=0, stderr="") + return b"", b"" - run.side_effect = write_converted_file + async def fake_create_subprocess_exec(*args, **kwargs): + assert args[0] == "/usr/bin/ffmpeg" + return FakeProcess() + + async def scenario(): with patch( + "app.api.endpoints.agent.shutil.which", + return_value="/usr/bin/ffmpeg", + ), patch( + "app.api.endpoints.agent.uuid.uuid4", + return_value=SimpleNamespace(hex="abcdef1234567890"), + ), patch( + "app.api.endpoints.agent.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ), patch( "app.api.endpoints.agent.get_api_runtime_config_snapshot", return_value=SimpleNamespace(temp_path=tmp_path), ): - output_path = _prepare_web_agent_audio_attachment_path(str(source_path)) + conversion_task = asyncio.create_task( + _prepare_web_agent_audio_attachment_path_async(str(source_path)) + ) + await asyncio.wait_for(started.wait(), timeout=1) + heartbeat = asyncio.create_task(asyncio.sleep(0)) + await heartbeat + assert not conversion_task.done() + release.set() + return await conversion_task + + output_path = asyncio.run(scenario()) assert output_path == converted_path assert output_path.read_bytes() == b"wav-bytes" +def test_prepare_web_agent_audio_attachment_async_cancellation_reaps_process(tmp_path): + """取消 WebAgent 转码时应终止并回收 ffmpeg，不能留下半成品。""" + source_path = tmp_path / "reply.opus" + source_path.write_bytes(b"opus-bytes") + output_path = tmp_path / "voice" / "reply_web_abcdef12.wav" + started = asyncio.Event() + killed = False + + class FakeProcess: + returncode = None + _release = asyncio.Event() + + def kill(self): + nonlocal killed + killed = True + self.returncode = -9 + self._release.set() + + async def communicate(self): + started.set() + if self.returncode is None: + await self._release.wait() + return b"", b"" + + async def fake_create_subprocess_exec(*args, **kwargs): + return FakeProcess() + + async def scenario(): + with patch( + "app.api.endpoints.agent.shutil.which", + return_value="/usr/bin/ffmpeg", + ), patch( + "app.api.endpoints.agent.uuid.uuid4", + return_value=SimpleNamespace(hex="abcdef1234567890"), + ), patch( + "app.api.endpoints.agent.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ), patch( + "app.api.endpoints.agent.get_api_runtime_config_snapshot", + return_value=SimpleNamespace(temp_path=tmp_path), + ): + conversion_task = asyncio.create_task( + _prepare_web_agent_audio_attachment_path_async(str(source_path)) + ) + await asyncio.wait_for(started.wait(), timeout=1) + conversion_task.cancel() + with pytest.raises(asyncio.CancelledError): + await conversion_task + + asyncio.run(scenario()) + + assert killed is True + assert not output_path.exists() + + +def test_prepare_web_agent_audio_attachment_async_communicate_error_reaps_process(tmp_path): + """ffmpeg 通信异常时应终止仍运行的进程并回退原文件。""" + source_path = tmp_path / "reply.opus" + source_path.write_bytes(b"opus-bytes") + started = asyncio.Event() + killed = False + communicate_calls = 0 + + class FakeProcess: + returncode = None + + def kill(self): + nonlocal killed + killed = True + self.returncode = -9 + + async def communicate(self): + nonlocal communicate_calls + communicate_calls += 1 + started.set() + if self.returncode is None: + raise OSError("pipe closed") + return b"", b"" + + async def fake_create_subprocess_exec(*args, **kwargs): + return FakeProcess() + + async def scenario(): + with patch( + "app.api.endpoints.agent.shutil.which", + return_value="/usr/bin/ffmpeg", + ), patch( + "app.api.endpoints.agent.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ), patch( + "app.api.endpoints.agent.get_api_runtime_config_snapshot", + return_value=SimpleNamespace(temp_path=tmp_path), + ): + output_path = await _prepare_web_agent_audio_attachment_path_async( + str(source_path) + ) + return output_path + + output_path = asyncio.run(scenario()) + + assert output_path == source_path + assert killed is True + assert communicate_calls == 2 + + +def test_prepare_web_agent_audio_attachment_async_cancellation_cleans_completed_output( + tmp_path, +): + """转码完成后检查产物期间取消，也应清理未登记的 WAV。""" + source_path = tmp_path / "reply.opus" + source_path.write_bytes(b"opus-bytes") + output_path = tmp_path / "voice" / "reply_web_abcdef12.wav" + exists_started = asyncio.Event() + + class FakeProcess: + returncode = 0 + + async def communicate(self): + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(b"wav-bytes") + return b"", b"" + + async def fake_create_subprocess_exec(*args, **kwargs): + return FakeProcess() + + async def fake_run_in_threadpool(func, *args, **kwargs): + if getattr(func, "__name__", "") == "exists": + exists_started.set() + await asyncio.Event().wait() + return await asyncio.to_thread(func, *args, **kwargs) + + async def scenario(): + with patch( + "app.api.endpoints.agent.shutil.which", + return_value="/usr/bin/ffmpeg", + ), patch( + "app.api.endpoints.agent.uuid.uuid4", + return_value=SimpleNamespace(hex="abcdef1234567890"), + ), patch( + "app.api.endpoints.agent.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ), patch( + "app.api.endpoints.agent.get_api_runtime_config_snapshot", + return_value=SimpleNamespace(temp_path=tmp_path), + ), patch( + "app.api.endpoints.agent.run_in_threadpool", + side_effect=fake_run_in_threadpool, + ): + conversion_task = asyncio.create_task( + _prepare_web_agent_audio_attachment_path_async(str(source_path)) + ) + await asyncio.wait_for(exists_started.wait(), timeout=1) + assert output_path.exists() + conversion_task.cancel() + with pytest.raises(asyncio.CancelledError): + await conversion_task + + asyncio.run(scenario()) + + assert not output_path.exists() + + +def test_prepare_web_agent_audio_attachment_async_timeout_falls_back(tmp_path): + """转码超时应回退原文件并回收 ffmpeg。""" + source_path = tmp_path / "reply.opus" + source_path.write_bytes(b"opus-bytes") + started = asyncio.Event() + killed = False + + class FakeProcess: + returncode = None + _release = asyncio.Event() + + def kill(self): + nonlocal killed + killed = True + self.returncode = -9 + self._release.set() + + async def communicate(self): + started.set() + if self.returncode is None: + await self._release.wait() + return b"", b"" + + async def fake_create_subprocess_exec(*args, **kwargs): + return FakeProcess() + + async def scenario(): + with patch( + "app.api.endpoints.agent.shutil.which", + return_value="/usr/bin/ffmpeg", + ), patch( + "app.api.endpoints.agent.asyncio.create_subprocess_exec", + side_effect=fake_create_subprocess_exec, + ), patch( + "app.api.endpoints.agent.get_api_runtime_config_snapshot", + return_value=SimpleNamespace(temp_path=tmp_path), + ), patch( + "app.api.endpoints.agent.WEB_AGENT_AUDIO_CONVERSION_TIMEOUT_SECONDS", + 0.01, + ): + conversion_task = asyncio.create_task( + _prepare_web_agent_audio_attachment_path_async(str(source_path)) + ) + await asyncio.wait_for(started.wait(), timeout=1) + return await conversion_task + + output_path = asyncio.run(scenario()) + + assert output_path == source_path + assert killed is True + + def test_transcribe_web_agent_audio_files_reads_registered_upload(tmp_path): """WebAgent 上传录音应从临时附件登记表读取并转写为文本。""" voice_path = tmp_path / "recording.webm"