refactor(agent): route chat persistence through database worker

tests/conftest.py

@@ -118,6 +118,10 @@ def configure_plugin_system_services():
         configure_chain_runtime_context_provider,
     )
     from app.application.messaging.message import MessageHelper, MessageQueueManager
+    from app.application.messaging.chat import (
+        AgentChatPersistenceService,
+        configure_agent_chat_persistence,
+    )
     from app.runtime.cache import AsyncFileCache, FileCache
     from app.runtime.events import EventManager
     from app.runtime.extensions.module_manager import ModuleManager
@@ -255,6 +259,12 @@ def configure_plugin_system_services():
         workflow=lambda: WorkflowOper(),
         plugin_data=lambda: PluginDataOper(),
     )
+    configure_agent_chat_persistence(
+        AgentChatPersistenceService(
+            repository=AgentChatOper,
+            async_executor=database_executor,
+        )
+    )
     from app.adapters.external.market import (
         PluginHelper,
         VERSION_BACKWARD_COMPATIBLE_FLAGS,

tests/test_agent_background_output.py

@@ -249,7 +249,11 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         )
         agent.send_agent_message = AsyncMock()
 
-        with patch.object(memory_manager, "save_agent_messages") as save_messages:
+        with patch.object(
+            memory_manager,
+            "async_save_agent_messages",
+            new=AsyncMock(),
+        ) as save_messages:
             await agent._execute_agent([HumanMessage(content="测试")])
 
         save_messages.assert_called_once()
@@ -277,7 +281,9 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
 
         with (
             patch.object(
-                memory_manager, "get_agent_messages", return_value=cached_messages
+                memory_manager,
+                "async_get_agent_messages",
+                new=AsyncMock(return_value=cached_messages),
             ),
             patch.object(agent, "prepare_chat_title", new=AsyncMock()),
             patch.object(agent, "_save_display_history_messages"),
@@ -305,7 +311,11 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         )
         agent.send_agent_message = AsyncMock()
 
