From 0cd6b3660f9fae084cbe2faed9c5352f7af6bc6a Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 8 Oct 2026 13:43:57 +0800 Subject: [PATCH] fix(agent): inject host dependencies into protocol factories Pass the session owner's data context and memory to Web and compatible protocol agent factories. Cover task creation, listing, session reuse and subagent tool injection across the shared entrypoints. Refs #6910 --- app/agent/session.py | 3 ++ tests/test_agent_data_ports.py | 90 +++++++++++++++++++++++++++++++++- 2 files changed, 92 insertions(+), 1 deletion(-) diff --git a/app/agent/session.py b/app/agent/session.py index 1396d5f81..dd492f4fb 100644 --- a/app/agent/session.py +++ b/app/agent/session.py @@ -685,6 +685,7 @@ class AgentSessionOwner: memory=self._memory, ) else: + # Web 与兼容协议工厂同样属于当前宿主，必须继承其数据和记忆依赖。 agent_kwargs: dict[str, Any] = { "session_id": session_id, "user_id": task.user_id, @@ -698,6 +699,8 @@ class AgentSessionOwner: "allow_message_tools": task.allow_message_tools, "output_callback": task.output_callback, "protected_output_callback": task.protected_output_callback, + "data": self._data, + "memory": self._memory, } if task.tool_event_callback is not None: agent_kwargs["tool_event_callback"] = task.tool_event_callback diff --git a/tests/test_agent_data_ports.py b/tests/test_agent_data_ports.py index 900ac41cc..f60d93d79 100644 --- a/tests/test_agent_data_ports.py +++ b/tests/test_agent_data_ports.py @@ -1,5 +1,6 @@ """Agent 数据上下文与类型化任务适配器测试。""" +import json from contextlib import asynccontextmanager from dataclasses import FrozenInstanceError, replace from types import SimpleNamespace @@ -11,6 +12,8 @@ import pytest import app.agent.tools.factory as tool_factory_module import app.application.agent as agent_services from app.agent.memory import MemoryManager +from app.agent.orchestrator import MoviePilotAgent +from app.agent.session import AgentSessionOwner, _MessageTask from app.agent.tools.factory import MoviePilotToolFactory from app.agent.tools.impl.agent_task import AgentTaskTool from app.agent.tools.manager import MoviePilotToolsManager @@ -24,7 +27,7 @@ from app.db.session import SessionFactory from app.scheduler import Scheduler from app.schemas.rule import CustomRule from app.schemas.system import FilterRuleGroup -from app.schemas.types import SystemConfigKey +from app.schemas.types import NotificationChannel, SystemConfigKey from app.startup.initializers import agent as agent_initializer @@ -314,6 +317,91 @@ def test_tool_factory_injects_context_only_into_builtin_tools(monkeypatch) -> No assert tools[0].data is context +@pytest.mark.asyncio +@pytest.mark.parametrize("entrypoint", ["channel", "web", "openai"]) +async def test_session_entrypoints_inject_data_and_memory_into_agent_tasks( + monkeypatch, entrypoint: str, +) -> None: + """渠道和协议工厂必须复用宿主上下文，支持同一会话创建及查询任务。""" + from app.agent.web import _get_web_agent_type + from app.api.endpoints.openai import _get_collecting_agent_type + + repository = TransactionalAgentTaskRepository(SessionFactory) + chat = MagicMock() + chat.get_sync.return_value = None + context = replace(_context(repository), chat=chat) + memory = MemoryManager() + owner = AgentSessionOwner(data=context, memory=memory) + factory = { + "channel": None, + "web": _get_web_agent_type, + "openai": _get_collecting_agent_type, + }[entrypoint] + agent_factory = factory() if factory is not None else None + channel = NotificationChannel.WebAgent.value if entrypoint == "web" else None + monkeypatch.setattr( + MoviePilotToolFactory, + "_get_builtin_tool_classes", + classmethod(lambda _cls, _channel=None: [AgentTaskTool]), + ) + monkeypatch.setattr(tool_factory_module, "_get_plugin_agent_tools", lambda: []) + monkeypatch.setattr( + "app.agent.tools.impl.agent_task.get_runtime_setting", + lambda key: {"AI_AGENT_ENABLE": True, "TZ": "UTC"}[key], + ) + register_job = MagicMock(return_value=None) + monkeypatch.setattr("app.application.scheduling.update_agent_task_job", register_job) + monkeypatch.setattr("app.application.scheduling.get_agent_task_next_run", lambda _id: None) + + async def run_blocking(_tool, _kind, function, *args, **kwargs): + """在测试临时库中执行同步仓储，避免创建全局阻塞执行器。""" + return function(*args, **kwargs) + + async def process(agent, action: str, **_kwargs) -> str: + """保留真实 Agent 与工具构造，只替换外部模型推理。""" + tool = agent._initialize_tools()[0] + if action == "create": + return await tool.run( + action="create", name="入口依赖回归", content="检查数据注入", + trigger_type="cron", trigger="0 8 * * *", + ) + return await tool.run(action="list") + + monkeypatch.setattr(AgentTaskTool, "run_blocking", run_blocking) + monkeypatch.setattr(MoviePilotAgent, "process", process) + task_kwargs = { + "session_id": f"data-context-{entrypoint}", + "user_id": f"data-context-user-{entrypoint}", + "source": entrypoint, + "channel": channel, + "agent_factory": agent_factory, + } + try: + created = json.loads(await owner._process_message_internal( + _MessageTask(message="create", **task_kwargs), + )) + agent = owner.active_agents[task_kwargs["session_id"]] + listed = json.loads(await owner._process_message_internal( + _MessageTask(message="list", **task_kwargs), + )) + + assert owner.active_agents[task_kwargs["session_id"]] is agent + assert agent._data is context + assert agent._memory is memory + assert agent._initialize_subagent_tools()[0].data is context + assert listed["total"] == 1 + assert listed["tasks"][0]["id"] == created["id"] + saved = repository.get(created["id"], user_id=task_kwargs["user_id"]) + assert saved is not None + assert saved.user_id == task_kwargs["user_id"] + assert saved.session_id == task_kwargs["session_id"] + assert saved.channel == channel + register_job.assert_called_once_with(created["id"]) + finally: + for task in repository.list(user_id=task_kwargs["user_id"]): + repository.delete(task.id, user_id=task_kwargs["user_id"]) + + def test_tool_manager_reassembly_invalidates_factory_catalog() -> None: """重复装配数据上下文时必须丢弃持有旧仓储的工厂工具快照。""" first = _context(TransactionalAgentTaskRepository(SessionFactory))