From f560c19d653f3931e75d40e9277577332fb1d25c Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 03:04:19 +0800 Subject: [PATCH] fix(agent): keep chat reads on native async path --- app/agent/memory/__init__.py | 13 +++++--- app/agent/orchestrator.py | 3 +- app/api/endpoints/agent.py | 8 ++--- app/application/messaging/chat.py | 48 +++++++++++----------------- tests/conftest.py | 3 ++ tests/test_agent_chat_history.py | 46 ++++++++++++++++++++++++++ tests/test_agent_chat_persistence.py | 35 ++++++++++---------- tests/test_agent_chat_transaction.py | 1 + tests/test_web_agent_stream.py | 15 +++++---- 9 files changed, 106 insertions(+), 66 deletions(-) diff --git a/app/agent/memory/__init__.py b/app/agent/memory/__init__.py index d65241926..8775e34b8 100644 --- a/app/agent/memory/__init__.py +++ b/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() - chat = await persistence.async_get( + service = get_configured_agent_chat_service() + 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 [] diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index c12995e16..249b3b0cb 100644 --- a/app/agent/orchestrator.py +++ b/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, ) diff --git a/app/api/endpoints/agent.py b/app/api/endpoints/agent.py index eef533bb7..c644da0ca 100644 --- a/app/api/endpoints/agent.py +++ b/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( - session_id - ) + existing_chat = await get_configured_agent_chat_service().get(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)), diff --git a/app/application/messaging/chat.py b/app/application/messaging/chat.py index f1e40b576..9bbf18fab 100644 --- a/app/application/messaging/chat.py +++ b/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, diff --git a/tests/conftest.py b/tests/conftest.py index e53f522d1..9b3395f0a 100644 --- a/tests/conftest.py +++ b/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, diff --git a/tests/test_agent_chat_history.py b/tests/test_agent_chat_history.py index b87ebe460..25d1207fc 100644 --- a/tests/test_agent_chat_history.py +++ b/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, + ) diff --git a/tests/test_agent_chat_persistence.py b/tests/test_agent_chat_persistence.py index 8b2cd6875..6501aaef3 100644 --- a/tests/test_agent_chat_persistence.py +++ b/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") - if chat is not None: - AgentChat.delete(rid=chat.id) + await AgentChatOper().async_delete( + session_id=session_id, + user_id="worker-user", + ) await worker.shutdown() diff --git a/tests/test_agent_chat_transaction.py b/tests/test_agent_chat_transaction.py index aa3ee1705..89e55b710 100644 --- a/tests/test_agent_chat_transaction.py +++ b/tests/test_agent_chat_transaction.py @@ -29,6 +29,7 @@ def _chat() -> SimpleNamespace: created_at=None, updated_at=None, display_messages=[], + agent_messages=[], ) diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 9b9efcbcb..01c0bd447 100644 --- a/tests/test_web_agent_stream.py +++ b/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(): - """异步 Web 会话解析应通过 AgentChat worker 端口读取历史。""" +def test_build_web_agent_session_id_async_uses_native_async_persistence(): + """异步 Web 会话解析应通过 native async 会话服务读取历史。""" user = SimpleNamespace(id=1, name="admin", is_superuser=True) - persistence = SimpleNamespace( - async_get=AsyncMock( + service = SimpleNamespace( + get=AsyncMock( return_value=SimpleNamespace( user_id="telegram-user", username="tester", + agent_messages=[], ) ) ) with patch( - "app.api.endpoints.agent.get_configured_agent_chat_persistence", - return_value=persistence, + "app.api.endpoints.agent.get_configured_agent_chat_service", + 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():