-        with patch.object(memory_manager, "save_agent_messages") as save_messages:
+        with patch.object(
+            memory_manager,
+            "async_save_agent_messages",
+            new=AsyncMock(),
+        ) as save_messages:
             await agent._execute_agent([])
 
         agent.send_agent_message.assert_awaited_once_with(

tests/test_agent_chat_persistence.py

@@ -0,0 +1,154 @@
+"""AgentChat 同步短事务经有界 worker 委托的应用端口测试。"""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+from uuid import uuid4
+from types import SimpleNamespace
+
+import pytest
+
+from app.application.messaging.chat import AgentChatPersistenceService
+from app.db.oper.agentchat import AgentChatOper
+from app.db.models.agentchat import AgentChat
+from app.db.worker import DatabaseWorker
+
+
+class _Executor:
+    """用独立线程模拟 G2B worker，验证调用方不会直接执行同步仓储。"""
+
+    def __init__(self) -> None:
+        self.calls = 0
+        self.worker_thread_id: int | None = None
+
+    async def run(self, operation):
+        """在线程中执行一个完整的同步操作。"""
+        self.calls += 1
+
+        def invoke():
+            self.worker_thread_id = threading.get_ident()
+            return operation()
+
+        return await asyncio.to_thread(invoke)
+
+
+class _Repository:
+    """记录 AgentChat 端口调用的同步仓储替身。"""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, dict]] = []
+
+    def get(self, **kwargs):
+        self.calls.append(("get", kwargs))
+        return SimpleNamespace(agent_messages=[])
+
+    def append_display_messages(self, **kwargs):
+        self.calls.append(("append_display_messages", kwargs))
+        return None
+
+    def save_display_messages(self, **kwargs):
+        self.calls.append(("save_display_messages", kwargs))
+        return None
+
+    def save_agent_messages(self, **kwargs):
+        self.calls.append(("save_agent_messages", kwargs))
+
+    def update_title_if_empty(self, **kwargs):
+        self.calls.append(("update_title_if_empty", kwargs))
+
+
+@pytest.mark.asyncio
+async def test_agent_chat_persistence_runs_sync_repository_inside_worker() -> None:
+    """同步查询和写入都必须经过一次 worker admission。"""
+    executor = _Executor()
+    repository = _Repository()
+    service = AgentChatPersistenceService(
+        repository=lambda: repository,
+        async_executor=executor,
+    )
+    caller_thread_id = threading.get_ident()
+
+    await service.async_get("session-1", user_id="1")
+    await service.async_append_display_messages(
+        session_id="session-1",
+        user_id="1",
+        messages=[{"role": "user", "content": "hello"}],
+    )
+    await service.async_save_display_messages(
+        session_id="session-1",
+        user_id="1",
+        messages=[],
+    )
+    await service.async_save_agent_messages(
+        session_id="session-1",
+        user_id="1",
+        messages=[],
+    )
+    await service.async_update_title_if_empty(
+        session_id="session-1",
+        user_id="1",
+        title="标题",
+    )
+
+    assert executor.calls == 5
+    assert executor.worker_thread_id != caller_thread_id
+    assert [name for name, _kwargs in repository.calls] == [
+        "get",
+        "append_display_messages",
+        "save_display_messages",
+        "save_agent_messages",
+        "update_title_if_empty",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_agent_chat_persistence_propagates_worker_failure() -> None:
+    """worker admission 或事务异常必须原样返回给 async 应用调用方。"""
+
+    class FailingExecutor:
+        async def run(self, _operation):
+            raise RuntimeError("worker failed")
+
+    service = AgentChatPersistenceService(
+        repository=_Repository,
+        async_executor=FailingExecutor(),
+    )
+
+    with pytest.raises(RuntimeError, match="worker failed"):
+        await service.async_save_agent_messages(
+            session_id="session-1",
+            user_id="1",
+            messages=[],
+        )
+
+
+@pytest.mark.asyncio
+async def test_agent_chat_persistence_uses_real_worker_and_sqlite_transaction() -> None:
+    """真实 AgentChat Oper 经 worker 写入后可被后续 worker 查询恢复。"""
+    worker = DatabaseWorker(max_workers=1, capacity=4)
+    await worker.start()
+    session_id = f"worker-{uuid4().hex}"
+    service = AgentChatPersistenceService(
+        repository=AgentChatOper,
+        async_executor=worker,
+    )
+
+    try:
+        await service.async_save_display_messages(
+            session_id=session_id,
+            user_id="worker-user",
+            username="worker-user",
+            channel="WebAgent",
+            source="worker-test",
+            messages=[{"role": "user", "content": "worker"}],
+        )
+        chat = await service.async_get(session_id, user_id="worker-user")
+        assert chat is not None
+        assert chat.message_count == 1
+        assert chat.display_messages[0]["content"] == "worker"
+    finally:
+        chat = AgentChatOper().get(session_id=session_id, user_id="worker-user")
+        if chat is not None:
+            AgentChat.delete(rid=chat.id)
+        await worker.shutdown()

tests/test_web_agent_stream.py

@@ -19,6 +19,7 @@ from app.api.endpoints.agent import (
     _build_web_agent_message_events,
     _build_web_agent_command_items,
     _build_web_agent_session_id,
+    _build_web_agent_session_id_async,
     _build_web_agent_traditional_callback_payload,
     _build_web_agent_display_message_from_events,
     _collect_web_agent_traditional_events,
@@ -174,6 +175,30 @@ def test_build_web_agent_session_id_reuses_accessible_history():
     assert _build_web_agent_session_id(user, "telegram-session") == "telegram-session"
 
 
+def test_build_web_agent_session_id_async_uses_worker_persistence():
+    """异步 Web 会话解析应通过 AgentChat worker 端口读取历史。"""
+    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
+    persistence = SimpleNamespace(
+        async_get=AsyncMock(
+            return_value=SimpleNamespace(
+                user_id="telegram-user",
+                username="tester",
+            )
+        )
+    )
+
+    with patch(
+        "app.api.endpoints.agent.get_configured_agent_chat_persistence",
+        return_value=persistence,
+    ):
+        session_id = asyncio.run(
+            _build_web_agent_session_id_async(user, "telegram-session")
+        )
+
+    assert session_id == "telegram-session"
+    persistence.async_get.assert_awaited_once_with("telegram-session")
+
+
 def test_apply_web_agent_display_event_updates_snapshot():
     """WebAgent SSE 事件应按到达顺序聚合为服务端展示快照。"""
     message = {
@@ -904,6 +929,7 @@ def test_web_agent_stream_emits_secret_result_only_as_protected_event():
             return_value=FakeProtectedAgent,
         ), patch(
             "app.api.endpoints.agent._save_web_agent_display_snapshot",
+            new_callable=AsyncMock,
         ) as save_snapshot:
             body = asyncio.run(scenario())
 
@@ -970,6 +996,7 @@ def test_web_agent_cancel_keeps_existing_display_history():
             new=AsyncMock(return_value="已取消敏感设置读取。"),
         ), patch(
             "app.api.endpoints.agent._save_web_agent_display_snapshot",
+            new_callable=AsyncMock,
         ) as save_snapshot:
             body = asyncio.run(scenario())
 
@@ -1113,6 +1140,7 @@ def test_web_agent_stream_drops_secret_result_after_disconnect():
             new=AsyncMock(side_effect=finish_after_disconnect),
         ) as process, patch(
             "app.api.endpoints.agent._save_web_agent_display_snapshot",
+            new_callable=AsyncMock,
         ) as save_snapshot:
             body = asyncio.run(scenario())
 
