Merge remote-tracking branch 'origin/v3' into v3

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,

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:

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)

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)
 

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")

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(

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"