fix(agent): isolate web and storage blocking I/O

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"