@@ -1156,7 +1184,7 @@ def test_web_agent_stream_emits_heartbeat_during_idle_tool_wait():
         "app.api.endpoints.agent._has_web_agent_traditional_interaction",
         return_value=False,
     ), patch(
-        "app.api.endpoints.agent._build_web_agent_session_id",
+        "app.api.endpoints.agent._build_web_agent_session_id_async",
         return_value="web-agent:heartbeat",
     ), patch.object(
         MessageChain,
@@ -1167,6 +1195,7 @@ def test_web_agent_stream_emits_heartbeat_during_idle_tool_wait():
         side_effect=slow_process_message,
     ), patch(
         "app.api.endpoints.agent._save_web_agent_display_snapshot",
+        new_callable=AsyncMock,
     ):
         body = asyncio.run(scenario())
 
@@ -1228,7 +1257,7 @@ def test_web_agent_stop_finishes_stream_without_error():
             "app.api.endpoints.agent._has_web_agent_traditional_interaction",
             return_value=False,
         ), patch(
-            "app.api.endpoints.agent._build_web_agent_session_id",
+            "app.api.endpoints.agent._build_web_agent_session_id_async",
             return_value=session_id,
         ), patch.object(
             MessageChain,
@@ -1238,6 +1267,7 @@ def test_web_agent_stop_finishes_stream_without_error():
             return_value=BlockingWebAgent,
         ), patch(
             "app.api.endpoints.agent._save_web_agent_display_snapshot",
+            new_callable=AsyncMock,
         ):
             body = asyncio.run(scenario())
     finally:
@@ -1275,9 +1305,10 @@ def test_web_agent_stream_rechecks_running_service_before_enqueue():
     ), patch(
         "app.api.endpoints.agent.get_running_agent_manager",
         side_effect=[stale_manager, None],
-    ), patch(
-        "app.api.endpoints.agent._save_web_agent_display_snapshot",
-    ):
+        ), patch(
+            "app.api.endpoints.agent._save_web_agent_display_snapshot",
+            new_callable=AsyncMock,
+        ):
         body = asyncio.run(scenario())
 
     assert '"type": "error"' in body
@@ -1299,11 +1330,11 @@ def test_web_agent_traditional_stream_keeps_alive_and_saves_after_done():
         await asyncio.sleep(0.035)
         return [{"type": "delta", "content": "状态正常"}]
 
-    def slow_snapshot(**_kwargs):
+    async def slow_snapshot(**_kwargs):
         """阻塞快照写入，便于断言 done 不等待落库。"""
