fix(agent): keep chat reads on native async path

app/agent/memory/__init__.py

@@ -10,7 +10,10 @@ from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
 from app.application.agentdata import AgentChatPort as AgentChatOper
-from app.application.messaging.chat import get_configured_agent_chat_persistence
+from app.application.messaging.chat import (
+    get_configured_agent_chat_persistence,
+    get_configured_agent_chat_service,
+)
 from app.runtime.log import logger
 from app.schemas.agent import ConversationMemory
 
@@ -109,19 +112,19 @@ class MemoryManager:
     async def async_get_agent_messages(
         self, session_id: str, user_id: str
     ) -> List[BaseMessage]:
-        """异步恢复 Agent 消息，持久化读取经有界数据库 worker 承接。"""
+        """异步恢复 Agent 消息，查询与会话应用服务保持同一异步端口。"""
         memory = self.get_memory(session_id, user_id)
         if memory:
             return memory.messages
 
         try:
-            persistence = get_configured_agent_chat_persistence()
+            service = get_configured_agent_chat_service()
-            chat = await persistence.async_get(
+            chat = await service.get(
                 session_id=session_id,
                 user_id=user_id,
             )
             if not chat:
-                chat = await persistence.async_get(session_id=session_id)
+                chat = await service.get(session_id=session_id)
         except Exception as e:
             logger.debug(f"读取持久化Agent会话失败: {e}")
             return []

app/agent/orchestrator.py

@@ -82,6 +82,7 @@ def _get_plugin_tools_revision() -> int:
 from app.application.agentdata import AgentTaskPort as AgentTaskOper
 from app.application.agentdata import UserPort as UserOper
 from app.application.messaging.chat import (
+    get_configured_agent_chat_service,
     get_configured_agent_chat_persistence,
     has_custom_agent_chat_title,
 )
@@ -579,7 +580,7 @@ class MoviePilotAgent:
             return
         self._tool_context["chat_title_prepared"] = True
         try:
-            chat = await get_configured_agent_chat_persistence().async_get(
+            chat = await get_configured_agent_chat_service().get(
                 session_id=self.session_id,
                 user_id=self.user_id,
             )

app/api/endpoints/agent.py

@@ -518,12 +518,12 @@ async def _build_web_agent_session_id_async(
     user: ApiPrincipal,
     session_id: Optional[str],
 ) -> str:
-    """异步解析 Web Agent 会话 ID，历史查询经有界数据库 worker 承接。"""
+    """异步解析 Web Agent 会话 ID，并复用异步会话查询端口。"""
     seed = str(session_id or "").strip() or uuid.uuid4().hex
     if seed.startswith(WEB_AGENT_SESSION_PREFIX):
         return seed
     try:
-        existing_chat = await get_configured_agent_chat_persistence().async_get(seed)
+        existing_chat = await get_configured_agent_chat_service().get(seed)
         if existing_chat and AgentChatService.can_access(existing_chat, user):
             return seed
     except Exception as e:
@@ -663,9 +663,7 @@ async def _save_web_agent_display_snapshot(
     保存 WebAgent 当前展示消息快照。
     """
     try:
-        existing_chat = await get_configured_agent_chat_persistence().async_get(
+        existing_chat = await get_configured_agent_chat_service().get(session_id)
-            session_id
-        )
         await get_configured_agent_chat_persistence().async_save_display_messages(
             session_id=session_id,
             user_id=(existing_chat.user_id if existing_chat else str(current_user.id)),

app/application/messaging/chat.py

@@ -85,14 +85,6 @@ class AsyncAgentChatRepository(Protocol):
 class SyncAgentChatRepository(Protocol):
     """仅包含 Agent 编排所需同步持久化方法的适配器端口。"""
 
-    def get(
-        self,
-        session_id: str,
-        user_id: Optional[str] = None,
-    ) -> Optional[Any]:
-        """读取服务端会话。"""
-        ...
-
     def append_display_messages(
         self,
         session_id: str,
@@ -165,6 +157,7 @@ class AgentChatRecord:
     created_at: Any
     updated_at: Any
     messages: list[dict]
+    agent_messages: list[dict]
 
 
 class AsyncUnitOfWork(Protocol):
@@ -222,9 +215,16 @@ class AgentChatService:
             return None
         return projected
 
-    async def get(self, session_id: str) -> Optional[AgentChatRecord]:
+    async def get(
+        self,
+        session_id: str,
+        user_id: Optional[str] = None,
+    ) -> Optional[AgentChatRecord]:
         """读取不附带授权判断的会话投影。"""
-        record = await self._repository.async_get(session_id=session_id)
+        record = await self._repository.async_get(
+            session_id=session_id,
+            user_id=user_id,
+        )
         if record is None:
             return None
         return self._project(record)
@@ -338,11 +338,12 @@ class AgentChatService:
             created_at=record.created_at,
             updated_at=record.updated_at,
             messages=list(record.display_messages or []),
+            agent_messages=list(record.agent_messages or []),
         )
 
 
 class AgentChatPersistenceService:
-    """把 Agent 编排所需的同步短事务委托给有界数据库 worker。"""
+    """把 Agent 编排所需的同步持久化操作委托给有界数据库 worker。"""
 
     def __init__(
         self,
@@ -354,24 +355,11 @@ class AgentChatPersistenceService:
         self._async_executor = async_executor
 
     async def _run(self, operation: Callable[[SyncAgentChatRepository], T]) -> T:
-        """在线程 worker 中执行一个完整的同步 AgentChat 短事务。"""
+        """在线程 worker 中执行一个同步 AgentChat 持久化操作。"""
         return await self._async_executor.run(
             lambda: operation(self._repository())
         )
 
-    async def async_get(
-        self,
-        session_id: str,
-        user_id: Optional[str] = None,
-    ) -> Optional[Any]:
-        """异步读取会话，实际查询由有界 worker 承接。"""
-        return await self._run(
-            lambda repository: repository.get(
-                session_id=session_id,
-                user_id=user_id,
-            )
-        )
-
     async def async_append_display_messages(
         self,
         *,
@@ -383,9 +371,9 @@ class AgentChatPersistenceService:
         source: Optional[str] = None,
         original_chat_id: Optional[str] = None,
         client_session_id: Optional[str] = None,
-    ) -> Optional[Any]:
+    ) -> None:
         """异步追加展示消息，等待同步事务取得确定终态。"""
-        return await self._run(
+        await self._run(
             lambda repository: repository.append_display_messages(
                 session_id=session_id,
                 user_id=user_id,
@@ -397,6 +385,7 @@ class AgentChatPersistenceService:
                 client_session_id=client_session_id,
             )
         )
+        return None
 
     async def async_save_display_messages(
         self,
@@ -409,9 +398,9 @@ class AgentChatPersistenceService:
         source: Optional[str] = None,
         original_chat_id: Optional[str] = None,
         client_session_id: Optional[str] = None,
-    ) -> Optional[Any]:
+    ) -> None:
         """异步保存展示消息快照，实际写入由有界 worker 承接。"""
-        return await self._run(
+        await self._run(
             lambda repository: repository.save_display_messages(
                 session_id=session_id,
                 user_id=user_id,
@@ -423,6 +412,7 @@ class AgentChatPersistenceService:
                 client_session_id=client_session_id,
             )
         )
+        return None
 
     async def async_save_agent_messages(
         self,

tests/conftest.py

@@ -119,7 +119,9 @@ def configure_plugin_system_services():
     )
     from app.application.messaging.message import MessageHelper, MessageQueueManager
     from app.application.messaging.chat import (
+        AgentChatService,
         AgentChatPersistenceService,
+        configure_agent_chat_service,
         configure_agent_chat_persistence,
     )
     from app.runtime.cache import AsyncFileCache, FileCache
@@ -265,6 +267,7 @@ def configure_plugin_system_services():
             async_executor=database_executor,
         )
     )
+    configure_agent_chat_service(AgentChatService(repository=AgentChatOper()))
     from app.adapters.external.market import (
         PluginHelper,
         VERSION_BACKWARD_COMPATIBLE_FLAGS,

tests/test_agent_chat_history.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 from langchain_core.messages import AIMessage, HumanMessage
 
@@ -278,3 +279,48 @@ def test_memory_manager_restores_agent_messages_from_database():
     assert len(messages) == 1
     assert isinstance(messages[0], HumanMessage)
     assert messages[0].content == "继续之前的话题"
+
+
+def test_async_memory_manager_restores_through_native_async_service(monkeypatch):
+    """异步记忆恢复只能通过会话应用服务的异步查询端口。"""
+    session_id = "session-memory-async"
+    user_id = "3"
+    memory_manager.clear_memory(session_id, user_id)
+    service = SimpleNamespace(
+        get=AsyncMock(
+            return_value=SimpleNamespace(
+                agent_messages=[
+                    {
+                        "type": "human",
+                        "data": {
+                            "content": "异步恢复",
+                            "additional_kwargs": {},
+                            "response_metadata": {},
+                            "type": "human",
+                            "name": None,
+                            "id": None,
+                            "example": False,
+                        },
+                    }
+                ]
+            )
+        )
+    )
+    monkeypatch.setattr(
+        "app.agent.memory.get_configured_agent_chat_service",
+        lambda: service,
+    )
+
+    messages = asyncio.run(
+        memory_manager.async_get_agent_messages(
+            session_id=session_id,
+            user_id=user_id,
+        )
+    )
+
+    assert len(messages) == 1
+    assert messages[0].content == "异步恢复"
+    service.get.assert_awaited_once_with(
+        session_id=session_id,
+        user_id=user_id,
+    )

tests/test_agent_chat_persistence.py

@@ -5,13 +5,11 @@ 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.application.messaging.chat import AgentChatPersistenceService, AgentChatService
 from app.db.oper.agentchat import AgentChatOper
-from app.db.models.agentchat import AgentChat
 from app.db.worker import DatabaseWorker
 
 
@@ -39,10 +37,6 @@ class _Repository:
     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
@@ -60,7 +54,7 @@ class _Repository:
 
 @pytest.mark.asyncio
 async def test_agent_chat_persistence_runs_sync_repository_inside_worker() -> None:
-    """同步查询和写入都必须经过一次 worker admission。"""
+    """同步 AgentChat 写入必须经过一次 worker admission。"""
     executor = _Executor()
     repository = _Repository()
     service = AgentChatPersistenceService(
@@ -69,7 +63,6 @@ async def test_agent_chat_persistence_runs_sync_repository_inside_worker() -> No
     )
     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",
@@ -91,10 +84,9 @@ async def test_agent_chat_persistence_runs_sync_repository_inside_worker() -> No
         title="标题",
     )
 
-    assert executor.calls == 5
+    assert executor.calls == 4
     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",
@@ -125,17 +117,18 @@ async def test_agent_chat_persistence_propagates_worker_failure() -> None:
 
 @pytest.mark.asyncio
 async def test_agent_chat_persistence_uses_real_worker_and_sqlite_transaction() -> None:
-    """真实 AgentChat Oper 经 worker 写入后可被后续 worker 查询恢复。"""
+    """真实 AgentChat Oper 经 worker 写入后可被 native async 查询恢复。"""
     worker = DatabaseWorker(max_workers=1, capacity=4)
     await worker.start()
     session_id = f"worker-{uuid4().hex}"
-    service = AgentChatPersistenceService(
+    persistence = AgentChatPersistenceService(
         repository=AgentChatOper,
         async_executor=worker,
     )
+    query = AgentChatService(repository=AgentChatOper())
 
     try:
-        await service.async_save_display_messages(
+        await persistence.async_save_display_messages(
             session_id=session_id,
             user_id="worker-user",
             username="worker-user",
@@ -143,12 +136,16 @@ async def test_agent_chat_persistence_uses_real_worker_and_sqlite_transaction()
             source="worker-test",
             messages=[{"role": "user", "content": "worker"}],
         )
-        chat = await service.async_get(session_id, user_id="worker-user")
+        chat = await query.get(
+            session_id,
+            user_id="worker-user",
+        )
         assert chat is not None
         assert chat.message_count == 1
-        assert chat.display_messages[0]["content"] == "worker"
+        assert chat.messages[0]["content"] == "worker"
     finally:
-        chat = AgentChatOper().get(session_id=session_id, user_id="worker-user")
+        await AgentChatOper().async_delete(
-        if chat is not None:
+            session_id=session_id,
-            AgentChat.delete(rid=chat.id)
+            user_id="worker-user",
+        )
         await worker.shutdown()

tests/test_agent_chat_transaction.py

@@ -29,6 +29,7 @@ def _chat() -> SimpleNamespace:
         created_at=None,
         updated_at=None,
         display_messages=[],
+        agent_messages=[],
     )
 
 

tests/test_web_agent_stream.py

@@ -175,28 +175,29 @@ 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():
+def test_build_web_agent_session_id_async_uses_native_async_persistence():
-    """异步 Web 会话解析应通过 AgentChat worker 端口读取历史。"""
+    """异步 Web 会话解析应通过 native async 会话服务读取历史。"""
     user = SimpleNamespace(id=1, name="admin", is_superuser=True)
-    persistence = SimpleNamespace(
+    service = SimpleNamespace(
-        async_get=AsyncMock(
+        get=AsyncMock(
             return_value=SimpleNamespace(
                 user_id="telegram-user",
                 username="tester",
+                agent_messages=[],
             )
         )
     )
 
     with patch(
-        "app.api.endpoints.agent.get_configured_agent_chat_persistence",
+        "app.api.endpoints.agent.get_configured_agent_chat_service",
-        return_value=persistence,
+        return_value=service,
     ):
         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")
+    service.get.assert_awaited_once_with("telegram-session")
 
 
 def test_apply_web_agent_display_event_updates_snapshot():