diff --git a/app/adapters/network/browser.py b/app/adapters/network/browser.py index aae633c65..bdfbf9f06 100644 --- a/app/adapters/network/browser.py +++ b/app/adapters/network/browser.py @@ -173,6 +173,7 @@ class _BrowserSessionState: active_index: int = 0 user_agent: Optional[str] = None cookies: Optional[str] = None + owner: Optional[Any] = field(default=None, repr=False) created_at: float = field(default_factory=time.monotonic) last_used_at: float = field(default_factory=time.monotonic) lock: threading.RLock = field(default_factory=threading.RLock) @@ -198,6 +199,7 @@ class BrowserSessionHelper: _sessions: dict[str, _BrowserSessionState] = {} _session_executors: dict[str, ThreadPoolExecutor] = {} + _session_executor_owners: dict[str, Any] = {} _session_thread_ids: dict[str, int] = {} _sessions_lock = threading.RLock() @@ -264,26 +266,28 @@ class BrowserSessionHelper: 关闭所有 Agent 浏览器会话。 """ with cls._sessions_lock: - session_keys = list( - set(cls._sessions.keys()) | set(cls._session_executors.keys()) - ) + session_keys = set(cls._sessions) | set(cls._session_executors) for session_key in session_keys: - cls.close_session(session_key) + with cls._sessions_lock: + session = cls._sessions.get(session_key) + owner = session.owner if session else cls._session_executor_owners.get(session_key) + cls.close_session(session_key, owner=owner) @classmethod - def close_session(cls, session_key: str) -> bool: + def close_session(cls, session_key: str, owner: Optional[Any] = None) -> bool: """ 关闭指定 Agent 浏览器会话。 :param session_key: 会话标识 + :param owner: 可选宿主作用域；提供时只允许关闭该作用域拥有的会话 :return: 找到并关闭会话时返回 True """ if cls._is_current_session_thread(session_key): - closed = cls._close_session_in_thread(session_key) + closed = cls._close_session_in_thread(session_key, owner=owner) cls._shutdown_session_executor(session_key, wait=False) return closed - executor = cls._get_existing_session_executor(session_key) + executor = cls._get_existing_session_executor(session_key, owner=owner) if executor: context = copy_context() # 会话线程保持空底层上下文，每次操作只使用当前调用快照。 @@ -294,16 +298,39 @@ class BrowserSessionHelper: session_key, cls._close_session_in_thread, session_key, + owner=owner, ) try: return future.result() finally: cls._shutdown_session_executor(session_key) - closed = cls._close_session_in_thread(session_key) + closed = cls._close_session_in_thread(session_key, owner=owner) cls._shutdown_session_executor(session_key) return closed + @classmethod + def close_owner(cls, owner: Any) -> bool: + """关闭宿主作用域创建的全部浏览器会话和其专属执行线程。""" + with cls._sessions_lock: + session_keys = { + key for key, session in cls._sessions.items() if session.owner is owner + } + session_keys.update( + key for key, session_owner in cls._session_executor_owners.items() + if session_owner is owner + ) + for session_key in session_keys: + cls.close_session(session_key, owner=owner) + with cls._sessions_lock: + return not any( + session.owner is owner + for session in cls._sessions.values() + ) and not any( + session_owner is owner + for session_owner in cls._session_executor_owners.values() + ) + def with_session( self, session_key: str, @@ -311,6 +338,7 @@ class BrowserSessionHelper: user_agent: Optional[str] = None, cookies: Optional[str] = None, timeout: Optional[int] = 30, + owner: Optional[Any] = None, ) -> Any: """ 获取或创建浏览器会话，并在持有会话锁时执行回调。 @@ -320,6 +348,7 @@ class BrowserSessionHelper: :param user_agent: 新建会话时使用的 User-Agent :param cookies: 本次操作要注入的 Cookie 请求头 :param timeout: 默认操作超时时间，单位秒 + :param owner: 可选宿主作用域；提供时会隔离并校验会话归属 :return: 回调函数返回值 """ self._prune_sessions() @@ -331,6 +360,7 @@ class BrowserSessionHelper: user_agent=user_agent, cookies=cookies, timeout=timeout, + owner=owner, ) def _with_session_in_thread( @@ -340,6 +370,7 @@ class BrowserSessionHelper: user_agent: Optional[str] = None, cookies: Optional[str] = None, timeout: Optional[int] = 30, + owner: Optional[Any] = None, ) -> Any: """ 在会话专属线程内获取浏览器会话并执行回调。 @@ -355,6 +386,7 @@ class BrowserSessionHelper: session_key=session_key, user_agent=user_agent, cookies=cookies, + owner=owner, ) with session.lock: session.last_used_at = time.monotonic() @@ -371,6 +403,7 @@ class BrowserSessionHelper: session_key: str, callback: Callable[..., Any], *args: Any, + owner: Optional[Any] = None, **kwargs: Any, ) -> Any: """ @@ -383,10 +416,10 @@ class BrowserSessionHelper: :return: 回调返回值 """ if cls._is_current_session_thread(session_key): - return callback(*args, **kwargs) + return callback(*args, owner=owner, **kwargs) for _ in range(2): - executor = cls._get_session_executor(session_key) + executor = cls._get_session_executor(session_key, owner=owner) try: context = copy_context() # 会话线程保持空底层上下文，每次操作只使用当前调用快照。 @@ -397,6 +430,7 @@ class BrowserSessionHelper: session_key, callback, *args, + owner=owner, **kwargs, ) except RuntimeError: @@ -447,7 +481,7 @@ class BrowserSessionHelper: return thread_id == threading.get_ident() @classmethod - def _get_session_executor(cls, session_key: str) -> ThreadPoolExecutor: + def _get_session_executor(cls, session_key: str, owner: Optional[Any] = None) -> ThreadPoolExecutor: """ 获取或创建指定会话的单线程执行器。 @@ -457,17 +491,21 @@ class BrowserSessionHelper: with cls._sessions_lock: executor = cls._session_executors.get(session_key) if executor: + existing_owner = cls._session_executor_owners.get(session_key) + if existing_owner is not owner: + raise PermissionError("浏览器会话不属于当前任务") return executor executor = ThreadPoolExecutor( max_workers=1, thread_name_prefix=cls.SESSION_WORKER_NAME_PREFIX, ) cls._session_executors[session_key] = executor + cls._session_executor_owners[session_key] = owner return executor @classmethod def _get_existing_session_executor( - cls, session_key: str + cls, session_key: str, owner: Optional[Any] = None ) -> Optional[ThreadPoolExecutor]: """ 获取指定会话已存在的执行器。 @@ -476,7 +514,10 @@ class BrowserSessionHelper: :return: 已存在的执行器，不存在时返回 None """ with cls._sessions_lock: - return cls._session_executors.get(session_key) + executor = cls._session_executors.get(session_key) + if executor and cls._session_executor_owners.get(session_key) is not owner: + raise PermissionError("浏览器会话不属于当前任务") + return executor @classmethod def _discard_session_executor( @@ -493,6 +534,7 @@ class BrowserSessionHelper: with cls._sessions_lock: if cls._session_executors.get(session_key) is executor: cls._session_executors.pop(session_key, None) + cls._session_executor_owners.pop(session_key, None) @classmethod def _shutdown_session_executor( @@ -508,19 +550,24 @@ class BrowserSessionHelper: """ with cls._sessions_lock: executor = cls._session_executors.pop(session_key, None) + cls._session_executor_owners.pop(session_key, None) cls._session_thread_ids.pop(session_key, None) if executor: executor.shutdown(wait=wait, cancel_futures=True) @classmethod - def _close_session_in_thread(cls, session_key: str) -> bool: + def _close_session_in_thread(cls, session_key: str, owner: Optional[Any] = None) -> bool: """ 在会话固定线程内关闭并移除浏览器会话。 :param session_key: 会话标识 + :param owner: 可选宿主作用域；提供时校验会话归属 :return: 找到并关闭会话时返回 True """ with cls._sessions_lock: + session = cls._sessions.get(session_key) + if session and session.owner is not owner: + raise PermissionError("浏览器会话不属于当前任务") session = cls._sessions.pop(session_key, None) if not session: return False @@ -731,10 +778,14 @@ class BrowserSessionHelper: session_key: str, user_agent: Optional[str] = None, cookies: Optional[str] = None, + owner: Optional[Any] = None, ) -> _BrowserSessionState: """复用匹配会话，或为新的会话键创建上下文。""" + self._validate_owner(owner) with self._sessions_lock: session = self._sessions.get(session_key) + if session and session.owner is not owner: + raise PermissionError("浏览器会话不属于当前任务") if session and user_agent and session.user_agent != user_agent: self._sessions.pop(session_key, None) self._close_session_state(session) @@ -756,16 +807,26 @@ class BrowserSessionHelper: pages=[page], user_agent=user_agent, cookies=cookies, + owner=owner, ) with self._sessions_lock: existing_session = self._sessions.get(session_key) if existing_session: + if existing_session.owner is not owner: + self._close_session_state(session) + raise PermissionError("浏览器会话不属于当前任务") self._close_session_state(session) return existing_session self._sessions[session_key] = session self._enforce_session_limit(protect_session_key=session_key) return session + @staticmethod + def _validate_owner(owner: Optional[Any]) -> None: + """拒绝已封口作用域创建或复用浏览器会话。""" + if owner is not None and bool(getattr(owner, "closed", False)): + raise PermissionError("浏览器会话所属任务已停止") + @classmethod def _prune_sessions(cls) -> None: """关闭超过空闲时限的浏览器会话。""" @@ -777,7 +838,10 @@ class BrowserSessionHelper: if now - session.last_used_at > cls.SESSION_TTL_SECONDS ] for session_key in expired_keys: - cls.close_session(session_key) + with cls._sessions_lock: + session = cls._sessions.get(session_key) + owner = session.owner if session else None + cls.close_session(session_key, owner=owner) @classmethod def _enforce_session_limit(cls, protect_session_key: Optional[str] = None) -> None: @@ -801,7 +865,8 @@ class BrowserSessionHelper: candidate_keys, key=lambda key: cls._sessions[key].last_used_at, ) - if not cls.close_session(oldest_key): + oldest_owner = cls._sessions[oldest_key].owner + if not cls.close_session(oldest_key, owner=oldest_owner): return @staticmethod diff --git a/app/agent/middleware/steering.py b/app/agent/middleware/steering.py new file mode 100644 index 000000000..3f3c6efe5 --- /dev/null +++ b/app/agent/middleware/steering.py @@ -0,0 +1,96 @@ +"""把运行中用户消息作为真实 HumanMessage 注入下一次模型调用。""" + +from __future__ import annotations + +import json +from typing import Any, Awaitable, Callable + +from langchain.agents.middleware import AgentMiddleware +from langchain.agents.middleware.types import ModelRequest, ModelResponse +from langchain_core.messages import AIMessage, HumanMessage + +from app.agent.steering import SteeringMessage, current_steering_inbox + +STEERING_MESSAGE_ID_KEY = "moviepilot_steering_message_id" + + +def _to_human_message(message: SteeringMessage) -> HumanMessage: + """将 steering 传输对象转换成可持久化、可审计的标准用户消息。""" + payload = { + "message": message.text, + "input": {"mode": "text", "steering": True}, + "images": [{"index": index + 1, "type": "image"} for index, _ in enumerate(message.images)], + "files": list(message.files), + } + content: list[dict[str, Any]] = [ + { + "type": "text", + "text": json.dumps(payload, ensure_ascii=False, indent=2), + } + ] + content.extend({"type": "image_url", "image_url": {"url": image}} for image in message.images) + return HumanMessage( + content=content, + additional_kwargs={STEERING_MESSAGE_ID_KEY: message.message_id}, + ) + + +class SteeringMiddleware(AgentMiddleware): # type: ignore[misc] + """在模型请求边界消费会话 inbox，避免并发启动第二条 Agent 图。""" + + @staticmethod + def _can_inject(messages: list[Any]) -> bool: + """工具调用尚未交付时不插入用户消息，保持 AI/Tool 配对协议完整。""" + return not (messages and isinstance(messages[-1], AIMessage) and bool(messages[-1].tool_calls)) + + @staticmethod + def _append_messages( + request: ModelRequest[Any], + messages: tuple[SteeringMessage, ...], + ) -> ModelRequest[Any]: + """把已消费消息写入图状态和本次模型请求，确保可持久化且只出现一次。""" + if not messages: + return request + injected = [_to_human_message(message) for message in messages] + request_messages = list(request.messages) + state_messages = request.state.get("messages") + if isinstance(state_messages, list): + state_messages.extend(injected) + request_messages.extend(injected) + return request.override(messages=request_messages) + + def _prepare_request( + self, + request: ModelRequest[Any], + consume: Callable[[], tuple[SteeringMessage, ...]], + ) -> ModelRequest[Any]: + """同步消费并装配消息；工具调用未完成时保留队列等待下一次请求。""" + if not self._can_inject(list(request.messages)): + return request + inbox = current_steering_inbox() + messages = consume() if inbox is not None else () + return self._append_messages(request, messages) + + def wrap_model_call( + self, + request: ModelRequest[Any], + handler: Callable[[ModelRequest[Any]], ModelResponse[Any]], + ) -> ModelResponse[Any]: + """同步模型调用前注入真实 HumanMessage，不增加 LangGraph 递归节点。""" + inbox = current_steering_inbox() + consume = inbox.consume_nowait if inbox is not None else lambda: () + return handler(self._prepare_request(request, consume)) + + async def awrap_model_call( + self, + request: ModelRequest[Any], + handler: Callable[[ModelRequest[Any]], Awaitable[ModelResponse[Any]]], + ) -> ModelResponse[Any]: + """异步模型调用前原子注入真实 HumanMessage，不启动并行 Agent。""" + inbox = current_steering_inbox() + if not self._can_inject(list(request.messages)) or inbox is None: + return await handler(request) + return await handler(self._append_messages(request, await inbox.consume())) + + +__all__ = ["STEERING_MESSAGE_ID_KEY", "SteeringMiddleware"] diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index 06db35eab..3a323c4f5 100644 --- a/app/agent/orchestrator.py +++ b/app/agent/orchestrator.py @@ -41,6 +41,7 @@ from app.agent.middleware.plan import PLAN_SNAPSHOT_KEY, PlanMiddleware, attach_ from app.agent.middleware.policy import AgentPolicyMiddleware from app.agent.middleware.selection import ToolSelectorMiddleware from app.agent.middleware.skills import SkillsMiddleware +from app.agent.middleware.steering import SteeringMiddleware from app.agent.middleware.subagents import ( create_subagent_middlewares, is_subagent_stream_metadata, @@ -63,6 +64,11 @@ from app.agent.policy.registry import requests_system_setting_secrets from app.agent.policy.sanitizer import sanitize_for_host from app.agent.prompt import prompt_manager from app.agent.runtime import agent_runtime_manager +from app.agent.steering import ( + SteeringInbox, + bind_steering_inbox, + reset_steering_inbox, +) from app.agent.terminal.ownership import ( TerminalAccessError, TerminalScope, @@ -388,6 +394,7 @@ class MoviePilotAgent: self._terminal_scope = TerminalScope( user_id=user_id or "", task_id=session_id, kind="conversation" ) + self._steering_inbox = SteeringInbox(session_id, str(user_id or "")) self._scheduled_terminal_scopes: set[TerminalScope] = set() self.channel = channel self.source = source @@ -417,6 +424,17 @@ class MoviePilotAgent: # 流式token管理 self.stream_handler = StreamingHandler() + @property + def steering_inbox(self) -> SteeringInbox: + """返回当前会话运行中消息的唯一 inbox。""" + return self._steering_inbox + + def configure_steering_inbox(self, inbox: SteeringInbox) -> None: + """把会话 owner 装配的 inbox 绑定到持久 Agent 实例。""" + if inbox.session_id != self.session_id or inbox.user_id != str(self.user_id or ""): + raise ValueError("steering inbox 与 Agent 会话身份不匹配") + self._steering_inbox = inbox + @classmethod def build_display_message( cls, @@ -1921,6 +1939,8 @@ class MoviePilotAgent: catalog=tool_catalog, tools=tools, ), + # 运行中补充消息只在模型回合边界进入同一张图，不启动并行 Agent。 + *([SteeringMiddleware()] if self._steering_inbox.running else []), output_middleware, *invocation_middlewares, # Skills @@ -2018,10 +2038,14 @@ class MoviePilotAgent: raise TerminalAccessError() if scope is not self._terminal_scope: self._scheduled_terminal_scopes.add(scope) - with bind_terminal_scope(scope): - return await self._process( - message, images=images, files=files, has_audio_input=has_audio_input - ) + steering_token = bind_steering_inbox(self._steering_inbox) + try: + with bind_terminal_scope(scope): + return await self._process( + message, images=images, files=files, has_audio_input=has_audio_input + ) + finally: + reset_steering_inbox(steering_token) async def release_terminal_scope(self, scope: TerminalScope) -> bool: """收口宿主临时任务的终端；未真实收敛的作用域留给 Agent 清理重试。""" @@ -2505,6 +2529,7 @@ class MoviePilotAgent: 清理智能体资源；子代理或终端未真实收敛时保留 owner 并返回 False。 """ self.begin_shutdown() + await self._steering_inbox.close() children_closed = await self._invalidate_cached_agent() terminals_closed = await close_terminal_scope(self._terminal_scope) for scope in tuple(self._scheduled_terminal_scopes): diff --git a/app/agent/session.py b/app/agent/session.py index 33b766edb..3ebf3e943 100644 --- a/app/agent/session.py +++ b/app/agent/session.py @@ -8,6 +8,7 @@ from typing import Any, Awaitable, Callable, Dict, List, Optional, cast from app.agent.contracts import ReplyMode from app.agent.memory import MemoryManager, memory_manager from app.agent.orchestrator import MoviePilotAgent, _SessionUsageSnapshot +from app.agent.steering import SteeringInbox, SteeringMessage, SteeringStatusCallback from app.agent.terminal.ownership import TerminalScope, close_terminal_scope from app.application.agent import AgentDataContext from app.chain.agent import AgentChain @@ -114,6 +115,8 @@ class _MessageTask: completion_future: Optional[asyncio.Future[str]] = None enqueued_at: Optional[float] = None scheduled_run_id: Optional[str] = None + steering_status_callback: Optional[SteeringStatusCallback] = None + steering_inbox: Optional[SteeringInbox] = field(default=None, init=False, repr=False) terminal_scope: Optional[TerminalScope] = field(default=None, init=False) agent: Optional[MoviePilotAgent] = field(default=None, init=False, repr=False) terminal_released: bool = field(default=False, init=False) @@ -178,6 +181,8 @@ class AgentSessionOwner: self._session_shutdown_pending: Dict[str, asyncio.Task[None]] = {} self._session_cleanup_pending: set[str] = set() self._session_deferred_cleanup_tasks: Dict[str, asyncio.Task[None]] = {} + self._session_active_tasks: Dict[str, _MessageTask] = {} + self._session_steering_inboxes: Dict[str, SteeringInbox] = {} self._session_cancel_requested: set[str] = set() self._close_finalizer_task: Optional[asyncio.Task[None]] = None self._closed = False @@ -221,6 +226,9 @@ class AgentSessionOwner: session_id in self._session_workers and not self._session_workers[session_id].done() ) + inbox = self._session_steering_inboxes.get(session_id) + status["steering_pending"] = inbox.pending_count if inbox else 0 + status["steering_active"] = bool(inbox and inbox.running) return status def matches_secret_confirmation( @@ -293,6 +301,7 @@ class AgentSessionOwner: output_callback: Optional[Callable[[str], None]] = None, protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None, message_callback: Optional[Callable[[Any], Awaitable[None] | None]] = None, + steering_status_callback: Optional[SteeringStatusCallback] = None, agent_factory: Optional[Callable[..., MoviePilotAgent]] = None, agent_setup: Optional[Callable[[MoviePilotAgent], None]] = None, wait_for_completion: bool = False, @@ -323,6 +332,7 @@ class AgentSessionOwner: output_callback=output_callback, protected_output_callback=protected_output_callback, message_callback=message_callback, + steering_status_callback=steering_status_callback, agent_factory=agent_factory, agent_setup=agent_setup, completion_future=completion_future, @@ -399,6 +409,28 @@ class AgentSessionOwner: raise return "" + async def submit_steering_message( + self, + *, + session_id: str, + user_id: str, + message: str, + images: Optional[List[str]] = None, + files: Optional[List[dict[str, Any]]] = None, + ) -> Optional[SteeringMessage]: + """把运行中补充消息提交到当前 Agent；没有活动运行时返回 None 让入口走普通轮次。""" + async with self._lifecycle_lock: + task = self._session_active_tasks.get(session_id) + inbox = self._session_steering_inboxes.get(session_id) + if task is None or task.user_id != str(user_id) or inbox is None: + return None + return await inbox.enqueue( + user_id=str(user_id), + text=message, + images=images, + files=files, + ) + @staticmethod async def _close_scheduled_task_scope(task: _MessageTask) -> bool: """取消等待与 worker 收尾共用任务归属，失败时保留已装配 Agent 的重试记录。""" @@ -457,6 +489,7 @@ class AgentSessionOwner: 3, ) await self._start_task_processing_status(task) + self._session_active_tasks[session_id] = task record_metric( "agent.active_tasks", 1, @@ -497,6 +530,8 @@ class AgentSessionOwner: try: await self._close_scheduled_task_scope(task) finally: + if self._session_active_tasks.get(session_id) is task: + self._session_active_tasks.pop(session_id, None) await self._finish_task_processing_status(task) queue.task_done() if session_id in self._session_cancel_requested: @@ -576,6 +611,7 @@ class AgentSessionOwner: f"Agent 会话 {session_id} 仍有子代理或终端在停止" ) self.active_agents.pop(session_id, None) + self._session_steering_inboxes.pop(session_id, None) if session_id not in self.active_agents: logger.info( @@ -638,6 +674,16 @@ class AgentSessionOwner: agent.set_message_callback(task.message_callback) task.agent = agent + inbox = self._session_steering_inboxes.get(session_id) + if inbox is None: + inbox = SteeringInbox(session_id, str(task.user_id)) + self._session_steering_inboxes[session_id] = inbox + task.steering_inbox = inbox + configure_inbox = getattr(agent, "configure_steering_inbox", None) + if callable(configure_inbox): + configure_inbox(inbox) + else: + setattr(agent, "steering_inbox", inbox) if task.agent_setup is not None: task.agent_setup(agent) @@ -649,7 +695,51 @@ class AgentSessionOwner: process_kwargs["has_audio_input"] = True if task.terminal_scope is not None: process_kwargs["terminal_scope"] = task.terminal_scope - return await agent.process(task.message, **process_kwargs) + await inbox.begin_run(task.steering_status_callback) + try: + result = await agent.process(task.message, **process_kwargs) + # 模型最终回合返回后仍可能有一条已入 inbox 的消息；在同一 worker + # 内继续处理，避免 WebAgent 已经 ACK 的补充消息变成无展示的后台轮次。 + while True: + pending = await inbox.consume() + if not pending: + break + for message in pending: + result = await agent.process( + message.text, + images=list(message.images) or None, + files=list(message.files) or None, + has_audio_input=any( + str(file.get("mime_type") or "").startswith("audio/") + for file in message.files + ), + terminal_scope=task.terminal_scope, + ) + # consume 与 finish_run 之间仍有一个极窄的入队窗口；已接受的消息 + # 也在当前 worker 内执行，且先发布 applied 以完成前端展示收口。 + pending = await inbox.finish_run() + if pending: + callback = task.steering_status_callback + for message in pending: + if callback is not None: + try: + callback(message, "applied") + except Exception: + pass + result = await agent.process( + message.text, + images=list(message.images) or None, + files=list(message.files) or None, + has_audio_input=any( + str(file.get("mime_type") or "").startswith("audio/") + for file in message.files + ), + terminal_scope=task.terminal_scope, + ) + return result + except BaseException: + await inbox.finish_run() + raise async def stop_current_task(self, session_id: str) -> bool: """ @@ -772,6 +862,7 @@ class AgentSessionOwner: ) return del self.active_agents[session_id] + self._session_steering_inboxes.pop(session_id, None) self._memory.clear_memory(session_id, user_id) logger.info(f"会话 {session_id} 的记忆已清空") @@ -841,6 +932,7 @@ class AgentSessionOwner: ) return self.active_agents.pop(session_id, None) + self._session_steering_inboxes.pop(session_id, None) self._memory.clear_memory(session_id, user_id) logger.info(f"会话 {session_id} 的记忆已清空") diff --git a/app/agent/steering.py b/app/agent/steering.py new file mode 100644 index 000000000..a0428d2b1 --- /dev/null +++ b/app/agent/steering.py @@ -0,0 +1,189 @@ +"""Agent 运行中消息的会话级排队、注入与状态回调。""" + +from __future__ import annotations + +import asyncio +import contextvars +import uuid +from collections import deque +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Callable, Deque, Optional + +STEERING_QUEUE_MAX_SIZE = 8 + + +@dataclass(frozen=True, slots=True) +class SteeringMessage: + """保存一条等待注入当前 Agent 运行的用户消息。""" + + message_id: str + session_id: str + user_id: str + text: str + images: tuple[str, ...] = () + files: tuple[dict[str, Any], ...] = () + created_at: str = "" + + @classmethod + def create( + cls, + *, + session_id: str, + user_id: str, + text: str, + images: Optional[list[str]] = None, + files: Optional[list[dict[str, Any]]] = None, + ) -> "SteeringMessage": + """生成宿主侧消息 ID，并复制输入附件以隔离调用方可变对象。""" + return cls( + message_id=uuid.uuid4().hex, + session_id=session_id, + user_id=user_id, + text=str(text or ""), + images=tuple(str(image) for image in (images or [])), + files=tuple(dict(file) for file in (files or [])), + created_at=datetime.now(timezone.utc).isoformat(), + ) + + +SteeringStatusCallback = Callable[[SteeringMessage, str], None] + + +class SteeringInbox: + """为一个用户会话管理运行中消息，并以锁定义提交与收尾的原子边界。""" + + def __init__(self, session_id: str, user_id: str) -> None: + """创建空 inbox；消息只接受固定会话和用户身份。""" + self.session_id = session_id + self.user_id = str(user_id) + self._pending: Deque[SteeringMessage] = deque() + self._lock = asyncio.Lock() + self._running = False + self._closed = False + self._status_callback: Optional[SteeringStatusCallback] = None + + @property + def running(self) -> bool: + """返回当前运行是否仍接受 steering 消息。""" + return self._running and not self._closed + + async def begin_run(self, status_callback: Optional[SteeringStatusCallback] = None) -> None: + """原子打开一次运行窗口，并绑定当前运行的状态发布回调。""" + async with self._lock: + if self._closed: + return + self._running = True + self._status_callback = status_callback + + async def finish_run(self) -> tuple[SteeringMessage, ...]: + """原子关闭运行窗口并返回收尾竞态中尚未注入的消息。""" + async with self._lock: + self._running = False + self._status_callback = None + messages = tuple(self._pending) + self._pending.clear() + return messages + + async def enqueue( + self, + *, + user_id: str, + text: str, + images: Optional[list[str]] = None, + files: Optional[list[dict[str, Any]]] = None, + ) -> Optional[SteeringMessage]: + """在运行窗口内排队消息；窗口结束的瞬间会原子地回退到普通轮次。""" + async with self._lock: + if self._closed or not self._running or str(user_id) != self.user_id: + return None + if len(self._pending) >= STEERING_QUEUE_MAX_SIZE: + return None + message = SteeringMessage.create( + session_id=self.session_id, + user_id=self.user_id, + text=text, + images=images, + files=files, + ) + self._pending.append(message) + return message + + async def consume(self) -> tuple[SteeringMessage, ...]: + """在下一次模型调用前一次性取出消息，并通知原 SSE 已应用。""" + async with self._lock: + if not self._pending: + return () + messages = tuple(self._pending) + self._pending.clear() + callback = self._status_callback + if callback is not None: + for message in messages: + try: + callback(message, "applied") + except Exception: + # 状态展示失败不能阻断真实模型请求；消息已经进入图状态。 + continue + return messages + + def consume_nowait(self) -> tuple[SteeringMessage, ...]: + """为同步 Agent 调用提供无等待消费；并发持锁时安全地等待下一轮。""" + if self._lock.locked() or not self._pending: + return () + messages = tuple(self._pending) + self._pending.clear() + callback = self._status_callback + if callback is not None: + for message in messages: + try: + callback(message, "applied") + except Exception: + continue + return messages + + @property + def pending_count(self) -> int: + """返回当前尚未进入模型上下文的消息数量。""" + return len(self._pending) + + async def close(self) -> tuple[SteeringMessage, ...]: + """关闭 inbox 并返回仍待处理的消息，供 owner 在停止路径决定收口策略。""" + async with self._lock: + self._closed = True + self._running = False + self._status_callback = None + messages = tuple(self._pending) + self._pending.clear() + return messages + + +_current_steering_inbox: contextvars.ContextVar[Optional[SteeringInbox]] = contextvars.ContextVar( + "moviepilot_current_steering_inbox", + default=None, +) + + +def bind_steering_inbox(inbox: Optional[SteeringInbox]) -> contextvars.Token[Optional[SteeringInbox]]: + """把当前 Agent 图执行绑定到对应会话的 inbox。""" + return _current_steering_inbox.set(inbox) + + +def reset_steering_inbox(token: contextvars.Token[Optional[SteeringInbox]]) -> None: + """恢复图执行前的 inbox 上下文。""" + _current_steering_inbox.reset(token) + + +def current_steering_inbox() -> Optional[SteeringInbox]: + """读取当前模型调用所属的会话 inbox。""" + return _current_steering_inbox.get() + + +__all__ = [ + "STEERING_QUEUE_MAX_SIZE", + "SteeringInbox", + "SteeringMessage", + "SteeringStatusCallback", + "bind_steering_inbox", + "current_steering_inbox", + "reset_steering_inbox", +] diff --git a/app/agent/terminal/ownership.py b/app/agent/terminal/ownership.py index 58beadd9c..7e46a0874 100644 --- a/app/agent/terminal/ownership.py +++ b/app/agent/terminal/ownership.py @@ -97,9 +97,16 @@ def require_terminal_scope() -> TerminalScope: async def close_terminal_scope(scope: TerminalScope) -> bool: - """只收敛已装配的终端管理器，无终端任务封口时不物化进程能力。""" + """收敛作用域拥有的终端和浏览器资源，无资源时不物化管理器。""" scope.seal() module = sys.modules.get("app.agent.terminal.manager") manager = getattr(module, "terminal_session_manager", None) manager_closed = True if manager is None else bool(await manager.close_owner(scope)) - return manager_closed and await scope.wait_runs() + browser_module = sys.modules.get("app.adapters.network.browser") + browser_helper = getattr(browser_module, "BrowserSessionHelper", None) + browser_closed = ( + True + if browser_helper is None + else bool(await asyncio.to_thread(browser_helper.close_owner, scope)) + ) + return manager_closed and browser_closed and await scope.wait_runs() diff --git a/app/agent/tools/impl/browse_webpage.py b/app/agent/tools/impl/browse_webpage.py index d3dd7384a..e105344b9 100644 --- a/app/agent/tools/impl/browse_webpage.py +++ b/app/agent/tools/impl/browse_webpage.py @@ -12,6 +12,7 @@ from pydantic import BaseModel, Field from app.adapters.network.browser import BrowserSessionHelper from app.agent.policy.contracts import ExecutionOutcome +from app.agent.terminal.ownership import current_terminal_scope from app.agent.tools.base import MoviePilotTool from app.agent.tools.result import inspect_tool_result from app.agent.tools.tags import ToolTag @@ -346,8 +347,9 @@ class BrowseWebpageTool(MoviePilotTool): """在同步上下文中执行 CloakBrowser 浏览器操作""" try: + owner = current_terminal_scope() if browser_action == BrowserAction.CLOSE_SESSION: - closed = BrowserSessionHelper.close_session(session_key) + closed = BrowserSessionHelper.close_session(session_key, owner=owner) message = "浏览器会话已关闭" if closed else "浏览器会话不存在" return self._json_response( { @@ -387,6 +389,7 @@ class BrowseWebpageTool(MoviePilotTool): user_agent=user_agent, cookies=cookies, timeout=timeout, + owner=owner, ) except Exception as e: diff --git a/app/agent/tools/impl/execute_command.py b/app/agent/tools/impl/execute_command.py index 03633d7e0..7448f0ff3 100644 --- a/app/agent/tools/impl/execute_command.py +++ b/app/agent/tools/impl/execute_command.py @@ -615,7 +615,11 @@ class ExecuteCommandTool(MoviePilotTool): {wait_task, scope_task}, timeout=normalized_timeout, return_when=asyncio.FIRST_COMPLETED, ) - scope_cancelled = scope_task in done + # The scope can be sealed in the same event-loop turn after + # ``wait_task`` wins. Recheck the authoritative owner state so + # a run that crossed the cancellation boundary never reports + # an unqualified success. + scope_cancelled = scope_task in done or scope.closed if wait_task not in done: timed_out = not scope_cancelled await self._cleanup_process(process, wait_task) diff --git a/app/application/messaging/agent.py b/app/application/messaging/agent.py index d595e2fb7..600968661 100644 --- a/app/application/messaging/agent.py +++ b/app/application/messaging/agent.py @@ -43,10 +43,9 @@ from app.application.messaging.chat import ( from app.application.messaging.router import has_pending_interaction from app.runtime.execution import run_in_threadpool from app.runtime.log import logger -from app.runtime.stop import runtime_stop_state from app.runtime.tasks import get_task_registry from app.schemas.message import Message -from app.schemas.types import NotificationChannel, ReplyMode +from app.schemas.types import NotificationChannel __all__ = ["dispatch_command"] # Agent 选择按钮回调前缀（新旧两种格式都必须继续兼容） @@ -1927,157 +1926,6 @@ def _build_traditional_web_agent_stream( return event_generator() -def _build_agent_web_agent_stream( - *, - command: WebAgentStreamCommand, - current_user: AgentChatPrincipal, - session_id: str, - prompt: str, - display_prompt: str, - has_audio_input: bool, - is_secret_confirmation_control: bool, - protected_transport_supported: bool, - service: AgentChatService, - persistence: AgentChatPersistenceService, - is_disconnected: Callable[[], Awaitable[bool]], -) -> AsyncIterator[dict[str, Any]]: - """构造标准 Agent 执行链路的 WebAgent 事件流。""" - bind_web_agent_user_session(str(current_user.id), session_id) - event_publisher = WebAgentEventPublisher() - user_attachments = build_web_agent_input_attachments( - images=command.images, - files=command.files, - audio_refs=command.audio_refs, - ) - display_messages = [] - if command.echo_user and not is_secret_confirmation_control: - user_display_message = build_web_agent_display_message( - role="user", - content=display_prompt or prompt, - attachments=user_attachments, - ) - if command.choice_selection: - user_display_message["choice_selection"] = command.choice_selection - display_messages.append(user_display_message) - assistant_display_message = build_web_agent_display_message( - role="assistant", - status="streaming", - ) - display_messages.append(assistant_display_message) - - def output_callback(delta: str) -> None: - """接收 Agent 文本增量并投影为展示事件。""" - for item in split_web_agent_output(delta): - apply_web_agent_display_event(item, assistant_display_message) - event_publisher.publish(item) - - async def message_callback(message: Message) -> None: - """接收 Agent 工具主动发送的 Web 通知。""" - for item in await build_web_agent_message_events_async(message): - apply_web_agent_display_event(item, assistant_display_message) - event_publisher.publish(item) - - def protected_output_callback(content: str) -> bool: - """发布不进入普通展示快照的敏感交互结果。""" - return event_publisher.publish({"type": "interaction-protected", "content": content}) - - async def event_generator() -> AsyncIterator[dict[str, Any]]: - """执行 Agent 并按断线与终态语义消费事件。""" - audio_ref_set = set(command.audio_refs) - files = [file for file in command.files if str(file.get("ref") or "") not in audio_ref_set] - files.extend({"ref": audio_ref, "mime_type": "audio/*"} for audio_ref in command.audio_refs) - - async def run_agent() -> None: - """后台执行 Agent，并在完成后持久化展示快照。""" - try: - runtime_manager = agent_application.get_running_agent_manager() - if runtime_manager is None: - raise RuntimeError("智能助手服务尚未就绪，请稍后重试。") - await runtime_manager.process_message( - session_id=session_id, - user_id=str(current_user.id), - message=prompt, - images=command.images, - files=files or None, - has_audio_input=has_audio_input, - channel=NotificationChannel.WebAgent.value, - source=WEB_AGENT_SOURCE, - username=current_user.name, - reply_mode=ReplyMode.CAPTURE_ONLY, - allow_message_tools=True, - output_callback=output_callback, - protected_output_callback=(protected_output_callback if protected_transport_supported else None), - message_callback=message_callback, - agent_factory=agent_application.get_web_agent_type(), - wait_for_completion=True, - ) - except asyncio.CancelledError: - # 显式停止会话沿用正常终止语义；服务关闭由 manager 的稳定异常分支处理。 - pass - except Exception as err: - logger.error(f"Web智能助手执行失败: {str(err)}") - error_event = { - "type": "error", - "message": "智能助手执行失败，请稍后重试", - } - apply_web_agent_display_event(error_event, assistant_display_message) - event_publisher.publish(error_event) - finally: - done_event = {"type": "done"} - apply_web_agent_display_event(done_event, assistant_display_message) - # 终态先进入事件队列，避免展示快照落库延迟前端结束动画。 - event_publisher.publish(done_event) - if not is_secret_confirmation_control: - try: - await save_web_agent_display_snapshot( - session_id=session_id, - current_user=current_user, - messages=display_messages, - client_session_id=command.session_id or session_id, - service=service, - persistence=persistence, - ) - except Exception as err: - logger.error(f"保存WebAgent展示历史失败：{err}") - - task = create_web_agent_background_task(run_agent()) - disconnected = False - terminal_sent = False - try: - yield {"type": "start", "session_id": session_id} - while not runtime_stop_state.is_system_stopped: - if await is_disconnected(): - disconnected = True - break - try: - event = await asyncio.wait_for( - event_publisher.get(), - timeout=WEB_AGENT_STREAM_HEARTBEAT_SECONDS, - ) - except asyncio.TimeoutError: - yield {"type": "heartbeat"} - continue - if event.get("type") == "done": - terminal_sent = True - yield event - if event.get("type") == "done": - break - except asyncio.CancelledError: - disconnected = True - return - finally: - if not task.done() and not disconnected and not terminal_sent: - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - await event_publisher.aclose() - # 客户端断线后保留 Agent 继续执行；发布器关闭后拒绝受保护结果。 - - return event_generator() - - async def build_web_agent_stream( command: WebAgentStreamCommand, *, @@ -2159,6 +2007,12 @@ async def build_web_agent_stream( } ) ) + from app.application.messaging.webagentstream import ( + WebAgentStreamDependencies, + _build_steering_ack_stream, + build_agent_web_agent_stream, + submit_web_agent_steering, + ) transcript = await transcribe_web_agent_audio_input(command.audio_refs) prompt = merge_web_agent_prompt_with_transcript(prompt, transcript) @@ -2176,9 +2030,37 @@ async def build_web_agent_stream( } ) ) + # 活动 Agent 运行内的补充输入走同一张图；无活动运行时才创建新一轮。 + steering_message = await submit_web_agent_steering( + manager=manager, + session_id=session_id, + user_id=str(current_user.id), + prompt=prompt, + images=command.images, + files=command.files, + audio_refs=command.audio_refs, + ) + if steering_message is not None: + return WebAgentStreamResult( + events=_build_steering_ack_stream(session_id=session_id, message=steering_message), + control="steering", + ) + stream_dependencies = WebAgentStreamDependencies( + event_publisher_factory=WebAgentEventPublisher, + bind_user_session=bind_web_agent_user_session, + apply_display_event=apply_web_agent_display_event, + build_display_message=build_web_agent_display_message, + build_input_attachments=build_web_agent_input_attachments, + build_message_events_async=build_web_agent_message_events_async, + save_display_snapshot=save_web_agent_display_snapshot, + split_output=split_web_agent_output, + create_background_task=create_web_agent_background_task, + source=WEB_AGENT_SOURCE, + heartbeat_seconds=WEB_AGENT_STREAM_HEARTBEAT_SECONDS, + ) return WebAgentStreamResult( - events=_build_agent_web_agent_stream( + events=build_agent_web_agent_stream( command=command, current_user=current_user, session_id=session_id, @@ -2190,6 +2072,7 @@ async def build_web_agent_stream( service=service, persistence=persistence, is_disconnected=is_disconnected, + dependencies=stream_dependencies, ), control=("secret-confirmation" if is_secret_confirmation_control else None), ) diff --git a/app/application/messaging/webagentstream.py b/app/application/messaging/webagentstream.py new file mode 100644 index 000000000..c682ead7d --- /dev/null +++ b/app/application/messaging/webagentstream.py @@ -0,0 +1,326 @@ +"""WebAgent 主运行流和运行中补充消息的展示桥接。""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from functools import partial +from typing import Any, AsyncIterator, Awaitable, Callable, Optional, cast + +from app.agent.steering import SteeringMessage +from app.application import agent as agent_application +from app.runtime.log import logger +from app.runtime.stop import runtime_stop_state +from app.schemas.message import Message +from app.schemas.types import NotificationChannel, ReplyMode + + +@dataclass(frozen=True, slots=True) +class WebAgentStreamDependencies: + """注入 WebAgent 流所需的展示、事件和后台任务能力，避免模块相互导入。""" + + event_publisher_factory: Callable[[], Any] + bind_user_session: Callable[[str, str], None] + apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None] + build_display_message: Callable[..., dict[str, Any]] + build_input_attachments: Callable[..., list[dict[str, Any]]] + build_message_events_async: Callable[[Message], Awaitable[list[dict[str, Any]]]] + save_display_snapshot: Callable[..., Awaitable[None]] + split_output: Callable[[str], list[dict[str, Any]]] + create_background_task: Callable[[Awaitable[object]], asyncio.Task[object]] + source: str + heartbeat_seconds: float + + +async def submit_web_agent_steering( + *, + manager: Any, + session_id: str, + user_id: str, + prompt: str, + images: list[str], + files: list[dict[str, Any]], + audio_refs: list[str], +) -> Optional[SteeringMessage]: + """把活动 WebAgent 的补充输入提交到会话 inbox；无活动运行时返回 None。""" + steering_files = [file for file in files if str(file.get("ref") or "") not in set(audio_refs)] + steering_files.extend({"ref": audio_ref, "mime_type": "audio/*"} for audio_ref in audio_refs) + submit = getattr(manager, "submit_steering_message", None) + if not callable(submit): + return None + return cast( + Optional[SteeringMessage], + await submit( + session_id=session_id, + user_id=user_id, + message=prompt, + images=images, + files=steering_files or None, + ), + ) + + +def _build_steering_ack_stream( + *, + session_id: str, + message: SteeringMessage, +) -> AsyncIterator[dict[str, Any]]: + """为运行中补充消息返回短 ACK 流，不创建第二个助手展示气泡。""" + + async def event_generator() -> AsyncIterator[dict[str, Any]]: + """按统一 WebAgent SSE 协议报告已排队的消息 ID。""" + yield {"type": "start", "session_id": session_id} + yield { + "type": "steering", + "status": "queued", + "message_id": message.message_id, + "content": message.text, + } + yield {"type": "done"} + + return event_generator() + + +def _publish_web_agent_protected_output( + event_publisher: Any, + content: str, +) -> bool: + """发布不进入普通展示快照的敏感交互结果。""" + return bool(event_publisher.publish({"type": "interaction-protected", "content": content})) + + +def _build_web_agent_steering_callback( + *, + display_messages: list[dict[str, Any]], + event_publisher: Any, + build_display_message: Callable[..., dict[str, Any]], + build_input_attachments: Callable[..., list[dict[str, Any]]], +) -> Callable[[SteeringMessage, str], None]: + """构造运行中补充消息的应用回调，并同步展示快照与 SSE。""" + + def steering_status_callback(message: SteeringMessage, status: str) -> None: + """把补充消息应用状态投影到当前助手流，保持单一助手气泡。""" + if status != "applied": + return + display_message = build_display_message( + role="user", + content=message.text, + attachments=build_input_attachments( + images=list(message.images), + files=list(message.files), + audio_refs=[], + ), + ) + # 当前助手气泡必须继续保持最后一项，补充的用户消息插入其前面。 + display_messages.insert(max(0, len(display_messages) - 1), display_message) + event_publisher.publish( + { + "type": "steering", + "status": "applied", + "message_id": message.message_id, + "content": message.text, + "display_message": display_message, + } + ) + + return steering_status_callback + + +def _build_web_agent_message_callback( + *, + assistant_display_message: dict[str, Any], + event_publisher: Any, + build_message_events_async: Callable[[Message], Awaitable[list[dict[str, Any]]]], + apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None], +) -> Callable[[Message], Awaitable[None]]: + """构造 Agent 主动消息的 Web 展示回调。""" + + async def message_callback(message: Message) -> None: + """接收 Agent 工具主动发送的 Web 通知。""" + for item in await build_message_events_async(message): + apply_display_event(item, assistant_display_message) + event_publisher.publish(item) + + return message_callback + + +def _build_web_agent_output_callback( + *, + assistant_display_message: dict[str, Any], + event_publisher: Any, + split_output: Callable[[str], list[dict[str, Any]]], + apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None], +) -> Callable[[str], None]: + """构造 Agent 文本增量的 Web 展示回调。""" + + def output_callback(delta: str) -> None: + """接收 Agent 文本增量并投影为展示事件。""" + for item in split_output(delta): + apply_display_event(item, assistant_display_message) + event_publisher.publish(item) + + return output_callback + + +def build_agent_web_agent_stream( + *, + command: Any, + current_user: Any, + session_id: str, + prompt: str, + display_prompt: str, + has_audio_input: bool, + is_secret_confirmation_control: bool, + protected_transport_supported: bool, + service: Any, + persistence: Any, + is_disconnected: Callable[[], Awaitable[bool]], + dependencies: WebAgentStreamDependencies, +) -> AsyncIterator[dict[str, Any]]: + """构造标准 Agent 执行链路的 WebAgent 事件流。""" + dependencies.bind_user_session(str(current_user.id), session_id) + event_publisher = dependencies.event_publisher_factory() + user_attachments = dependencies.build_input_attachments( + images=command.images, + files=command.files, + audio_refs=command.audio_refs, + ) + display_messages = [] + if command.echo_user and not is_secret_confirmation_control: + user_display_message = dependencies.build_display_message( + role="user", + content=display_prompt or prompt, + attachments=user_attachments, + ) + if command.choice_selection: + user_display_message["choice_selection"] = command.choice_selection + display_messages.append(user_display_message) + assistant_display_message = dependencies.build_display_message( + role="assistant", + status="streaming", + ) + display_messages.append(assistant_display_message) + output_callback = _build_web_agent_output_callback( + assistant_display_message=assistant_display_message, + event_publisher=event_publisher, + split_output=dependencies.split_output, + apply_display_event=dependencies.apply_display_event, + ) + message_callback = _build_web_agent_message_callback( + assistant_display_message=assistant_display_message, + event_publisher=event_publisher, + build_message_events_async=dependencies.build_message_events_async, + apply_display_event=dependencies.apply_display_event, + ) + protected_output_callback = partial(_publish_web_agent_protected_output, event_publisher) + steering_status_callback = _build_web_agent_steering_callback( + display_messages=display_messages, + event_publisher=event_publisher, + build_display_message=dependencies.build_display_message, + build_input_attachments=dependencies.build_input_attachments, + ) + + async def event_generator() -> AsyncIterator[dict[str, Any]]: + """执行 Agent 并按断线与终态语义消费事件。""" + audio_ref_set = set(command.audio_refs) + files = [file for file in command.files if str(file.get("ref") or "") not in audio_ref_set] + files.extend({"ref": audio_ref, "mime_type": "audio/*"} for audio_ref in command.audio_refs) + + async def run_agent() -> None: + """后台执行 Agent，并在完成后持久化展示快照。""" + try: + runtime_manager = agent_application.get_running_agent_manager() + if runtime_manager is None: + raise RuntimeError("智能助手服务尚未就绪，请稍后重试。") + await runtime_manager.process_message( + session_id=session_id, + user_id=str(current_user.id), + message=prompt, + images=command.images, + files=files or None, + has_audio_input=has_audio_input, + channel=NotificationChannel.WebAgent.value, + source=dependencies.source, + username=current_user.name, + reply_mode=ReplyMode.CAPTURE_ONLY, + allow_message_tools=True, + output_callback=output_callback, + protected_output_callback=(protected_output_callback if protected_transport_supported else None), + message_callback=message_callback, + steering_status_callback=steering_status_callback, + agent_factory=agent_application.get_web_agent_type(), + wait_for_completion=True, + ) + except asyncio.CancelledError: + # 显式停止会话沿用正常终止语义；服务关闭由 manager 的稳定异常分支处理。 + pass + except Exception as err: + logger.error(f"Web智能助手执行失败: {str(err)}") + error_event = { + "type": "error", + "message": "智能助手执行失败，请稍后重试", + } + dependencies.apply_display_event(error_event, assistant_display_message) + event_publisher.publish(error_event) + finally: + done_event = {"type": "done"} + dependencies.apply_display_event(done_event, assistant_display_message) + # 终态先进入事件队列，避免展示快照落库延迟前端结束动画。 + event_publisher.publish(done_event) + if not is_secret_confirmation_control: + try: + await dependencies.save_display_snapshot( + session_id=session_id, + current_user=current_user, + messages=display_messages, + client_session_id=command.session_id or session_id, + service=service, + persistence=persistence, + ) + except Exception as err: + logger.error(f"保存WebAgent展示历史失败：{err}") + + task = dependencies.create_background_task(run_agent()) + disconnected = False + terminal_sent = False + try: + yield {"type": "start", "session_id": session_id} + while not runtime_stop_state.is_system_stopped: + if await is_disconnected(): + disconnected = True + break + try: + event = await asyncio.wait_for( + event_publisher.get(), + timeout=dependencies.heartbeat_seconds, + ) + except asyncio.TimeoutError: + yield {"type": "heartbeat"} + continue + if event.get("type") == "done": + terminal_sent = True + yield event + if event.get("type") == "done": + break + except asyncio.CancelledError: + disconnected = True + return + finally: + if not task.done() and not disconnected and not terminal_sent: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + await event_publisher.aclose() + # 客户端断线后保留 Agent 继续执行；发布器关闭后拒绝受保护结果。 + + return event_generator() + + +__all__ = [ + "WebAgentStreamDependencies", + "_build_steering_ack_stream", + "build_agent_web_agent_stream", + "submit_web_agent_steering", +] diff --git a/docs/agent-evaluation.md b/docs/agent-evaluation.md index a1df2b087..7267ec3c0 100644 --- a/docs/agent-evaluation.md +++ b/docs/agent-evaluation.md @@ -39,6 +39,14 @@ uv run --locked --no-sync python -m scripts.evaluation --live \ --output report.json ``` +### 2026-09-11 供应商实测 + +本轮使用 Agnes AI Hub 的 `agnes-2.5-pro`，推理档位为供应商支持的最高档 `xhigh`，通过 Responses provider 配置和环境变量传递凭据。`max` 请求被供应商以 400 拒绝（支持档位为 `low`、`medium`、`xhigh`），因此不能把失败的 `max` 请求算作模型能力证据。 + +在同一提交的隔离评测中，`dedup_existing` 通过（8 次模型调用，约 50 秒）；`unknown_download` 和 `honest_unknown` 均有真实模型响应，但最终报告未通过独立 JSON 验收（各达到 12 次调用上限）。这组结果说明当前实现已经能进行真实供应商评测，也明确暴露了复杂场景的收敛问题，不能据此宣称已达到 Codex 整体水平。 + +原生探针在修正动态 `tool_search` 目录兼容性后可以启动，但原生 Codex 自带模型目录没有 Agnes 模型，无法在同一模型上形成有效的 MoviePilot/Codex 配对运行；探针没有调用真实模型。后续比较必须先取得双方都能调用的同一模型和推理档位，或把结果明确标为不同基线。 + 只有显式 `--live` 或 `--native` 才会调用真实模型并产生费用。默认读取 `~/.codex/config.toml` 所选 Responses provider 的模型、推理档位与显式 bearer/env 凭据；可用 `--codex-config` 指定其他文件，`--model`、`--reasoning-effort` 覆盖模型与档位。不会借用其他服务的登录凭据，也不会自动降低被供应商拒绝的参数。报告同时保留请求模型与供应商返回的模型标识；本地 Codex 配置不能证明运行中的 MoviePilot 使用相同配置。 调用配置经私有标准输入传给 worker，凭据不进入命令行、提示词或报告。worker 只继承必要的平台环境，先创建临时 `CONFIG_DIR`，再导入后端；每轮拥有独立回执库、记忆、会话和工具实例。生产 `process/_create_agent`、Skills、计划、权限、持久回执、工具输出预算、压缩和子代理仍按真实路径执行。主工具目录限定为假业务 API 和生产内部工具，API transport 拒绝任何外部目标及场景外 operation；插件、外部 MCP、通知和任意 shell/文件/浏览器工具不开放。此受控目录是当前评测边界，不代表默认部署工具全集。 diff --git a/docs/agent.md b/docs/agent.md index f13a47f85..eae126900 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -1,6 +1,6 @@ # Agent 复杂任务执行与恢复 -MoviePilot Agent 通过模型、Skills、工具和会话状态共同完成任务。本次增强关注长任务的连续性：明确目标和检查结果，按需补充工具，并在执行中断后保留已知事实。能力对齐路线与每轮真实运行验收见 [Agent 与 Codex Harness 对齐路线](agent-codex-parity-plan.md)。 +MoviePilot Agent 通过模型、Skills、工具和会话状态共同完成任务。本次增强关注长任务的连续性：明确目标和检查结果，按需补充工具，并在执行中断后保留已知事实。能力对齐路线与每轮真实运行验收见 [Agent 与 Codex Harness 对齐路线](refactor/agent-codex-parity-plan.md)。 ## 任务计划 @@ -12,6 +12,10 @@ MoviePilot Agent 通过模型、Skills、工具和会话状态共同完成任务 - 计划快照随聊天消息保存，在会话图重建后恢复。用户询问进度时继续当前目标，明确开始新任务时替换计划。 - 计划是模型记录的工作状态，不代表工具执行结果或写操作授权。仍需通过工具核实实际结果。 +## 运行中补充消息 + +WebAgent 在 Agent 正在运行时仍可提交文本和附件。宿主为每个会话生成消息 ID，先把消息放入有界 steering inbox；下一次模型调用前由中间件把它作为真实 `HumanMessage` 写入图状态，并通过当前 SSE 报告 `queued`、`applied`。提交和运行收尾共享原子边界，收尾竞态中已接受的消息会在同一 worker 内继续处理，停止会话不会继续派发新的工具动作。补充消息不会启动第二张 Agent 图，也不会替换当前输出回调。 + ## 按需发现工具 设置 `LLM_MAX_TOOLS > 0` 时，Agent 首轮仍只筛选一批相关工具。后续读取 Skill 或发现新问题后，可以通过内部 `search_tools(search, limit)` 按名称、说明或标签搜索当前会话工具目录，并在下一次模型调用获得匹配工具的完整参数定义。 diff --git a/docs/agent-codex-parity-plan.md b/docs/refactor/agent-codex-parity-plan.md similarity index 78% rename from docs/agent-codex-parity-plan.md rename to docs/refactor/agent-codex-parity-plan.md index 793287b0b..158928536 100644 --- a/docs/agent-codex-parity-plan.md +++ b/docs/refactor/agent-codex-parity-plan.md @@ -1,5 +1,7 @@ # MoviePilot Agent 与 Codex Harness 对齐路线 +> 归档位置：`docs/refactor/`。 + 更新时间：2026-09-11 这份文档是 Agent 能力对齐的交付路线和每轮验收合同。它记录当前证据与未完成目标，不把一次成功的模型调用当成“已经和 Codex 一样聪明”。最终判断必须同时看模型行为、工具执行、任务生命周期、业务终态和失败后的收敛结果。 @@ -8,15 +10,15 @@ 让 MoviePilot Agent 在 MoviePilot 的真实业务边界内具备可复核的 Codex 级 Harness 能力：模型能看到准确的工具合同，工具能完成完整的命令行、浏览器、终端输入输出、子代理和长任务协作闭环，宿主能隔离任务身份、收口进程并在新消息到达时继续推进。高影响业务动作仍由宿主的授权和确认策略控制。 -真实模型评测的固定档位使用 **luna-max**：模型标识为 `gpt-5.6-luna`，推理预算为 `max`。报告必须同时记录这两个字段；更换模型、推理预算、提示集或判定器后，结果属于新的基线，不能与旧报告直接合并。 +真实模型评测必须使用显式配置的供应商模型和推理档位。本轮 Agnes AI Hub 基线为 `agnes-2.5-pro + xhigh`；该供应商只接受 `low`、`medium`、`xhigh`，因此不能继续使用 `luna-max` 的 `max` 档位。不同模型、供应商或推理档位的结果属于不同基线，不能直接合并。 ## 交付顺序 | 目标 | 状态 | 验收重点 | | --- | --- | --- | | C1.4 终端工具完整闭环 | 已完成基线 | `run/pipe/PTY` 共用 shell、cwd、login 和 UTF-8 策略；stdin 写入、EOF、分页、interrupt/kill、超时和进程组收尾有真实进程测试 | -| C1.3 终端任务作用域 | 本轮进行中 | 终端归属由宿主对象身份决定；定时运行、会话、子任务和内部工具管理器隔离；封口先于清理，排队或运行中的命令都不会在任务结束后迟到启动 | -| S2.3 运行中消息排队与 WebAgent 输入 | 下一目标 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 能报告 queued/applied，停止后不再派发后续工具 | +| C1.3 终端任务作用域 | 已完成 | 终端归属由宿主对象身份决定；定时运行、会话、子任务和内部工具管理器隔离；封口先于清理，排队或运行中的命令都不会在任务结束后迟到启动 | +| S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 报告 queued/applied，停止后不再派发后续工具；剩余边界由真实长任务回归继续覆盖 | | 浏览器能力对齐 | S2.3 后续 | 导航、页面读取、点击/输入、等待、截图和失败收口使用真实浏览器状态；工具清单、权限、超时、重试和会话生命周期与命令行能力同样可观测 | | S3.1 通用子代理 | S2.3 后 | 主 Agent 按任务动态派发通用子代理；保留专用画像必须有独立收益证据。子代理的授权、工具角色、终端分享和副作用边界不能因画像切换而放宽 | @@ -25,7 +27,7 @@ 每个影响 Agent、工具、会话、浏览器、子代理或提示词的代码轮次都要执行下面四层检查，并把结果与最终提交 SHA 绑定： 1. **确定性回归**：运行受影响的 pytest、全量 Agent 测试、类型/格式/架构 ratchet 和必要的全量测试。任何已知基线失败都要在未改动基线复现后再归因。 -2. **真实 MoviePilot 运行**：使用 `gpt-5.6-luna` + `max`，在隔离 worker 中跑固定场景，保留真实模型请求、工具轨迹、耗时、token、终态和进程收尾证据。业务副作用只能进入评测假世界，不能把测试凭据或私有响应写入报告。 +2. **真实 MoviePilot 运行**：使用评测配置中显式声明的供应商模型和支持档位（本轮为 `agnes-2.5-pro` + `xhigh`），在隔离 worker 中跑固定场景，保留真实模型请求、工具轨迹、耗时、token、终态和进程收尾证据。业务副作用只能进入评测假世界，不能把测试凭据或私有响应写入报告。 3. **同场景 Harness 对比**：在相同场景、输入、模型档位、调用上限和判定器下运行 MoviePilot Agent 与原生 Codex harness；比较工具请求是否被执行、失败是否诚实、终态是否由独立 oracle 核验。原生 harness 不可用时报告 `blocked`，不能以离线脚本或单元测试替代。 4. **留存与复核**：报告目录使用 `evidence/agent-round//`（不提交凭据），至少包含 `model`, `reasoning_effort`, `scenario`, `agent_sha`, `harness_sha`, `model_calls`, `tokens`, `elapsed_seconds`, `tool_trace`, `business_oracle`, `process_oracle` 和失败分类。重复运行同一轮时保留每次报告，不能覆盖异常样本。 @@ -35,7 +37,7 @@ UV_PROJECT_ENVIRONMENT=/Users/jxxghp/MPProjects/MoviePilot/.venv \ uv run --locked --no-sync python -m scripts.evaluation \ --live --scenario unknown_download \ - --model gpt-5.6-luna --reasoning-effort max \ + --model agnes-2.5-pro --reasoning-effort xhigh \ --output evidence/agent-round//moviepilot-unknown_download.json ``` diff --git a/docs/music-lifecycle-audit.md b/docs/refactor/music-lifecycle-audit.md similarity index 99% rename from docs/music-lifecycle-audit.md rename to docs/refactor/music-lifecycle-audit.md index 8b1cc9e60..60d102f3d 100644 --- a/docs/music-lifecycle-audit.md +++ b/docs/refactor/music-lifecycle-audit.md @@ -1,5 +1,7 @@ # 音乐全流程审计（2026-09-09） +> 归档位置：`docs/refactor/`。 + ## 结论 普通分轨专辑和独立单曲已接入通用搜索、订阅、下载、整理、刮削及媒体库刷新流程。 diff --git a/scripts/evaluation/proxy.py b/scripts/evaluation/proxy.py index fc933c73d..25119cd3b 100644 --- a/scripts/evaluation/proxy.py +++ b/scripts/evaluation/proxy.py @@ -79,7 +79,9 @@ def project_tools(tools: Any, prefix: str = "") -> tuple[list[dict[str, Any]], l elif tool["type"] == "tool_search": description = tool.get("description") sources = re.findall(r"(?m)^- ([^:\n]+):", description) if isinstance(description, str) else [] - if prefix or tool.get("execution") != "client" or sources != ["evaluation"]: + allowed_sources = {"Multi-agent tools", "evaluation"} + if (prefix or tool.get("execution") != "client" or not sources + or "evaluation" not in sources or any(source not in allowed_sources for source in sources)): raise ValueError("原生工具搜索不是固定评测目录") projected.append(copy.deepcopy(tool)) retained.append(qualified) diff --git a/tests/fixtures/architecture/concurrency-baseline.json b/tests/fixtures/architecture/concurrency-baseline.json index 34b95d527..fdc104aaf 100644 --- a/tests/fixtures/architecture/concurrency-baseline.json +++ b/tests/fixtures/architecture/concurrency-baseline.json @@ -175,6 +175,12 @@ }, "target": "asyncio.to_thread" }, + "app/agent/terminal/ownership.py:asyncio.to_thread": { + "owners": { + "close_terminal_scope": 1 + }, + "target": "asyncio.to_thread" + }, "app/agent/tools/base.py:concurrent.futures.ThreadPoolExecutor": { "owners": { "_submit_blocking_call": 1 diff --git a/tests/test_agent_evaluation_proxy.py b/tests/test_agent_evaluation_proxy.py index 2c9a387ce..6284ed9dd 100644 --- a/tests/test_agent_evaluation_proxy.py +++ b/tests/test_agent_evaluation_proxy.py @@ -426,6 +426,27 @@ async def test_native_additional_tools_are_projected_in_place_and_audited() -> N assert record["tool_catalogs"][0]["location"] == "input[0].tools" +@pytest.mark.asyncio +async def test_native_tool_search_accepts_the_bundled_multi_agent_source() -> None: + """当前原生 CLI 会同时声明协作工具与评测源，已知两者都应保留并继续隔离。""" + outbound = [] + + def handler(request: httpx.Request) -> httpx.Response: + """接收投影后的目录，不访问真实模型服务。""" + outbound.append(json.loads(request.content)) + return httpx.Response(200, json=_completed()) + + payload = _native_payload() + payload["input"][0]["tools"][-1] = _search_tool( + "Tools from the following sources:\n- Multi-agent tools: Bundled collaboration tools\n- evaluation: Controlled world" + ) + proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(handler)) + async with _client(proxy) as client: + assert (await client.post("/v1/responses", json=payload)).status_code == 200 + assert "tool_search" in proxy.snapshot()["model_requests"][0]["retained_tools"] + assert outbound[0]["input"][0]["tools"][-1]["execution"] == "client" + + @pytest.mark.parametrize("catalog", [ [{"type": "namespace", "name": "unknown", "tools": [{"type": "function", "name": "update_plan"}]}], [_search_tool('Tools from sources:\n- evaluation: fake\n- private: external')], diff --git a/tests/test_agent_steering.py b/tests/test_agent_steering.py new file mode 100644 index 000000000..d90474c81 --- /dev/null +++ b/tests/test_agent_steering.py @@ -0,0 +1,168 @@ +"""运行中 Agent 消息排队、真实 HumanMessage 注入及身份隔离测试。""" + +import asyncio + +import pytest +from langchain.agents import create_agent +from langchain.agents.middleware.types import ModelRequest, ModelResponse +from langchain_core.language_models.fake_chat_models import GenericFakeChatModel +from langchain_core.messages import AIMessage, HumanMessage +from langgraph.checkpoint.memory import InMemorySaver + +from app.agent.middleware.steering import SteeringMiddleware +from app.agent.session import AgentSessionOwner +from app.agent.steering import SteeringInbox, bind_steering_inbox, reset_steering_inbox + + +@pytest.mark.asyncio +async def test_steering_message_is_injected_as_real_human_message() -> None: + """下一次模型调用应在图状态中看到真实 HumanMessage，而不是提示词拼接。""" + inbox = SteeringInbox("session", "user") + await inbox.begin_run() + queued = await inbox.enqueue(user_id="user", text="补充要求") + assert queued is not None + + graph = create_agent( + model=GenericFakeChatModel(messages=iter([AIMessage(content="完成")])), + tools=[], + middleware=[SteeringMiddleware()], + checkpointer=InMemorySaver(), + ) + token = bind_steering_inbox(inbox) + try: + result = await graph.ainvoke( + {"messages": [HumanMessage(content="开始任务")]}, + config={"configurable": {"thread_id": "session"}}, + ) + finally: + reset_steering_inbox(token) + + messages = result["messages"] + assert [message.type for message in messages] == ["human", "human", "ai"] + assert "补充要求" in messages[1].content[0]["text"] + assert messages[1].additional_kwargs["moviepilot_steering_message_id"] == queued.message_id + + +@pytest.mark.asyncio +async def test_steering_does_not_break_unresolved_tool_call_pairing() -> None: + """未交付工具调用所在的状态边界不应插入用户消息。""" + inbox = SteeringInbox("session", "user") + await inbox.begin_run() + assert await inbox.enqueue(user_id="user", text="稍等") is not None + middleware = SteeringMiddleware() + token = bind_steering_inbox(inbox) + seen_messages = [] + + async def handler(request): + """记录未注入的模型请求，模拟下游模型处理。""" + seen_messages.extend(request.messages) + return ModelResponse(result=[AIMessage(content="继续执行工具")]) + + try: + pending_tool_message = AIMessage( + content="", + tool_calls=[{"name": "tool", "args": {}, "id": "call"}], + ) + await middleware.awrap_model_call( + request=ModelRequest( + model=GenericFakeChatModel(messages=iter([AIMessage(content="unused")])), + tools=[], + system_message=None, + response_format=None, + messages=[pending_tool_message], + tool_choice=None, + state={"messages": [pending_tool_message]}, + runtime=None, + ), + handler=handler, + ) + finally: + reset_steering_inbox(token) + assert seen_messages[-1].tool_calls + assert inbox.pending_count == 1 + + +@pytest.mark.asyncio +async def test_owner_routes_matching_user_message_to_active_steering_inbox() -> None: + """活动会话只接受同一用户的补充消息，其他用户回退到普通入口。""" + started = asyncio.Event() + release = asyncio.Event() + processed_messages: list[str] = [] + + class BlockingAgent: + """模拟持久会话 Agent，暴露运行边界供 owner 测试。""" + + def __init__(self, **kwargs): + """保存宿主注入字段。""" + self.__dict__.update(kwargs) + self.user_id = str(kwargs["user_id"]) + + def set_output_callback(self, callback): + """兼容会话 owner 的输出回调更新。""" + self.output_callback = callback + + def set_protected_output_callback(self, callback): + """兼容会话 owner 的敏感输出回调更新。""" + self.protected_output_callback = callback + + def get_session_status(self): + """返回 owner 状态合并所需的最小快照。""" + return {"model": "fake", "is_processing": True} + + async def process(self, message, **kwargs): + """阻塞到测试释放，模拟长时间模型运行。""" + processed_messages.append(message) + started.set() + await release.wait() + return message + + async def cleanup(self): + """报告测试 Agent 已收敛。""" + return True + + owner = AgentSessionOwner() + owner._accepting_tasks = True + running = asyncio.create_task( + owner.process_message( + session_id="session", + user_id="user", + message="开始", + agent_factory=BlockingAgent, + wait_for_completion=True, + ) + ) + await asyncio.wait_for(started.wait(), 1) + accepted = await owner.submit_steering_message( + session_id="session", + user_id="user", + message="继续检查", + ) + rejected = await owner.submit_steering_message( + session_id="session", + user_id="other", + message="越权", + ) + assert accepted is not None + assert rejected is None + assert owner.get_session_status("session")["steering_pending"] == 1 + + release.set() + assert await asyncio.wait_for(running, 1) == "继续检查" + assert processed_messages == ["开始", "继续检查"] + worker = owner._session_workers.get("session") + if worker: + worker.cancel() + await asyncio.gather(worker, return_exceptions=True) + + +@pytest.mark.asyncio +async def test_steering_finish_boundary_returns_pending_messages_once() -> None: + """运行收尾与入队共享锁时，未注入消息只能被交给下一轮一次。""" + inbox = SteeringInbox("session", "user") + await inbox.begin_run() + message = await inbox.enqueue(user_id="user", text="收尾前到达") + assert message is not None + pending = await inbox.finish_run() + assert pending == (message,) + assert await inbox.finish_run() == () + assert await inbox.enqueue(user_id="user", text="已关闭运行") is None diff --git a/tests/test_browser_helper.py b/tests/test_browser_helper.py index b1d67b783..53ea873a2 100644 --- a/tests/test_browser_helper.py +++ b/tests/test_browser_helper.py @@ -17,6 +17,7 @@ from app.adapters.network.browser import ( launch_browser_context, launch_browser_context_async, ) +from app.agent.terminal.ownership import TerminalScope, close_terminal_scope from app.agent.tools.impl.browse_webpage import BrowserAction, BrowseWebpageTool from app.runtime.correlation import correlation_scope, get_correlation_id @@ -417,6 +418,54 @@ def test_browser_session_helper_closes_session_on_worker_thread(): assert context.close_thread_id == session_thread_id +def test_browser_session_helper_isolates_owner_and_closes_owner_sessions(): + """相同模型 session_key 在不同宿主作用域下不能串用，owner 收口会关闭上下文。""" + first_page = _FakePage("first") + second_page = _FakePage("second") + contexts = [_FakeContext([first_page]), _FakeContext([second_page])] + helper = BrowserSessionHelper() + first_owner = TerminalScope("alice", "task-one", "conversation") + second_owner = TerminalScope("alice", "task-two", "conversation") + + with patch.object(BrowserSessionHelper, "_launch_context", side_effect=contexts): + assert helper.with_session("shared", lambda session: session.owner, owner=first_owner) is first_owner + with pytest.raises(PermissionError): + helper.with_session("shared", lambda _session: None, owner=second_owner) + assert BrowserSessionHelper.close_owner(first_owner) + + assert first_page.closed and first_page.page_id == "first" + assert contexts[0].closed + with patch.object(BrowserSessionHelper, "_launch_context", return_value=contexts[1]): + assert helper.with_session("shared", lambda session: session.owner, owner=second_owner) is second_owner + assert BrowserSessionHelper.close_owner(second_owner) + assert second_page.closed and contexts[1].closed + + +def test_browser_session_helper_rejects_closed_owner_before_context_creation(): + """作用域封口后不得重新创建或复用浏览器上下文。""" + owner = TerminalScope("alice", "closed-browser-task", "conversation") + owner.seal() + with patch.object(BrowserSessionHelper, "_launch_context") as launch_context: + with pytest.raises(PermissionError, match="所属任务已停止"): + BrowserSessionHelper().with_session( + "closed", lambda _session: None, owner=owner + ) + launch_context.assert_not_called() + + +@pytest.mark.asyncio +async def test_close_terminal_scope_closes_browser_owner_without_blocking_event_loop(): + """通用作用域收口应回收其浏览器上下文，且通过线程避免阻塞事件循环。""" + page = _FakePage("owned") + context = _FakeContext([page]) + helper = BrowserSessionHelper() + owner = TerminalScope("alice", "browser-task", "conversation") + with patch.object(BrowserSessionHelper, "_launch_context", return_value=context): + helper.with_session("owned", lambda session: session.active_page, owner=owner) + assert await close_terminal_scope(owner) + assert page.closed and context.closed + + def test_browse_webpage_returns_snapshot_with_refs_after_goto(): """goto 后应返回包含可交互元素 ref 的页面快照。""" page = _FakePage() diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 8fd85cd33..3a19f3a35 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -12,6 +12,7 @@ from app.agent.contracts import ReplyMode # pylint: disable=no-name-in-module # 旧公开入口由 runtime compat 惰性注入，Pylint 无法静态解析。 from app.agent.orchestrator import agent_manager +from app.agent.steering import SteeringMessage from app.agent.web import _get_web_agent_type from app.api.endpoints import agent as agent_endpoint from app.api.endpoints.agent import web_agent_stream @@ -79,6 +80,7 @@ from app.application.messaging.agent import ( ) from app.application.messaging.chat import AgentChatService, configure_agent_chat_service from app.application.messaging.skill import skill_interaction_manager +from app.application.messaging.webagentstream import _build_steering_ack_stream from app.chain.message import MessageChain from app.db.oper.agentchat import AgentChatOper from app.runtime.events import Event @@ -138,6 +140,24 @@ def test_web_agent_non_transport_helpers_are_application_owned(): assert all(not hasattr(agent_endpoint, name) for name in ownership) +@pytest.mark.asyncio +async def test_steering_ack_stream_reports_a_queued_message_without_an_assistant_bubble(): + """运行中补充消息只返回短确认流，主 Agent 的展示气泡由原流继续持有。""" + message = SteeringMessage.create(session_id="session", user_id="user", text="补充要求") + events = [event async for event in _build_steering_ack_stream(session_id="session", message=message)] + + assert events == [ + {"type": "start", "session_id": "session"}, + { + "type": "steering", + "status": "queued", + "message_id": message.message_id, + "content": "补充要求", + }, + {"type": "done"}, + ] + + def test_split_web_agent_output_extracts_verbose_tool_message(): """应将啰嗦模式工具提示拆成独立工具事件，并保留渠道展示文案。""" events = _split_web_agent_output("准备查询。\n\n⚙️ => 查询站点\n\n已完成")