-        snapshot_started.set()
-        snapshot_release.wait(timeout=2)
-        snapshot_finished.set()
+        await asyncio.to_thread(snapshot_started.set)
+        await asyncio.to_thread(snapshot_release.wait, 2)
+        await asyncio.to_thread(snapshot_finished.set)
 
     async def scenario():
         response = await web_agent_stream(payload, request, user)
@@ -1323,6 +1354,8 @@ def test_web_agent_traditional_stream_keeps_alive_and_saves_after_done():
             await asyncio.sleep(0.001)
         assert snapshot_started.is_set()
         assert not snapshot_finished.is_set()
+        snapshot_release.set()
+        await asyncio.to_thread(snapshot_finished.wait, 1)
         await iterator.aclose()
         return "".join(received)
 
@@ -1340,13 +1373,14 @@ def test_web_agent_traditional_stream_keeps_alive_and_saves_after_done():
             "app.api.endpoints.agent._get_web_agent_unknown_command_message",
             return_value=None,
         ), patch(
-            "app.api.endpoints.agent._build_web_agent_session_id",
+            "app.api.endpoints.agent._build_web_agent_session_id_async",
             return_value="web-agent:traditional-heartbeat",
         ), patch(
             "app.api.endpoints.agent._collect_web_agent_traditional_events",
             side_effect=slow_collect,
         ), patch(
             "app.api.endpoints.agent._save_web_agent_display_snapshot",
+            new_callable=AsyncMock,
             side_effect=slow_snapshot,
         ):
             body = asyncio.run(scenario())
@@ -1354,7 +1388,6 @@ def test_web_agent_traditional_stream_keeps_alive_and_saves_after_done():
         assert ": heartbeat\n\n" in body
         assert '"type": "delta"' in body
         assert '"type": "done"' in body
-        assert not snapshot_finished.is_set()
     finally:
         snapshot_release.set()
 
@@ -1374,11 +1407,11 @@ def test_web_agent_stream_sends_done_before_snapshot_persistence_finishes():
         """立即生成一段文本，随后进入终态。"""
         kwargs["output_callback"]("检查完成")
 
-    def slow_snapshot(**_kwargs):
+    async def slow_snapshot(**_kwargs):
         """阻塞快照写入，便于验证 done 的发送时机。"""
-        snapshot_started.set()
-        snapshot_release.wait(timeout=2)
-        snapshot_finished.set()
+        await asyncio.to_thread(snapshot_started.set)
+        await asyncio.to_thread(snapshot_release.wait, 2)
+        await asyncio.to_thread(snapshot_finished.set)
 
     async def scenario():
         response = await web_agent_stream(payload, request, user)
@@ -1398,6 +1431,8 @@ def test_web_agent_stream_sends_done_before_snapshot_persistence_finishes():
         assert snapshot_started.is_set()
         assert not snapshot_finished.is_set()
 
+        snapshot_release.set()
+        await asyncio.to_thread(snapshot_finished.wait, 1)
         await iterator.aclose()
         return "".join(received)
 
@@ -1412,7 +1447,7 @@ def test_web_agent_stream_sends_done_before_snapshot_persistence_finishes():
             "app.api.endpoints.agent._has_web_agent_traditional_interaction",
             return_value=False,
         ), patch(
-            "app.api.endpoints.agent._build_web_agent_session_id",
+            "app.api.endpoints.agent._build_web_agent_session_id_async",
             return_value="web-agent:snapshot",
         ), patch.object(
             MessageChain,
@@ -1423,12 +1458,12 @@ def test_web_agent_stream_sends_done_before_snapshot_persistence_finishes():
             side_effect=immediate_process_message,
         ), patch(
             "app.api.endpoints.agent._save_web_agent_display_snapshot",
+            new_callable=AsyncMock,
             side_effect=slow_snapshot,
         ):
             body = asyncio.run(scenario())
 
         assert '"type": "done"' in body
-        assert not snapshot_finished.is_set()
     finally:
         snapshot_release.set()