From 7584d4b9fcca7effdd8c1954eaab79e71c7ef56a Mon Sep 17 00:00:00 2001 From: Aqr-K <95741669+Aqr-K@users.noreply.github.com> Date: Fri, 23 Oct 2026 08:20:59 +0800 Subject: [PATCH] fix: isolate agent sessions and retain runtime resource ownership (#7149) * fix: arbitrate subscription identity and restore workflow state * fix: isolate agent sessions and retain runtime resource ownership * test: preserve legacy empty-owner subscription transactions * test: handle reaped terminal descendants during shutdown polling * fix: align shared atomic configuration method with lifecycle prerequisite * test: cover subagent cancellation and completion lifecycle boundaries --------- Co-authored-by: jxxghp --- app/agent/history/storage.py | 13 +- app/agent/llm/discovery.py | 49 +- app/agent/mcp.py | 19 +- app/agent/middleware/subagents.py | 13 +- app/agent/runtime.py | 9 +- app/agent/skills/registry.py | 5 +- app/agent/terminal/manager.py | 40 +- app/agent/tools/impl/_file_write_utils.py | 48 +- app/agent/tools/impl/apply_patch.py | 6 +- app/agent/tools/impl/edit_file.py | 5 +- app/agent/tools/impl/view_image.py | 2 +- app/agent/tools/impl/write_file.py | 6 +- app/agent/tools/manager.py | 2 + app/application/messaging/agent.py | 7 +- app/application/messaging/webagent/stream.py | 1 + app/chain/message.py | 48 +- app/startup/initializers/agent.py | 21 +- docs/architecture-overview.md | 2 +- docs/mcp-api.md | 4 + scripts/generate_agent_skill_docs.py | 2 +- skills/moviepilot-api/SKILL.md | 10 +- .../architecture/concurrency-baseline.json | 6 + .../architecture/dependency-baseline.json | 15 +- tests/test_agent_apply_patch.py | 7 +- tests/test_agent_data_ports.py | 4 + tests/test_agent_interaction.py | 3 +- .../test_agent_runtime_review_regressions.py | 786 ++++++++++++++++++ tests/test_agent_session_status.py | 4 +- tests/test_agent_subagent_review_lifecycle.py | 219 +++++ tests/test_agent_terminal_ownership.py | 68 ++ tests/test_builtin_skill_boundaries.py | 2 +- tests/test_llm_provider_transport.py | 3 +- tests/test_web_agent_stream.py | 4 +- 33 files changed, 1345 insertions(+), 88 deletions(-) create mode 100644 tests/test_agent_runtime_review_regressions.py create mode 100644 tests/test_agent_subagent_review_lifecycle.py diff --git a/app/agent/history/storage.py b/app/agent/history/storage.py index 18693b908..af4ba4a82 100644 --- a/app/agent/history/storage.py +++ b/app/agent/history/storage.py @@ -249,11 +249,18 @@ class SqliteRecallRepository: """ 最近会话先走活动时间索引限量取候选，不扫描全库会话链。 """ - rows = connection.execute("SELECT * FROM sessions WHERE source NOT IN ('kanban','subagent','tool') ORDER BY last_active DESC LIMIT ?", - (query.limit + 15,)) + where = ["source NOT IN ('kanban','subagent','tool')", "id<>?"] + parameters: list[Any] = [query.current_session_id] + for operator, bound in ((">=", query.after), ("<", query.before)): + if bound is not None: + where.append(f"started_at {operator} ?") + parameters.append(bound) + excluded = {lineage(connection, identity) for identity in query.exclude_session_ids} + order = "started_at ASC,id ASC" if query.sort == "oldest" else "last_active DESC,id DESC" + rows = connection.execute(f"SELECT * FROM sessions WHERE {' AND '.join(where)} ORDER BY {order}", parameters) results: list[dict[str, Any]] = [] for row in rows: - if row['id'] == query.current_session_id: + if row['id'] == query.current_session_id or lineage(connection, row['id']) in excluded: continue first, _ = bookends(connection, row['id'], 1, 0) results.append(dict(session_id=row['id'], title=row['title'], source=row['source'], diff --git a/app/agent/llm/discovery.py b/app/agent/llm/discovery.py index f829a9dda..272c0d984 100644 --- a/app/agent/llm/discovery.py +++ b/app/agent/llm/discovery.py @@ -436,27 +436,38 @@ class _ProviderDiscovery: try: response = await client.aio.models.list() results = [] - for model in response.page: - supported = set(model.supported_actions or []) - if "generateContent" not in supported: - continue - model_id = model.name - metadata = ( - await self._models_dev_model( - "google", - model_id, - use_proxy=use_proxy, + tokens: set[str] = set() + for _page in range(100): + for model in response.page: + supported = set(model.supported_actions or []) + if "generateContent" not in supported: + continue + model_id = model.name + metadata = ( + await self._models_dev_model( + "google", + model_id, + use_proxy=use_proxy, + ) + or {} ) - or {} - ) - results.append( - self._normalize_model_record( - model_id=model_id, - display_name=model.display_name or metadata.get("name") or model_id, - metadata=metadata, - source="provider", + results.append( + self._normalize_model_record( + model_id=model_id, + display_name=model.display_name or metadata.get("name") or model_id, + metadata=metadata, + source="provider", + ) ) - ) + token = response.config.get("page_token") + if not token: + break + if token in tokens: + raise LLMProviderError("Google 模型列表分页游标重复") + tokens.add(token) + await response.next_page() + else: + raise LLMProviderError("Google 模型列表分页超过上限") return sorted(results, key=lambda item: item["name"].lower()) finally: try: diff --git a/app/agent/mcp.py b/app/agent/mcp.py index ff0f511a8..33f24f337 100644 --- a/app/agent/mcp.py +++ b/app/agent/mcp.py @@ -669,8 +669,23 @@ class AgentMcpManager: session_manager = await _open_mcp_session(normalized_server) async with session_manager as session: await self.initialize_session(session) - result = await session.request("tools/list") - tools_payload = result.get("tools", []) if isinstance(result, dict) else [] + tools_payload = [] + cursors: set[str] = set() + params = None + for _page in range(100): + result = await session.request("tools/list", params) if params else await session.request("tools/list") + if not isinstance(result, dict) or not isinstance(result.get("tools", []), list): + raise ValueError("MCP 工具列表响应格式无效") + tools_payload.extend(result.get("tools", [])) + cursor = result.get("nextCursor") + if cursor is None or cursor == "": + break + if not isinstance(cursor, str) or cursor in cursors: + raise ValueError("MCP 工具列表分页游标无效或重复") + cursors.add(cursor) + params = {"cursor": cursor} + else: + raise ValueError("MCP 工具列表分页超过上限") tool_specs: list[AgentMcpToolSpec] = [] for item in tools_payload: if not isinstance(item, dict) or not item.get("name"): diff --git a/app/agent/middleware/subagents.py b/app/agent/middleware/subagents.py index 4e5af470c..22483873a 100644 --- a/app/agent/middleware/subagents.py +++ b/app/agent/middleware/subagents.py @@ -976,7 +976,7 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): if cancel_requested is not None: cancel_requested.discard(task) record = self._tasks.get(task_id) - if record: + if record and record.task is task: record.finished_at = datetime.now() if task.cancelled(): return @@ -1005,12 +1005,7 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): name=task_id, ) record.task = task - task.add_done_callback( - lambda finished_task: self._mark_task_finished( - task_id, - finished_task, - ) - ) + task.add_done_callback(self._task_finished_callback(task_id)) self._tasks[task_id] = record records.append(record) logger.info( @@ -1109,10 +1104,14 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): """ 有界取消旧任务后复用 task_id 启动更新后的子代理任务。 """ + if not self._accepting_tasks or self._tasks.get(record.task_id) is not record: + return False, "子代理任务控制器正在关闭或任务已被回收，更新被拒绝。" if not record.task.done(): pending = await self._cancel_records([record]) if pending: return False, "旧子代理任务未能在取消窗口内收敛，更新被拒绝。" + if not self._accepting_tasks or self._tasks.get(record.task_id) is not record: + return False, "子代理任务控制器正在关闭或任务已被回收，更新被拒绝。" record.description = description.strip() record.terminal_sessions = terminal_sessions or [] record.started_at = None diff --git a/app/agent/runtime.py b/app/agent/runtime.py index 9f2ed8cf7..d752fd05d 100644 --- a/app/agent/runtime.py +++ b/app/agent/runtime.py @@ -334,6 +334,8 @@ class AgentRuntimeManager: self.bundled_defaults_dir = bundled_defaults_dir or (Path(__file__).parent / "defaults") self._cache_lock = threading.Lock() self._cached_signature: Optional[tuple[tuple[str, int, int], ...]] = None + # 磁盘观测签名与已加载配置版本分开，检查新文件不能提前使旧配置命中。 + self._observed_signature: Optional[tuple[tuple[str, int, int], ...]] = None self._cached_config: Optional[AgentRuntimeConfig] = None self._cached_signature_checked_at = 0.0 self._signature_check_interval = 1.0 @@ -388,6 +390,7 @@ class AgentRuntimeManager: """ with self._cache_lock: self._cached_signature = None + self._observed_signature = None self._cached_config = None self._cached_signature_checked_at = 0.0 self._layout_ready = False @@ -418,14 +421,14 @@ class AgentRuntimeManager: now = time.monotonic() with self._cache_lock: if ( - self._cached_signature is not None + self._observed_signature is not None and now - self._cached_signature_checked_at < self._signature_check_interval ): - return self._cached_signature + return self._observed_signature signature = self._build_signature() with self._cache_lock: - self._cached_signature = signature + self._observed_signature = signature self._cached_signature_checked_at = now return signature diff --git a/app/agent/skills/registry.py b/app/agent/skills/registry.py index 07f122f75..c0ebf5052 100644 --- a/app/agent/skills/registry.py +++ b/app/agent/skills/registry.py @@ -669,7 +669,7 @@ class SkillHelper(metaclass=WeakSingleton): if not skill.repo_url or not skill.skill_path: return False, "技能来源信息不完整，无法安装" - repo = self._parse_market_repo(skill.repo_url) + repo = self._parse_market_repo(skill.path) if "/tree/" in skill.path else self._parse_market_repo(skill.repo_url) if not repo: return False, "技能市场地址无效" @@ -1094,8 +1094,7 @@ class SkillHelper(metaclass=WeakSingleton): """ 下载市场仓库压缩包，并在缺省分支之间做回退尝试。 """ - branches = [repo.get("branch")] if repo.get("branch") else [] - branches.extend([branch for branch in _DEFAULT_BRANCHES if branch not in branches]) + branches = [repo["branch"]] if repo.get("branch") else list(_DEFAULT_BRANCHES) for branch in branches: archive_url = ( f"https://codeload.github.com/{repo['owner']}/{repo['repo']}/zip/refs/heads/{branch}" diff --git a/app/agent/terminal/manager.py b/app/agent/terminal/manager.py index d73ea070d..20bea5e44 100644 --- a/app/agent/terminal/manager.py +++ b/app/agent/terminal/manager.py @@ -707,21 +707,39 @@ class _TerminalSessionManager: 同一终端的关闭串行执行，真实收尾前不丢弃进程或输出读取器。 """ async with session.termination_lock: - if session.status == "running": + if session.status == "running" or os.name == "posix": session.kill_requested = True self._send_signal(session, signal.SIGTERM) if not await self._wait_for_exit(session): - if session.status == "running": + if session.status == "running" or os.name == "posix": self._send_signal(session, _KILL_SIGNAL) if not await self._wait_for_exit(session): logger.error(f"终端会话关闭超时: session_id={session.session_id}, pid={session.pid}") return False + # 组长先结束也不代表组内后代已退出；关闭该owner时完成整组最终收口。 + if os.name == "posix" and not await self._kill_remaining_process_group(session): + return False if any(not task.done() for task in session.reader_tasks): return False session.finish_output() session.close_pty() return True + async def _kill_remaining_process_group(self, session: _TerminalSession) -> bool: + """ + 有界重试组长退出后的进程组收口，兼容 macOS 后代退出期间的临时权限错误。 + """ + deadline = asyncio.get_running_loop().time() + TERMINAL_KILL_GRACE_SECONDS + while True: + try: + self._send_signal(session, _KILL_SIGNAL) + return True + except PermissionError: + if asyncio.get_running_loop().time() >= deadline: + logger.error(f"终端进程组关闭被拒绝: session_id={session.session_id}, pid={session.pid}") + return False + await asyncio.sleep(TERMINAL_PTY_POLL_INTERVAL) + @staticmethod async def _wait_for_exit(session: _TerminalSession) -> bool: """ @@ -830,7 +848,7 @@ class _TerminalSessionManager: @staticmethod def _send_signal(session: _TerminalSession, sig: int) -> None: """ - 优先向进程组发信号，失败时回退到单进程。 + 按平台向所属进程组或进程发信号，已消失的目标视为收尾完成。 """ try: if os.name == "posix": @@ -851,6 +869,21 @@ class _TerminalSessionManager: """ return sum(1 for session in self._sessions.values() if session.status == "running") + @staticmethod + def _process_group_exists(session: _TerminalSession) -> bool: + """ + 保留组长退出但仍有后代的终端归属，避免过期清理先丢弃唯一收口句柄。 + """ + if os.name != "posix": + return session.status == "running" + try: + os.killpg(session.pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True + def _cleanup_finished_sessions_locked(self) -> None: """ 清理已经结束且超过保留时间的会话。 @@ -861,6 +894,7 @@ class _TerminalSessionManager: for session_id, session in self._sessions.items() if session.status != "running" and now - session.updated_at > TERMINAL_RETENTION_SECONDS + and not self._process_group_exists(session) ] for session_id in expired_ids: session = self._sessions.pop(session_id) diff --git a/app/agent/tools/impl/_file_write_utils.py b/app/agent/tools/impl/_file_write_utils.py index a827add02..740806ea6 100644 --- a/app/agent/tools/impl/_file_write_utils.py +++ b/app/agent/tools/impl/_file_write_utils.py @@ -5,7 +5,25 @@ Agent 文件写入工具的共享辅助函数。 import hashlib import os import tempfile +import threading from pathlib import Path +from weakref import WeakValueDictionary + +_write_locks: WeakValueDictionary[str, threading.Lock] = WeakValueDictionary() +_write_locks_guard = threading.Lock() + + +def _write_lock(path: Path) -> threading.Lock: + """ + 为同一规范目标保留进程内互斥，闲置后自动回收路径锁。 + """ + key = os.path.normcase(str(path.resolve())) + with _write_locks_guard: + lock = _write_locks.get(key) + if lock is None: + lock = threading.Lock() + _write_locks[key] = lock + return lock class FileVersionConflictError(RuntimeError): @@ -29,9 +47,26 @@ def atomic_write_text( path: Path, content: str, expected_sha256: str | None = None, -) -> None: + *, + overwrite: bool = True, +) -> str: """ - 校验目标版本后，在同目录写入临时文件并原子替换文本。 + 在同目标写锁内校验版本并原子替换，返回本次提交内容的SHA256。 + + :param path: 已通过权限检查的目标路径 + :param content: 本次完整文本内容 + :param expected_sha256: 当前版本摘要；None表示未提供版本前提 + :param overwrite: False使用原子排他创建，已有目标不得替换 + :return: 本次实际提交文本的SHA256，不重新读取可能被后续调用修改的文件 + :raises FileVersionConflictError: 版本变化或排他创建时目标已存在 + """ + with _write_lock(path): + return _write_text_locked(path, content, expected_sha256, overwrite=overwrite) + + +def _write_text_locked(path: Path, content: str, expected_sha256: str | None, *, overwrite: bool) -> str: + """ + 在已持有目标锁时写入临时文件，禁止覆盖使用原子link拒绝已存在目标。 """ path.parent.mkdir(parents=True, exist_ok=True) descriptor, temp_name = tempfile.mkstemp( @@ -55,7 +90,14 @@ def atomic_write_text( raise FileVersionConflictError(str(path)) if path.exists(): os.chmod(temp_path, path.stat().st_mode) - os.replace(temp_path, path) + if overwrite: + os.replace(temp_path, path) + else: + try: + os.link(temp_path, path) + except FileExistsError as error: + raise FileVersionConflictError(str(path)) from error + return hashlib.sha256(content.encode("utf-8")).hexdigest() finally: if temp_path.exists(): temp_path.unlink() diff --git a/app/agent/tools/impl/apply_patch.py b/app/agent/tools/impl/apply_patch.py index 8db356e11..2c49e26f9 100644 --- a/app/agent/tools/impl/apply_patch.py +++ b/app/agent/tools/impl/apply_patch.py @@ -348,11 +348,9 @@ class ApplyPatchTool(MoviePilotTool): f"错误：文件 {resolved_path} 在应用补丁期间被创建，拒绝覆盖。" "请确认文件状态后重新应用补丁。" ) - await self.run_blocking( - "default", atomic_write_text, local_path, new_content, sha256 - ) new_sha256 = await self.run_blocking( - "default", calculate_file_sha256, local_path + "default", atomic_write_text, local_path, new_content, sha256, + overwrite=file_patch.operation != "add", ) verb = "新增" if file_patch.operation == "add" else "更新" results.append(f"{verb} {resolved_path}（sha256={new_sha256}）") diff --git a/app/agent/tools/impl/edit_file.py b/app/agent/tools/impl/edit_file.py index d9abac148..78abf91b6 100644 --- a/app/agent/tools/impl/edit_file.py +++ b/app/agent/tools/impl/edit_file.py @@ -151,16 +151,13 @@ class EditFileTool(MoviePilotTool): new_text, -1 if replace_all else 1, ) - await self.run_blocking( + new_sha256 = await self.run_blocking( "default", atomic_write_text, local_path, new_content, current_sha256, ) - new_sha256 = await self.run_blocking( - "default", calculate_file_sha256, local_path - ) logger.info( f"成功编辑文件 {resolved_path}，替换了 {replacement_count} 处内容" diff --git a/app/agent/tools/impl/view_image.py b/app/agent/tools/impl/view_image.py index f1c6f5de7..44839096d 100644 --- a/app/agent/tools/impl/view_image.py +++ b/app/agent/tools/impl/view_image.py @@ -224,7 +224,7 @@ class ViewImageTool(MoviePilotTool): if has_file_path: resolved_path, access_error = await self._check_local_file_access( clean_file_path, - operation="读取图片", + operation="读取", ) if access_error: return self._failure("local_path_access_denied", access_error, source_type) diff --git a/app/agent/tools/impl/write_file.py b/app/agent/tools/impl/write_file.py index 6d0ff040f..8ac91816a 100644 --- a/app/agent/tools/impl/write_file.py +++ b/app/agent/tools/impl/write_file.py @@ -129,15 +129,13 @@ class WriteFileTool(MoviePilotTool): "请重新读取文件并基于最新内容写入。" ) - await self.run_blocking( + new_sha256 = await self.run_blocking( "default", atomic_write_text, local_path, content, current_sha256, - ) - new_sha256 = await self.run_blocking( - "default", calculate_file_sha256, local_path + overwrite=overwrite, ) logger.info(f"成功写入文件 {resolved_path}") diff --git a/app/agent/tools/manager.py b/app/agent/tools/manager.py index ff7f3447e..42a8e4ebe 100644 --- a/app/agent/tools/manager.py +++ b/app/agent/tools/manager.py @@ -322,6 +322,8 @@ class MoviePilotToolsManager: """ 根据字段类型规范化单个值 """ + if value is None: + return None if field_type == "integer" and isinstance(value, str): try: return int(value) diff --git a/app/application/messaging/agent.py b/app/application/messaging/agent.py index b655ac485..8174cdaf4 100644 --- a/app/application/messaging/agent.py +++ b/app/application/messaging/agent.py @@ -583,9 +583,12 @@ async def save_web_agent_display_snapshot( client_session_id: Optional[str] = None, service: Optional[AgentChatService] = None, persistence: Optional[AgentChatPersistenceService] = None, + append: bool = False, ) -> None: """ 保存 WebAgent 展示快照前重新检查归属，避免入口之后的会话变更造成越权覆盖。 + + :param append: 后台单轮结果为True，按消息ID追加合并；前端完整快照保留替换合同 """ if service is None: # 直接调用该内部 helper 时没有 FastAPI 依赖注入上下文。 @@ -596,7 +599,8 @@ async def save_web_agent_display_snapshot( if persistence is None: # 直接调用该内部 helper 时没有 FastAPI 依赖注入上下文。 persistence = get_configured_agent_chat_persistence() - await persistence.async_save_display_messages( + save_messages = persistence.async_append_display_messages if append else persistence.async_save_display_messages + await save_messages( session_id=session_id, user_id=(existing_chat.user_id if existing_chat else str(current_user.id)), username=(existing_chat.username if existing_chat else current_user.name), @@ -1569,6 +1573,7 @@ def _build_traditional_web_agent_stream( client_session_id=command.session_id or session_id, service=service, persistence=persistence, + append=True, ) except Exception as err: logger.error(f"保存WebAgent传统消息快照失败: {str(err)}") diff --git a/app/application/messaging/webagent/stream.py b/app/application/messaging/webagent/stream.py index 6725da10e..c800fa897 100644 --- a/app/application/messaging/webagent/stream.py +++ b/app/application/messaging/webagent/stream.py @@ -346,6 +346,7 @@ def _build_web_agent_event_generator( client_session_id=command.session_id or session_id, service=service, persistence=persistence, + append=True, ) except Exception as err: logger.error(f"保存WebAgent展示历史失败：{err}") diff --git a/app/chain/message.py b/app/chain/message.py index 4e2bc31a0..30e1f27e9 100644 --- a/app/chain/message.py +++ b/app/chain/message.py @@ -4,6 +4,7 @@ import asyncio import base64 +import json import mimetypes import re import threading @@ -146,7 +147,7 @@ class MessageChain(ChainBase): _ai_prefix = "/ai" _no_ai_prefix = "/noai" - # 用户会话信息 {userid: (session_id, last_time)} + # 用户会话信息按渠道、接入源与用户组合键隔离，旧直接调用保留原始键。 _user_sessions: Dict[Union[str, int], tuple[str, datetime]] = {} # 会话超时时间（分钟） _session_timeout_minutes: int = 24 * 60 @@ -186,10 +187,24 @@ class MessageChain(ChainBase): """ 用类级兼容映射构建可测试的用户会话服务。 """ + def expired(session_id: str, user_key: Union[str, int]) -> None: + """ + 映射键只隔离入口，会话回收仍使用真正的渠道用户身份。 + """ + user_id = user_key + if isinstance(user_key, str) and user_key.startswith("message:"): + try: + identity = json.loads(user_key.removeprefix("message:")) + if isinstance(identity, list) and len(identity) == 3 and isinstance(identity[2], str): + user_id = identity[2] + except ValueError: + pass + self._schedule_agent_session_clear(session_id, user_id) + return MessageSessionService( sessions=self._user_sessions, timeout_minutes=self._session_timeout_minutes, - expired_handler=self._schedule_agent_session_clear, + expired_handler=expired, ) def _plugin_input_interaction_handler(self) -> PluginInputInteractionHandler: @@ -466,7 +481,7 @@ class MessageChain(ChainBase): if images or audio_refs or files or has_audio_input: return False - session_info = self._user_sessions.get(userid) + session_info = self._user_sessions.get(self._session_user_key(userid, channel, source)) if not session_info: return False session_id, _ = session_info @@ -965,7 +980,7 @@ class MessageChain(ChainBase): prompt=request.prompt, selected_label=option.label, ) - self._bind_session_id(context.user_id, request.session_id) + self._bind_session_id(self._session_user_key(context.user_id, context.channel, context.source), request.session_id) return self._handle_ai_message( text=selected_text, channel=context.channel, @@ -1005,6 +1020,18 @@ class MessageChain(ChainBase): text=feedback_text, ) + @staticmethod + def _session_user_key(userid: Union[str, int], channel: NotificationChannel, source: Optional[str]) -> str: + """ + 以无歧义序列化生成渠道和接入源隔离的会话键，不回退读取旧裸用户绑定。 + + :param userid: 渠道内原始用户标识 + :param channel: 接收消息的渠道 + :param source: 该渠道的配置实例，未提供时使用空标识 + :return: 内部会话映射键，不作为发送消息的用户ID + """ + return "message:" + json.dumps([channel.value, source or "", str(userid)], ensure_ascii=False, separators=(",", ":")) + def _get_or_create_session_id(self, userid: Union[str, int]) -> str: """ 获取或创建会话ID @@ -1033,7 +1060,7 @@ class MessageChain(ChainBase): :param userid: 用户 ID :param session_id: 智能体会话 ID """ - self._bind_session_id(userid, session_id) + self._bind_session_id(self._session_user_key(userid, NotificationChannel.WebAgent, "web-agent"), session_id) def _record_user_message( self, @@ -1084,7 +1111,7 @@ class MessageChain(ChainBase): """ 清除用户会话（远程命令接口） """ - session_id = self._message_session_service().clear(userid) + session_id = self._message_session_service().clear(self._session_user_key(userid, channel, source)) if session_id: logger.info(f"已清除用户 {userid} 的会话: {session_id}") @@ -1124,7 +1151,7 @@ class MessageChain(ChainBase): 停止后用户仍可继续对话。 """ # 查找用户的会话ID（不弹出，保留会话） - session_info = self._message_session_service().get(userid) + session_info = self._message_session_service().get(self._session_user_key(userid, channel, source)) if session_info: session_id, _ = session_info manager = get_running_agent_manager() @@ -1303,7 +1330,7 @@ class MessageChain(ChainBase): """ 查询当前用户的智能体会话状态。 """ - session_info = self._message_session_service().get(userid) + session_info = self._message_session_service().get(self._session_user_key(userid, channel, source)) if not session_info: self.post_message( Message( @@ -1408,8 +1435,9 @@ class MessageChain(ChainBase): ) return False - session_id = session_id or self._get_or_create_session_id(userid) - self._bind_session_id(userid, session_id) + session_key = self._session_user_key(userid, channel, source) + session_id = session_id or self._get_or_create_session_id(session_key) + self._bind_session_id(session_key, session_id) # 将可直接输入给 LLM 的附件统一转换为 data URL original_images = images diff --git a/app/startup/initializers/agent.py b/app/startup/initializers/agent.py index cd2f85775..21dea2f51 100644 --- a/app/startup/initializers/agent.py +++ b/app/startup/initializers/agent.py @@ -32,6 +32,7 @@ from app.application.messaging.skill import ( reset_skill_catalog_provider, ) from app.runtime.events import Event, eventmanager +from app.runtime.execution import run_in_threadpool_to_completion from app.runtime.log import logger from app.runtime.settings import get_runtime_setting from app.schemas.types import EventType @@ -124,9 +125,10 @@ def _get_agent_manager() -> Any: """ if agent_manager is not None: return agent_manager + manager = get_runtime_agent_manager() if _agent_data_context is not None: - return _get_injected_agent_manager() - return get_runtime_agent_manager() + manager.configure_data_context(_agent_data_context) + return manager def _get_running_agent_manager() -> Any | None: @@ -236,14 +238,15 @@ class AgentInitializer: try: self._shutdown_started = False self._shutdown_complete = False - if agent_manager is not None or _agent_data_context is not None: + if agent_manager is not None: if not get_runtime_setting("AI_AGENT_ENABLE"): logger.info("AI智能体功能未启用") return True - self._manager = agent_manager if agent_manager is not None else _get_injected_agent_manager() + self._manager = agent_manager self._compat_injected = True await self._manager.initialize() else: + await self._bind_runtime_data() self._manager = await activate_agent_service() self._compat_injected = False if self._manager is None: @@ -257,6 +260,14 @@ class AgentInitializer: logger.error(f"AI智能体管理器初始化失败: {e}") return False + @staticmethod + async def _bind_runtime_data() -> None: + """ + 首启及关闭后重新启用前，在受控线程给canonical实例绑定同一批数据端口。 + """ + if _agent_data_context is not None and get_runtime_setting("AI_AGENT_ENABLE"): + await run_in_threadpool_to_completion(_get_agent_manager) + async def handle_config_changed(self, event: Event) -> None: """ 仅在 manifest watch 命中时协调 service，关闭态保持 fail closed。 @@ -265,6 +276,8 @@ class AgentInitializer: if not changed_keys or self._compat_injected or self._shutdown_started or self._shutdown_complete: return try: + if "AI_AGENT_ENABLE" in changed_keys: + await self._bind_runtime_data() self._manager = await reconcile_agent_service( reason="agent_service_config_changed", changed_keys=changed_keys, diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 1523e4896..d74e1c354 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -817,7 +817,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1129 | -| 内部导入边 | 9,710 | +| 内部导入边 | 9,711 | | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） | diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 8238390d5..adbaac5e2 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -350,6 +350,10 @@ Emby、Jellyfin、Plex、Navidrome 的目录、条目、搜索、季集和活动 - 设置更新优先回传 `expected_revision`。发生冲突时重新读取并比较，不直接覆盖；`generic_write_allowed=false` 时使用返回的专用操作。 - 插件存在分身时，显式选择目标实例或使用已经配置的默认目标。遇到实例歧义、停用目标或插件门禁拒绝时，按错误处理，不猜测另一个实例。 - 参数省略、空字符串与 `null` 的含义由具体操作决定。例如更新订阅时，省略字段通常保留原值，允许清空的字段可使用空值；不要把整份读取结果不加筛选地回写。 +- `agent_task` 更新的 `enabled:null` 保留原启用状态，仅重命名暂停任务不会恢复调度。MCP 工具发现和 Google 模型发现会读取全部分页；重复游标或超过分页上限时报告失败。 +- 传统消息会话按渠道、配置接入源与原始用户 ID 隔离，停止、清除、状态和确认操作使用同一范围。WebAgent 后台完成追加本轮展示历史，前端完整快照仍按原合同替换。 +- Agent 文件工具的禁止覆盖创建使用排他提交；同目标并发版本修改只允许仍匹配原 SHA 的请求提交。冲突后重新读取当前文件再编辑。`view_image` 使用只读权限，普通用户可读取允许范围内的公共技能图片。 +- 安装市场技能保留用户明确选择的 GitHub 分支，该分支不可用时不改装默认分支。Agent 运行时人格文件变化使缓存失效；关闭 Agent 配置会通过能力 owner 停止服务。 - 订阅身份包含用户名、媒体类型、来源、媒体 ID、季号、剧集组和音乐实体类型。同编号的电影与电视剧可分别订阅；按来源定位时保留剧集组。同用户并发创建同一目标会复用真实订阅 ID。 - 更新订阅的作品、季号、剧集组或音乐实体后，旧目标的下载与洗版进度在同一事务清除，缺失集数按新总集数重新开始；显式清空剧集组也属于目标切换。目标身份已存在时取消该次修改，原有两条订阅及进度保持不变。 - Servarr 重复添加已监控的剧集返回真实既有订阅 ID。搜索任务取消后若执行者丢失，租约到期会结算取消并释放活动占位，再次搜索可重新入队。 diff --git a/scripts/generate_agent_skill_docs.py b/scripts/generate_agent_skill_docs.py index a2058cd28..52f36675b 100644 --- a/scripts/generate_agent_skill_docs.py +++ b/scripts/generate_agent_skill_docs.py @@ -17,7 +17,7 @@ PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) -MOVIEPILOT_API_SKILL_VERSION = "62" +MOVIEPILOT_API_SKILL_VERSION = "63" DATABASE_OPERATION_SKILL_VERSION = "14" diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index d86351f9d..a18b1b7e7 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -1,6 +1,6 @@ --- name: moviepilot-api -version: 62 +version: 63 description: >- Use this skill for MoviePilot product operations such as media search, torrent search, downloads, subscriptions, library checks, sites, storage, workflows, @@ -340,6 +340,14 @@ token exchange downloads retain their existing credential rules. `read_tool_result` for the next page instead of repeating the operation. These receipt and result tools are internal to the Agent, not external MCP tools. +## Agent Task And Conversation State + +- For `agent_task` updates, `enabled=null` preserves the current enabled state. Renaming a paused task does not resume its schedule. +- Traditional message sessions are scoped by channel, configured source, and user ID. A stop, clear, status query, or confirmation targets the current source's session. +- Background WebAgent completion appends the current turn to display history. Frontend complete snapshots keep their existing replacement contract. +- File creation with `overwrite=false` rejects a concurrent creator. Versioned edits reject a peer's intervening write; reread the file and use its current SHA before retrying. +- Market skill installation preserves an explicitly selected GitHub branch. If that branch is unavailable, treat installation as failed instead of silently switching branches. + ## Subscription Identity And Workflow Recovery - Preserve media type, source, media ID, season, episode group, and music entity type when finding or creating a subscription. Movies and TV series with the same numeric ID are distinct targets. Keep the episode group in source-based lookup. diff --git a/tests/fixtures/architecture/concurrency-baseline.json b/tests/fixtures/architecture/concurrency-baseline.json index 94fd3e056..6ffde6cc2 100644 --- a/tests/fixtures/architecture/concurrency-baseline.json +++ b/tests/fixtures/architecture/concurrency-baseline.json @@ -667,6 +667,12 @@ }, "target": "asyncio.to_thread" }, + "app/startup/initializers/agent.py:app.runtime.execution.run_in_threadpool_to_completion": { + "owners": { + "AgentInitializer._bind_runtime_data": 1 + }, + "target": "app.runtime.execution.run_in_threadpool_to_completion" + }, "app/startup/initializers/modules.py:app.runtime.execution.run_in_threadpool_to_completion": { "owners": { "stop_modules.run_step": 1 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 95c4c169f..8b65a7fb0 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -481,13 +481,14 @@ ] }, { - "source": "app.modules.discord.discord", - "target": "discord", - "kind": "network_sdk", "bindings": [ "from:discord.app_commands as app_commands", "import:discord as discord" ], + "fingerprint": "2761ef9a9e5a805bfb20ea832c5d34edea12db21ca3eb27c3db7ce512a5ee7f5", + "kind": "network_sdk", + "source": "app.modules.discord.discord", + "target": "discord", "uses": [ "Discord.__init__|call:Client", "Discord.__init__|call:Intents.default", @@ -528,8 +529,7 @@ "Discord._sync_registered_commands|call:sync", "Discord._sync_registered_commands|call:wait_until_ready", "Discord.stop|call:close" - ], - "fingerprint": "2761ef9a9e5a805bfb20ea832c5d34edea12db21ca3eb27c3db7ce512a5ee7f5" + ] }, { "bindings": [ @@ -959,8 +959,8 @@ "runtime_only": true } }, - "edge_count": 9710, - "edge_sha256": "488e191954d9ab1e56db00a899877988a36d9e5c94e1ddd6ee40257a3b7b434c", + "edge_count": 9711, + "edge_sha256": "ab6b9c2a3353c8520468b3f7752ed1ed1c2c208a093afb50a7a5e6895d6f8ea0", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -10178,6 +10178,7 @@ "app.startup.initializers.agent -> app.chain.message", "app.startup.initializers.agent -> app.runtime", "app.startup.initializers.agent -> app.runtime.events", + "app.startup.initializers.agent -> app.runtime.execution", "app.startup.initializers.agent -> app.runtime.log", "app.startup.initializers.agent -> app.runtime.settings", "app.startup.initializers.agent -> app.schemas", diff --git a/tests/test_agent_apply_patch.py b/tests/test_agent_apply_patch.py index 7c07a1699..b8eb655fc 100644 --- a/tests/test_agent_apply_patch.py +++ b/tests/test_agent_apply_patch.py @@ -211,9 +211,12 @@ def test_apply_patch_detects_version_conflict_during_write(tmp_path): ) original_write = apply_patch_module.atomic_write_text - def _conflicting_write(path, content, expected_sha256=None): + def _conflicting_write(path, content, expected_sha256=None, *, overwrite=True): + """ + 在补丁提交前模拟另一写者修改，验证正式原子写入口拒绝旧版本。 + """ file_path.write_text("changed elsewhere\n", encoding="utf-8") - original_write(path, content, expected_sha256) + return original_write(path, content, expected_sha256, overwrite=overwrite) with patch.object( apply_patch_module, "atomic_write_text", _conflicting_write diff --git a/tests/test_agent_data_ports.py b/tests/test_agent_data_ports.py index d01818863..e70868e2d 100644 --- a/tests/test_agent_data_ports.py +++ b/tests/test_agent_data_ports.py @@ -13,6 +13,7 @@ import pytest import app.agent.tools.factory as tool_factory_module import app.application.agent as agent_services +from app.agent.manager import AgentManager from app.agent.memory import MemoryManager from app.agent.orchestrator import MoviePilotAgent from app.agent.session import AgentSessionOwner, _MessageTask @@ -514,11 +515,14 @@ def test_agent_initializer_resolves_data_bound_manager_for_agent_task(monkeypatc monkeypatch.setattr(agent_initializer, "agent_manager", None) monkeypatch.setattr(agent_initializer, "_agent_data_context", None) monkeypatch.setattr(agent_initializer, "_injected_agent_manager", None) + canonical = AgentManager(memory=MemoryManager()) + monkeypatch.setattr(agent_initializer, "get_runtime_agent_manager", lambda: canonical) agent_initializer.configure_agent_data_context(context) manager = agent_initializer._get_agent_manager() assert manager._data is context + assert manager is canonical def test_transactional_agent_task_repository_projects_frozen_snapshots() -> None: diff --git a/tests/test_agent_interaction.py b/tests/test_agent_interaction.py index ad19722f5..97364c3f3 100644 --- a/tests/test_agent_interaction.py +++ b/tests/test_agent_interaction.py @@ -307,10 +307,11 @@ def test_secret_confirmation_preempts_plugin_interaction_on_message_channels() - TG/飞书确认必须回到已有 Agent 会话，不被其它输入会话消费。 """ chain = MessageChain() - MessageChain._user_sessions["10001"] = ("session-secret", datetime.now()) try: for channel in (NotificationChannel.Telegram, NotificationChannel.Feishu): + key = chain._session_user_key("10001", channel, f"{channel.value}-test") + MessageChain._user_sessions[key] = ("session-secret", datetime.now()) manager = Mock() manager.matches_secret_confirmation.return_value = True with ( diff --git a/tests/test_agent_runtime_review_regressions.py b/tests/test_agent_runtime_review_regressions.py new file mode 100644 index 000000000..7df68660e --- /dev/null +++ b/tests/test_agent_runtime_review_regressions.py @@ -0,0 +1,786 @@ +""" +智能体运行时审查发现的真实任务、文件、历史和发现回归。 +""" + +import asyncio +import io +import json +import os +import shlex +import signal +import sys +import zipfile +from concurrent.futures import ThreadPoolExecutor +from contextlib import asynccontextmanager +from pathlib import Path +from threading import Barrier +from types import MethodType, SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from PIL import Image + +from app.agent.history.storage import SqliteRecallRepository +from app.agent.middleware.subagents import SubAgentTaskControlMiddleware +from app.agent.runtime import AgentRuntimeManager +from app.agent.tools.impl import _file_write_utils as writes +from app.application.messaging.recall import RecallMessage, RecallQuery, RecallSession + + +def test_runtime_disk_change_invalidates_loaded_persona(tmp_path): + """ + 外部改动人格文件后签名与实际缓存不能被提前写成同一版本。 + """ + manager = AgentRuntimeManager(agent_root_dir=tmp_path, + bundled_defaults_dir=Path(__file__).parents[1] / "app/agent/defaults") + manager._signature_check_interval = 0 + first = manager.load_runtime_config() + target = first.persona.path + target.write_text(target.read_text() + "\nREVIEW_CHANGED_PERSONA\n") + second = manager.load_runtime_config() + assert "REVIEW_CHANGED_PERSONA" in second.persona.text + assert second is not first + + +def test_atomic_write_serializes_sha_compare_and_replace(tmp_path, monkeypatch): + """ + 两个线程同时通过旧SHA检查后，只允许一个版本成功提交。 + """ + target = tmp_path / "target.txt" + target.write_text("old") + previous = writes.calculate_file_sha256(target) + rendezvous = Barrier(2) + real_hash = writes.calculate_file_sha256 + + def compared(path): + """ + 在真实哈希读取后同步，固定旧实现两个校验均成功的窗口。 + """ + digest = real_hash(path) + try: + rendezvous.wait(timeout=0.3) + except Exception: + pass + return digest + + monkeypatch.setattr(writes, "calculate_file_sha256", compared) + + def update(value): + """ + 保留真实临时文件、fsync、版本核验和原子替换。 + """ + try: + writes.atomic_write_text(target, value, previous) + return "changed" + except writes.FileVersionConflictError: + return "conflict" + + with ThreadPoolExecutor(max_workers=2) as pool: + outcomes = list(pool.map(update, ["first", "second"])) + assert sorted(outcomes) == ["changed", "conflict"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("edit", [False, True]) +async def test_concurrent_file_tools_do_not_overwrite_peer(tmp_path, monkeypatch, edit): + """ + 从真实工具入口同时完成读检查后，新建禁止覆盖和编辑SHA都只允许一个提交。 + """ + from app.agent.tools.impl.edit_file import EditFileTool + from app.agent.tools.impl.write_file import WriteFileTool + + target = tmp_path / "file.txt" + if edit: + target.write_text("alpha beta") + barrier = Barrier(2) + actual_write = writes.atomic_write_text + + def write(*args, **kwargs): + """ + 在真实原子写入前加屏障，固定两个调用已通过读侧检查的窗口。 + """ + barrier.wait(timeout=5) + return actual_write(*args, **kwargs) + + monkeypatch.setattr("app.agent.tools.impl.edit_file.atomic_write_text" if edit else + "app.agent.tools.impl.write_file.atomic_write_text", write) + tool = EditFileTool(session_id="review", user_id="review") if edit else WriteFileTool(session_id="review", user_id="review") + tool.set_agent_context({"is_admin": True}) + calls = [tool.run(file_path=str(target), old_text=old, new_text=new) + for old, new in [("alpha", "ALPHA"), ("beta", "BETA")]] if edit else [ + tool.run(file_path=str(target), content=value, overwrite=False) for value in ("first", "second")] + results = await asyncio.gather(*calls) + assert sum(result.startswith("成功") for result in results) == 1 + assert writes.calculate_file_sha256(target) in next(result for result in results if result.startswith("成功")) + + +def test_remote_session_clear_only_clears_current_channel_source(monkeypatch): + """ + 清除一个渠道接入不删除相同裸用户在另一接入的会话。 + """ + from datetime import datetime + + from app.chain.message import MessageChain + from app.schemas.types import NotificationChannel + + monkeypatch.setattr(MessageChain, "_user_sessions", {}) + chain = MessageChain() + monkeypatch.setattr(chain, "_schedule_agent_session_clear", lambda *_args: None) + monkeypatch.setattr(chain, "post_message", lambda *_args: None) + first = chain._session_user_key("same", NotificationChannel.Telegram, "one") + second = chain._session_user_key("same", NotificationChannel.Telegram, "two") + chain._user_sessions.update({first: ("first", datetime.now()), second: ("second", datetime.now())}) + chain.remote_clear_session(NotificationChannel.Telegram, "same", "one") + assert first not in chain._user_sessions + assert chain._user_sessions[second][0] == "second" + + +@pytest.mark.asyncio +async def test_finished_subagent_callback_keeps_each_identity(): + """ + 同批两个子任务先后结束时，完成时间属于实际结束的任务。 + """ + middleware = SubAgentTaskControlMiddleware(model=None, profiles=(), tools=[]) + first, second = asyncio.Event(), asyncio.Event() + + async def provider(*, description, **_kwargs): + """ + 用独立事件代替远端模型，保留真实任务调度和完成回调。 + """ + await (first if description == "first" else second).wait() + return description + + middleware._provider = SimpleNamespace(run_task=provider) + try: + result = json.loads(await middleware._control_task(action="start", + tasks=[{"description": "first"}, {"description": "second"}])) + records = [middleware._tasks[item["task_id"]] for item in result["tasks"]] + first.set() + await asyncio.wait_for(asyncio.shield(records[0].task), 2) + await asyncio.sleep(0) + assert records[0].finished_at is not None + assert records[1].finished_at is None + finally: + first.set() + second.set() + await middleware.close() + + +@pytest.mark.asyncio +async def test_update_subagent_cannot_restart_after_owner_close(): + """ + 更新等待旧任务取消时发生关闭，旧任务收敛后不能产生逃逸新任务。 + """ + middleware = SubAgentTaskControlMiddleware(model=None, profiles=(), tools=[]) + started, cancelling, finish = asyncio.Event(), asyncio.Event(), asyncio.Event() + new_tasks = [] + + async def provider(*, description, **_kwargs): + """ + 旧任务延迟确认取消，让关闭精确进入更新的等待窗口。 + """ + if description == "old": + started.set() + while not finish.is_set(): + try: + await finish.wait() + except asyncio.CancelledError: + cancelling.set() + return "old complete" + new_tasks.append(asyncio.current_task()) + await finish.wait() + return "new complete" + + middleware._provider = SimpleNamespace(run_task=provider) + updating = closing = None + try: + result = json.loads(await middleware._control_task(action="start", description="old")) + await started.wait() + updating = asyncio.create_task(middleware._control_task(action="update", + task_id=result["tasks"][0]["task_id"], description="new")) + await cancelling.wait() + closing = asyncio.create_task(middleware.close()) + await asyncio.sleep(0) + finish.set() + assert json.loads(await updating)["success"] is False + assert await closing is True + assert not new_tasks + finally: + finish.set() + await asyncio.gather(*(task for task in (updating, closing) if task), return_exceptions=True) + await asyncio.gather(*new_tasks, return_exceptions=True) + await middleware.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("repeated_cursor", [False, True]) +async def test_mcp_discovery_consumes_next_cursor(monkeypatch, repeated_cursor): + """ + 同一真实MCP会话依次发现各页工具，不把首屏当成完整目录。 + """ + from app.agent.mcp import AgentMcpManager + from app.schemas.agent import AgentMcpServerConfig + + requests = [] + + async def request(method, params=None): + """ + 只替换远端响应，保留初始化和工具投影。 + """ + if method == "initialize": + return {"protocolVersion": "2025-11-25", "capabilities": {"tools": {}}} + requests.append(params) + if params and params.get("cursor") == "page-two": + return {"tools": [{"name": "second"}], "nextCursor": "page-two" if repeated_cursor else None} + return {"tools": [{"name": "first"}], "nextCursor": "page-two"} + + @asynccontextmanager + async def session(_server): + """ + 会话只持有本次测试的调用记录。 + """ + yield SimpleNamespace(request=request, notify=AsyncMock()) + + async def open_session(server): + """ + 返回隔离的远端会话，保持正式发现入口的异步打开协议。 + """ + return session(server) + + monkeypatch.setattr("app.agent.mcp._open_mcp_session", open_session) + server = AgentMcpServerConfig(id="review", name="review", transport="http", url="https://review.invalid/mcp") + if repeated_cursor: + with pytest.raises(ValueError, match="重复"): + await AgentMcpManager().list_server_tools(server) + assert len(requests) == 2 + return + tools = await AgentMcpManager().list_server_tools(server) + assert [tool.name for tool in tools] == ["first", "second"] + assert requests == [None, {"cursor": "page-two"}] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("repeated_cursor", [False, True]) +async def test_google_model_discovery_consumes_real_sdk_pager(monkeypatch, repeated_cursor): + """ + 使用真实SDK分页载具验证第二页模型确实由应用消费。 + """ + from google.genai import pagers, types + + from app.agent.llm.provider import LLMProviderManager + + first = types.ListModelsResponse(models=[types.Model(name="models/first", display_name="First", + supported_actions=["generateContent"])], next_page_token="second") + second = types.ListModelsResponse(models=[types.Model(name="models/second", display_name="Second", + supported_actions=["generateContent"])], next_page_token="second" if repeated_cursor else None) + request = AsyncMock(return_value=second) + pager = pagers.AsyncPager("models", request, first, {}) + client = SimpleNamespace(aio=SimpleNamespace(models=SimpleNamespace(list=AsyncMock(return_value=pager)), + aclose=AsyncMock()), close=lambda: None) + monkeypatch.setattr("google.genai.Client", lambda **_kwargs: client) + manager = LLMProviderManager() + monkeypatch.setattr(manager, "_models_dev_model", AsyncMock(return_value={})) + if repeated_cursor: + with pytest.raises(RuntimeError, match="重复"): + await manager._list_models_from_google("fictional", use_proxy=False) + assert request.await_count == 1 + client.aio.aclose.assert_awaited_once() + return + result = await manager._list_models_from_google("fictional", use_proxy=False) + assert [model["id"] for model in result] == ["models/first", "models/second"] + assert request.await_count == 1 + client.aio.aclose.assert_awaited_once() + + +def test_recall_browse_applies_time_exclusion_and_sort(tmp_path): + """ + 真实独立SQLite浏览执行开始时间范围、排除列表和排序。 + """ + repository = SqliteRecallRepository(tmp_path) + for identity, timestamp in [("old", 100), ("in-range", 200), ("new", 300)]: + repository.append("review", RecallSession(identity, title=identity, started_at=timestamp, + last_active=timestamp), (RecallMessage(identity, "user", identity, timestamp=timestamp),)) + ranged = repository.search("review", RecallQuery(after=150, before=250)) + assert [item["session_id"] for item in ranged["results"]] == ["in-range"] + excluded = repository.search("review", RecallQuery(exclude_session_ids=("new",))) + assert "new" not in [item["session_id"] for item in excluded["results"]] + oldest = repository.search("review", RecallQuery(sort="oldest")) + assert oldest["results"][0]["session_id"] == "old" + + +def test_market_skill_install_preserves_explicit_branch(tmp_path, monkeypatch): + """ + 显式release分支的市场展示和实际下载使用同一分支及来源记录。 + """ + from app.agent.skills.registry import SkillHelper + + requests = [] + + def request(**kwargs): + """ + 用不同真实ZIP内容区分默认分支和选定分支。 + """ + branch = kwargs["url"].rsplit("/", 1)[-1] + requests.append(branch) + stream = io.BytesIO() + with zipfile.ZipFile(stream, "w") as package: + package.writestr(f"repo-{branch}/skills/review-demo/SKILL.md", + f"---\nname: review-demo\ndescription: review\nversion: 2\n---\n{branch} contents\n") + return SimpleNamespace(status_code=200, content=stream.getvalue()) + + monkeypatch.setattr(SkillHelper, "_request_github", staticmethod(request)) + monkeypatch.setattr(SkillHelper, "get_user_skills_dir", staticmethod(lambda: tmp_path)) + helper = SkillHelper() + skills = SkillHelper._list_market_repo_skills.__wrapped__(helper, + "https://github.com/example/review/tree/release/skills") + success, message = helper.install_market_skill(skills[0]) + assert success, message + assert requests == ["release", "release"] + assert "release contents" in (tmp_path / "review-demo/SKILL.md").read_text() + assert json.loads((tmp_path / "review-demo/.moviepilot-skill-source.json").read_text())["branch"] == "release" + + +def test_explicit_market_branch_failure_does_not_install_default(monkeypatch): + """ + 指定分支不可用时受控失败，不悄悄改装默认分支内容。 + """ + from app.agent.skills.registry import SkillHelper + + requests = [] + + def request(**kwargs): + """ + 记录明确分支的下载请求并模拟该分支不存在。 + """ + requests.append(kwargs["url"]) + return SimpleNamespace(status_code=404) + + monkeypatch.setattr(SkillHelper, "_request_github", staticmethod(request)) + helper = SkillHelper() + repo = helper._parse_market_repo("https://github.com/example/review/tree/release/skills") + assert helper._download_repo_archive(repo) is None + assert len(requests) == 1 and requests[0].endswith("/release") + + +@pytest.mark.asyncio +async def test_nullable_tool_boolean_preserves_none(): + """ + 工具规范化不能把合法Optional布尔的null变成启用。 + """ + from app.agent.tools.manager import MoviePilotToolsManager + + assert MoviePilotToolsManager._normalize_scalar_value("boolean", None, "enabled") is None + + +@pytest.mark.asyncio +async def test_mcp_null_task_update_does_not_enable_paused_task(monkeypatch): + """ + 真实MCP公开入口将enabled=null传到任务更新服务时保留暂停状态。 + """ + from app.agent.tools.impl.agent_task import AgentTaskTool + from app.agent.tools.manager import MoviePilotToolsManager + from app.api.endpoints import mcp + + task = SimpleNamespace(id=1, last_status="waiting", enabled=False, trigger_type="cron", + cron_expression="0 0 * * *", run_at=None, name="paused") + payloads = [] + + def update(*, payload, **_kwargs): + """ + 仅替换持久化边界，记录真实工具生成的写入字段。 + """ + payloads.append(dict(payload)) + for key, value in payload.items(): + setattr(task, key, value) + return True + + repository = SimpleNamespace(get=lambda **_kwargs: task, update=update) + tool = AgentTaskTool(session_id="review", user_id="1", data=SimpleNamespace(tasks=repository)) + tool.set_agent_context({"is_admin": True}) + manager = MoviePilotToolsManager(user_id="1", session_id="review") + manager.tools = [tool] + monkeypatch.setattr(mcp, "moviepilot_tool_manager", manager) + monkeypatch.setattr("app.application.scheduling.update_agent_task_job", lambda *_args, **_kwargs: None) + monkeypatch.setattr("app.agent.tools.impl.agent_task.agent_task_to_dict", + lambda item, **_kwargs: {"id": item.id, "name": item.name, "enabled": item.enabled}) + try: + response = await mcp.handle_tools_call({"name": "agent_task", "arguments": { + "action": "update", "task_id": 1, "name": "renamed", "enabled": None, + }}) + assert response.get("isError") is not True + assert task.name == "renamed" and task.enabled is False + assert "enabled" not in payloads[-1] + finally: + await manager.close() + + +@pytest.mark.asyncio +async def test_production_data_context_keeps_config_reconciliation(monkeypatch): + """ + 正式组合根发布数据上下文后，关闭开关仍会协调真实能力owner。 + """ + from app.startup.composition.agent import publish_agent_services + from app.startup.initializers import agent as module + + manager = SimpleNamespace(initialize=AsyncMock(), close=AsyncMock(return_value=True)) + initializer = module.AgentInitializer() + composition = SimpleNamespace(data=SimpleNamespace(chat=object(), chat_persistence=object()), execution=object()) + monkeypatch.setattr(module, "agent_manager", None) + monkeypatch.setattr(module, "_agent_data_context", None) + monkeypatch.setattr(module, "_injected_agent_manager", None) + monkeypatch.setattr(module, "_get_injected_agent_manager", lambda: manager) + monkeypatch.setattr(module, "_get_agent_manager", lambda: manager) + monkeypatch.setattr(module, "get_runtime_setting", lambda _key: True) + monkeypatch.setattr(module, "activate_agent_service", AsyncMock(return_value=manager)) + reconcile = AsyncMock(return_value=None) + monkeypatch.setattr(module, "reconcile_agent_service", reconcile) + for name in ("configure_agent_chat_service", "configure_agent_chat_persistence", "configure_agent_task_execution"): + monkeypatch.setattr("app.startup.composition.agent." + name, lambda *_args: None) + publish_agent_services(composition, data_context_registrar=module.configure_agent_data_context) + assert await initializer.initialize() + await initializer.handle_config_changed(SimpleNamespace(event_data={"key": {"AI_AGENT_ENABLE"}})) + reconcile.assert_awaited_once() + assert initializer._manager is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("initially_enabled", [False, True]) +async def test_production_data_context_uses_one_real_capability_manager(monkeypatch, initially_enabled): + """ + 正式组合根、真实能力owner和真实AgentManager共享同一实例，关闭配置收敛后台任务。 + """ + from app.agent import loader + from app.agent.manager import AgentManager + from app.startup.composition.agent import publish_agent_services + from app.startup.initializers import agent as module + + memory = SimpleNamespace(configure=Mock(), initialize=Mock(), close=AsyncMock()) + manager = AgentManager(memory=memory) + enabled = {"value": initially_enabled} + monkeypatch.setattr(loader, "_agent_runtime", None) + monkeypatch.setattr("app.agent.manager.agent_manager", manager) + monkeypatch.setattr(module, "agent_manager", None) + monkeypatch.setattr(module, "_agent_data_context", None) + monkeypatch.setattr(module, "_injected_agent_manager", None) + monkeypatch.setattr(module, "get_runtime_setting", lambda _key: enabled["value"]) + monkeypatch.setattr("app.agent.capabilities.adapter.get_runtime_setting", lambda _key: enabled["value"]) + initializer = module.AgentInitializer() + monkeypatch.setattr(module, "agent_initializer", initializer) + context = SimpleNamespace(chat=object(), chat_persistence=object()) + for name in ("configure_agent_chat_service", "configure_agent_chat_persistence", "configure_agent_task_execution"): + monkeypatch.setattr("app.startup.composition.agent." + name, lambda *_args: None) + publish_agent_services(SimpleNamespace(data=context, execution=object()), + data_context_registrar=module.configure_agent_data_context) + try: + assert await initializer.initialize() + if not initially_enabled: + assert module._get_running_agent_manager() is None + assert not manager._accepting_tasks + enabled["value"] = True + await initializer.handle_config_changed(SimpleNamespace(event_data={"key": {"AI_AGENT_ENABLE"}})) + assert initializer._compat_injected is False + assert module._get_running_agent_manager() is manager + assert module._get_agent_manager() is manager + assert manager._data is context + assert manager._accepting_tasks + enabled["value"] = False + await initializer.handle_config_changed(SimpleNamespace(event_data={"key": {"AI_AGENT_ENABLE"}})) + assert module._get_running_agent_manager() is None + assert not manager._accepting_tasks + assert manager._idle_cleanup_task is None + memory.close.assert_awaited_once() + finally: + await loader.begin_agent_shutdown() + await initializer.cleanup() + await manager.close() + + +@pytest.mark.asyncio +async def test_message_ingress_separates_channels_and_sources(monkeypatch): + """ + 真实消息入口按渠道和接入源隔离会话，同一入口连续消息仍复用。 + """ + from app.chain.message import MessageChain + from app.schemas.types import NotificationChannel + + received, submitted = [], [] + + async def process(**kwargs): + """ + 记录消息入口交给真实会话路由的身份与内容。 + """ + received.append(kwargs) + + ingress = SimpleNamespace(_ai_prefix="/ai", _user_sessions={}, _session_timeout_minutes=1440, + runtime_config=SimpleNamespace(ai_agent_enable=True), + _schedule_agent_session_clear=lambda *_args: None, _has_ai_prefix=lambda text: text.startswith("/ai"), + _prepare_agent_files=lambda **_kwargs: [], post_message=lambda *_args, **_kwargs: None) + for name in ("_message_session_service", "_get_or_create_session_id", "_bind_session_id", "_handle_ai_message"): + setattr(ingress, name, MethodType(getattr(MessageChain, name), ingress)) + if hasattr(MessageChain, "_session_user_key"): + ingress._session_user_key = MessageChain._session_user_key + original_submit = asyncio.run_coroutine_threadsafe + + def submit(coroutine, loop): + """ + 仅记录跨线程提交回执，实际协程继续在测试事件循环执行。 + """ + future = original_submit(coroutine, loop) + submitted.append(future) + return future + + monkeypatch.setattr("app.chain.message.get_running_agent_manager", lambda: SimpleNamespace(process_message=process)) + monkeypatch.setattr("app.chain.message.main_loop_registry.require", asyncio.get_running_loop) + monkeypatch.setattr("app.chain.message.asyncio.run_coroutine_threadsafe", submit) + for channel, source in [(NotificationChannel.Telegram, "one"), (NotificationChannel.Feishu, "one"), + (NotificationChannel.Telegram, "two"), (NotificationChannel.Telegram, "one")]: + assert ingress._handle_ai_message("/ai review", channel, source, "same", "review") + await asyncio.wrap_future(submitted[-1]) + identities = [item["session_id"] for item in received] + assert len(set(identities[:3])) == 3 + assert identities[0] == identities[3] + + +@pytest.mark.asyncio +async def test_message_namespace_separates_real_agent_memory(monkeypatch): + """ + 真实会话owner和记忆读取链不向同裸用户的另一渠道提供前一渠道消息。 + """ + from langchain_core.messages import AIMessage + + from app.agent.manager import AgentManager + from app.agent.memory import MemoryManager + from app.agent.orchestrator import MoviePilotAgent + from app.chain.message import MessageChain + from app.schemas.types import NotificationChannel + + port = SimpleNamespace(get=AsyncMock(return_value=None), async_save_agent_messages=AsyncMock(), + async_append_display_messages=AsyncMock()) + memory = MemoryManager(chat=port, persistence=port) + manager = AgentManager(memory=memory) + manager._accepting_tasks = True + calls, submitted = [], [] + ingress = SimpleNamespace(_ai_prefix="/ai", _user_sessions={}, _session_timeout_minutes=1440, + runtime_config=SimpleNamespace(ai_agent_enable=True), + _schedule_agent_session_clear=lambda *_args: None, _has_ai_prefix=lambda text: text.startswith("/ai"), + _prepare_agent_files=lambda **_kwargs: [], post_message=lambda *_args, **_kwargs: None, + _session_user_key=MessageChain._session_user_key) + for name in ("_message_session_service", "_get_or_create_session_id", "_bind_session_id", "_handle_ai_message"): + setattr(ingress, name, MethodType(getattr(MessageChain, name), ingress)) + + async def graph(agent, streaming=False): + """ + 只在模型图边界提供确定响应，真实记忆读写和消息owner继续执行。 + """ + del streaming + state = {} + agent._last_agent_cache_hit = False + + async def invoke(payload, config): + """ + 为正式记忆与会话流程提供确定的模型回复，并保留已传入的上下文。 + """ + del config + calls.append(json.dumps([message.content for message in payload["messages"]], ensure_ascii=False)) + state["messages"] = [*payload["messages"], AIMessage(content="review reply")] + + return SimpleNamespace(ainvoke=invoke, get_state=lambda _config: SimpleNamespace(values=state)) + + original_submit = asyncio.run_coroutine_threadsafe + + def submit(coroutine, loop): + """ + 保留跨线程提交回执，供测试等待真实消息流程完成。 + """ + future = original_submit(coroutine, loop) + submitted.append(future) + return future + + monkeypatch.setattr("app.chain.message.get_running_agent_manager", lambda: manager) + monkeypatch.setattr("app.chain.message.main_loop_registry.require", asyncio.get_running_loop) + monkeypatch.setattr("app.chain.message.asyncio.run_coroutine_threadsafe", submit) + monkeypatch.setattr("app.agent.session._async_start_processing_status", AsyncMock(return_value=None)) + monkeypatch.setattr("app.agent.orchestrator.get_configured_agent_chat_persistence", lambda: port) + monkeypatch.setattr(MoviePilotAgent, "_create_agent", graph) + monkeypatch.setattr(MoviePilotAgent, "prepare_chat_title", AsyncMock()) + monkeypatch.setattr(MoviePilotAgent, "_learning_enabled", lambda _self: False) + monkeypatch.setattr(MoviePilotAgent, "_should_stream", lambda _self: False) + monkeypatch.setattr(MoviePilotAgent, "send_agent_message", AsyncMock()) + try: + for text, channel, source in [("first private", NotificationChannel.Telegram, "one"), + ("second private", NotificationChannel.Feishu, "two"), + ("third private", NotificationChannel.Telegram, "one")]: + assert ingress._handle_ai_message("/ai " + text, channel, source, "same", "review") + await asyncio.wrap_future(submitted[-1]) + key = ingress._session_user_key("same", channel, source) + session_id = ingress._user_sessions[key][0] + await manager._session_queues[session_id].join() + assert len(manager.active_agents) == 2 + assert "first private" not in calls[1] + assert "second private" in calls[1] + assert "first private" in calls[2] + assert "third private" in calls[2] + assert "second private" not in calls[2] + finally: + await manager.close() + + +@pytest.mark.asyncio +async def test_webagent_background_turns_preserve_display_history(tmp_path, monkeypatch): + """ + 两轮真实SQLite后台快照保存完整展示历史，不依赖前端二次上传。 + """ + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + from app.application.messaging import agent as application + from app.application.messaging.chat import AgentChatPersistenceService, AgentChatService + from app.application.messaging.webagent.stream import WebAgentStreamDependencies, build_agent_web_agent_stream + from app.db.models.agentchat import AgentChat + from app.db.oper.agentchat import AgentChatOper + + engine = create_engine(f"sqlite:///{tmp_path / 'chat.db'}") + AgentChat.__table__.create(engine) + factory = sessionmaker(engine, expire_on_commit=False) + + async def get(session_id, user_id=None): + """ + 每次查询使用真实短会话。 + """ + with factory() as session: + return AgentChatOper(session).get(session_id, user_id) + + async def run(operation): + """ + 在隔离数据库测试中直接执行持久化服务提交的同步操作。 + """ + return operation() + + def transaction(operation): + """ + 每轮实际提交独立的临时数据库事务。 + """ + with factory() as session: + result = operation(session) + session.commit() + return result + + async def process(**kwargs): + """ + 通过正式输出回调提供本轮回复，触发展示历史的追加保存。 + """ + kwargs["output_callback"]("reply to " + kwargs["message"]) + + service = AgentChatService(SimpleNamespace(async_get=get)) + persistence = AgentChatPersistenceService(AgentChatOper, SimpleNamespace(run=run), transaction) + tasks = [] + + def create_task(coroutine): + """ + 记录后台任务句柄，确保两轮历史写入都完成后再检查数据库。 + """ + task = asyncio.create_task(coroutine) + tasks.append(task) + return task + + dependencies = WebAgentStreamDependencies(event_publisher_factory=application.WebAgentEventPublisher, + bind_user_session=lambda *_args: None, apply_display_event=application.apply_web_agent_display_event, + build_display_message=application.build_web_agent_display_message, + build_input_attachments=application.build_web_agent_input_attachments, + build_message_events_async=application.build_web_agent_message_events_async, + save_display_snapshot=application.save_web_agent_display_snapshot, + split_output=application.split_web_agent_output, create_background_task=create_task, + source="web-agent", heartbeat_seconds=1) + monkeypatch.setattr(application.agent_application, "get_running_agent_manager", lambda: SimpleNamespace(process_message=process)) + monkeypatch.setattr(application.agent_application, "get_web_agent_type", lambda: object) + user = SimpleNamespace(id=1, name="review", is_superuser=True) + try: + for text in ("first", "second"): + stream = build_agent_web_agent_stream(command=application.WebAgentStreamCommand(text=text, + session_id="review"), current_user=user, session_id="review", prompt=text, display_prompt=text, + has_audio_input=False, is_secret_confirmation_control=False, protected_transport_supported=False, + service=service, persistence=persistence, is_disconnected=AsyncMock(return_value=False), + dependencies=dependencies) + async for _event in stream: + pass + await asyncio.gather(*tasks) + tasks.clear() + record = await service.get("review") + assert [message["content"] for message in record.messages] == [ + "first", "reply to first", "second", "reply to second", + ] + finally: + await asyncio.gather(*tasks, return_exceptions=True) + engine.dispose() + + +@pytest.mark.asyncio +async def test_view_image_uses_read_permission_for_public_skill(tmp_path, monkeypatch): + """ + 普通用户读取公共技能图片沿用读取门禁，写入权限保持由基类决定。 + """ + from app.agent.tools.impl.view_image import ViewImageTool + + target = tmp_path / "skills/review/reference.png" + target.parent.mkdir(parents=True) + Image.new("RGB", (2, 2)).save(target) + monkeypatch.setattr(ViewImageTool, "_get_non_admin_local_file_roots", classmethod(lambda _cls: [tmp_path])) + tool = ViewImageTool(session_id="review", user_id="review") + tool.set_agent_context({"is_admin": False}) + result = json.loads(await tool.run(file_path=str(target))) + assert result["success"] is True + _path, write_error = await tool._check_local_file_access(str(target), operation="写入") + assert write_error is not None + + +@pytest.mark.asyncio +@pytest.mark.skipif(os.name != "posix", reason="此场景依赖POSIX独立进程组") +@pytest.mark.parametrize("retention_expired", [False, True]) +async def test_terminal_close_cleans_descendant_after_leader_exit(tmp_path, retention_expired): + """ + 仅测试所创建进程组：组长先退出，关闭owner仍结束存活后代。 + """ + import psutil + + from app.agent.terminal.manager import _TerminalSessionManager + from app.agent.terminal.ownership import TerminalScope, bind_terminal_scope + + manager = _TerminalSessionManager() + owner = TerminalScope("review", "review-descendant", "test") + pidfile = tmp_path / "child.pid" + child = "import time; time.sleep(30)" + parent = ("import subprocess,sys,pathlib; " + f"p=subprocess.Popen([sys.executable,'-c',{child!r}],stdin=subprocess.DEVNULL," + "stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL); " + f"pathlib.Path({str(pidfile)!r}).write_text(str(p.pid))") + pid = None + try: + with bind_terminal_scope(owner): + result = await manager.start(command=shlex.join([sys.executable, "-c", parent]), + use_pty=False, yield_time_ms=0, shell="/bin/sh", login=False) + session = manager._sessions[result["session_id"]] + await asyncio.wait_for(asyncio.shield(session.wait_task), 5) + pid = int(pidfile.read_text()) + assert os.getpgid(pid) == session.pid + if retention_expired: + session.updated_at = 0 + manager._cleanup_finished_sessions_locked() + assert session.session_id in manager._sessions + assert await manager.close_owner(owner) + async with asyncio.timeout(2): + while True: + try: + status = psutil.Process(pid).status() + except psutil.NoSuchProcess: + break + if status == psutil.STATUS_ZOMBIE: + break + await asyncio.sleep(0.02) + assert not manager._sessions + finally: + if pid is not None: + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + await manager.close() diff --git a/tests/test_agent_session_status.py b/tests/test_agent_session_status.py index 80f6c65b6..7b5d96064 100644 --- a/tests/test_agent_session_status.py +++ b/tests/test_agent_session_status.py @@ -75,7 +75,9 @@ def test_remote_session_status_sends_usage_summary() -> None: 远程会话状态回复包含用量、估算偏差和运行状态。 """ chain = MessageChain() - chain._user_sessions["10001"] = ("session-1", datetime.now()) + chain._user_sessions[chain._session_user_key("10001", NotificationChannel.Telegram, "telegram-test")] = ( + "session-1", datetime.now(), + ) status = { "session_id": "session-1", "model": "gpt-4o-mini", diff --git a/tests/test_agent_subagent_review_lifecycle.py b/tests/test_agent_subagent_review_lifecycle.py new file mode 100644 index 000000000..537e9b92e --- /dev/null +++ b/tests/test_agent_subagent_review_lifecycle.py @@ -0,0 +1,219 @@ +""" +通过已注册子任务工具验证关闭、完成和更新的真实异步生命周期。 + +仅替换外部模型提供端，任务创建、取消、完成回调和状态生成均使用正式实现。 +""" + +import asyncio +import json +from types import SimpleNamespace + +import pytest + +from app.agent.middleware import subagents + + +def _controller(provider): + """ + 构造正式控制器，把远端模型边界替换为事件驱动的本地协程。 + """ + controller = subagents.SubAgentTaskControlMiddleware(model=None, profiles=(), tools=[]) + controller._provider = SimpleNamespace(run_task=provider) + return controller + + +async def _call(controller, **arguments): + """ + 从正式注册的 StructuredTool 入口调用并读取公开 JSON 响应。 + """ + return json.loads(await controller.tools[0].ainvoke(arguments)) + + +@pytest.mark.asyncio +async def test_repeated_close_retains_owner_and_does_not_cancel_cleanup_twice(monkeypatch): + """ + 关闭超时保留真实任务，只取消一次，拒绝更新，收敛后可完成最终关闭。 + """ + monkeypatch.setattr(subagents, "SUBAGENT_CANCEL_GRACE_SECONDS", 0.01) + started, release = asyncio.Event(), asyncio.Event() + cancellations = [] + + async def provider(**_kwargs): + """ + 延迟取消清理，以观察重复关闭是否打断仍在途的同一任务。 + """ + started.set() + while not release.is_set(): + try: + await release.wait() + except asyncio.CancelledError: + cancellations.append("cancel") + return "清理完成" + + controller = _controller(provider) + try: + started_result = await _call(controller, action="start", description="持有清理任务") + task_id = started_result["tasks"][0]["task_id"] + await asyncio.wait_for(started.wait(), 1) + assert await controller.close() is False + status = await _call(controller, action="status", task_id=task_id) + assert status["tasks"][0]["status"] == "running" + assert status["tasks"][0]["execution_outcome"] == "pending" + updated = await _call(controller, action="update", task_id=task_id, description="不得替换") + assert updated["success"] is False + assert "关闭" in updated["error"] + assert updated["tasks"][0]["description"] == "持有清理任务" + assert await controller.close() is False + assert cancellations == ["cancel"] + release.set() + finished = await _call(controller, action="wait", task_id=task_id, timeout_ms=1000) + assert finished["tasks"][0]["result"] == "清理完成" + # 已完成任务的wait可立即返回，完成回调在下一轮事件循环后发布结束时间。 + await asyncio.sleep(0) + status = await _call(controller, action="status", task_id=task_id) + assert status["tasks"][0]["finished_at"] is not None + assert await controller.close() is True + assert (await _call(controller, action="status"))["tasks"] == [] + finally: + release.set() + await controller.close() + + +@pytest.mark.asyncio +async def test_update_timeout_keeps_original_task_and_description(monkeypatch): + """ + 旧任务尚未完成取消时公开更新返回失败，不改描述或产生第二个执行者。 + """ + monkeypatch.setattr(subagents, "SUBAGENT_CANCEL_GRACE_SECONDS", 0.01) + started, release = asyncio.Event(), asyncio.Event() + descriptions = [] + + async def provider(*, description, **_kwargs): + """ + 保留取消中的旧执行者，直到测试显式允许其清理完成。 + """ + descriptions.append(description) + started.set() + while not release.is_set(): + try: + await release.wait() + except asyncio.CancelledError: + continue + return description + + controller = _controller(provider) + try: + started_result = await _call(controller, action="start", description="原始任务") + task_id = started_result["tasks"][0]["task_id"] + await asyncio.wait_for(started.wait(), 1) + updated = await _call(controller, action="update", task_id=task_id, description="替代任务") + assert updated["success"] is False + assert "取消窗口" in updated["error"] + assert updated["execution_outcome"] == "failed" + assert updated["tasks"][0]["description"] == "原始任务" + assert updated["tasks"][0]["task_id"] == task_id + assert descriptions == ["原始任务"] + release.set() + finished = await _call(controller, action="wait", task_id=task_id, timeout_ms=1000) + assert finished["tasks"][0]["result"] == "原始任务" + finally: + release.set() + await controller.close() + + +@pytest.mark.asyncio +async def test_failed_task_remains_observable_until_controller_close(): + """ + 提供端执行失败经公开等待和查询返回失败证据，完成后关闭仍能收口。 + """ + async def provider(**_kwargs): + """ + 在模型边界报告确定失败，使用真实任务传播和完成回调。 + """ + raise LookupError("测试模型执行失败") + + controller = _controller(provider) + try: + completed = await _call(controller, action="run", description="失败任务", timeout_ms=1000) + task = completed["tasks"][0] + assert task["status"] == "failed" + assert task["execution_outcome"] == "failed" + assert "测试模型执行失败" in task["error"] + assert task["finished_at"] is not None + observed = await _call(controller, action="wait", task_id=task["task_id"], timeout_ms=1000) + assert observed["tasks"] == completed["tasks"] + assert await controller.close() is True + assert (await _call(controller, action="status"))["tasks"] == [] + finally: + await controller.close() + + +@pytest.mark.asyncio +async def test_completed_task_can_update_same_id_and_publish_new_completion(): + """ + 完成后的任务可复用ID更新，旧完成时间不会冒充新任务完成状态。 + """ + replacement_started, release = asyncio.Event(), asyncio.Event() + descriptions = [] + + async def provider(*, description, **_kwargs): + """ + 首轮立即完成，替代任务通过事件固定在运行中便于查询。 + """ + descriptions.append(description) + if description == "替代任务": + replacement_started.set() + await release.wait() + return f"{description}结果" + + controller = _controller(provider) + try: + completed = await _call(controller, action="run", description="原始任务", timeout_ms=1000) + first = completed["tasks"][0] + assert first["status"] == "completed" and first["finished_at"] is not None + updated = await _call(controller, action="update", task_id=first["task_id"], description="替代任务") + assert updated["success"] is True + assert updated["task_id"] == first["task_id"] + await asyncio.wait_for(replacement_started.wait(), 1) + status = await _call(controller, action="status", task_id=first["task_id"]) + assert status["tasks"][0]["status"] == "running" + assert status["tasks"][0]["finished_at"] is None + release.set() + finished = await _call(controller, action="wait", task_id=first["task_id"], timeout_ms=1000) + assert finished["tasks"][0]["status"] == "completed" + assert finished["tasks"][0]["result"] == "替代任务结果" + await asyncio.sleep(0) + status = await _call(controller, action="status", task_id=first["task_id"]) + assert status["tasks"][0]["finished_at"] is not None + assert descriptions == ["原始任务", "替代任务"] + finally: + release.set() + await controller.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("arguments,expected_error", [ + ({"description": "新任务"}, "必须提供"), + ({"task_id": "missing", "description": " "}, "必须提供"), + ({"task_id": "missing", "description": "新任务"}, "未知子代理任务"), +]) +async def test_invalid_public_update_returns_recovery_without_starting(arguments, expected_error): + """ + 缺少更新身份、空描述和未知任务均返回恢复指引，模型边界不得被调用。 + """ + async def provider(**_kwargs): + """ + 无效更新若启动了执行者即报告测试失败。 + """ + raise AssertionError("无效更新不得启动子任务") + + controller = _controller(provider) + try: + response = await _call(controller, action="update", **arguments) + assert response["success"] is False + assert response["execution_outcome"] == "failed" + assert expected_error in response["error"] + assert response["recovery"] + assert (await _call(controller, action="status"))["tasks"] == [] + finally: + await controller.close() diff --git a/tests/test_agent_terminal_ownership.py b/tests/test_agent_terminal_ownership.py index 0d526c15e..e80345b4d 100644 --- a/tests/test_agent_terminal_ownership.py +++ b/tests/test_agent_terminal_ownership.py @@ -5,6 +5,7 @@ import asyncio import os import shlex +import signal import subprocess import sys from collections.abc import AsyncIterator @@ -285,6 +286,73 @@ async def test_close_nonconvergence_preserves_record_for_retry(manager: _Termina assert session.process.returncode is not None and handle not in manager._sessions +@pytest.mark.asyncio +@pytest.mark.skipif(os.name != "posix", reason="此场景依赖 POSIX 进程组信号") +@pytest.mark.parametrize("persistent", [False, True]) +async def test_close_owner_retries_group_exit_permission_race( + manager: _TerminalSessionManager, monkeypatch: pytest.MonkeyPatch, persistent: bool, +) -> None: + """ + 真实进程结束后的临时拒绝允许重试，持续拒绝则保留 owner 直到下一次成功收尾。 + """ + owner = _scope() + handle = (await _start(manager, owner))["session_id"] + await _finish(manager, owner, handle) + session = manager._sessions[handle] + attempts = [] + original = os.killpg + + def denied_while_exiting(pid: int, sig: int) -> None: + """ + 仅在本例的最终 KILL 上模拟内核退出窗口，不影响真实进程和其他信号。 + """ + if pid == session.pid and sig == signal.SIGKILL and (persistent or not attempts): + attempts.append(sig) + raise PermissionError("process group is exiting") + original(pid, sig) + + with monkeypatch.context() as patch: + patch.setattr(os, "killpg", denied_while_exiting) + patch.setattr(terminal, "TERMINAL_KILL_GRACE_SECONDS", 0.01) + assert await manager.close_owner(owner) is not persistent + assert attempts + if persistent: + assert manager._sessions[handle] is session and session.owner is owner + assert await manager.close_owner(owner) + assert handle not in manager._sessions + + +@pytest.mark.asyncio +@pytest.mark.skipif(os.name != "posix", reason="此场景依赖 POSIX 进程组探测") +async def test_retention_preserves_group_when_existence_probe_is_denied( + manager: _TerminalSessionManager, monkeypatch: pytest.MonkeyPatch, +) -> None: + """ + 过期清理无法确认进程组消失时保留唯一归属记录，供关闭流程继续回收。 + """ + owner = _scope() + handle = (await _start(manager, owner))["session_id"] + await _finish(manager, owner, handle) + session = manager._sessions[handle] + session.updated_at = 0 + original = os.killpg + + def denied_probe(pid: int, sig: int) -> None: + """ + 仅拒绝本例进程组的存在性探测。 + """ + if pid == session.pid and sig == 0: + raise PermissionError("process group is exiting") + original(pid, sig) + + with monkeypatch.context() as patch: + patch.setattr(os, "killpg", denied_probe) + manager._cleanup_finished_sessions_locked() + assert manager._sessions[handle] is session + assert await manager.close_owner(owner) + assert handle not in manager._sessions + + @pytest.mark.asyncio async def test_scope_cleanup_is_lazy_and_new_manager_rejects_old_handle(manager: _TerminalSessionManager, monkeypatch: pytest.MonkeyPatch) -> None: """ diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index 41f80634d..9de6ef2a1 100644 --- a/tests/test_builtin_skill_boundaries.py +++ b/tests/test_builtin_skill_boundaries.py @@ -130,7 +130,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None: "database-operation": "14", "feedback-issue": "12", "anysearch": "3", - "moviepilot-api": "62", + "moviepilot-api": "63", "moviepilot-update": "7", "organize-files": "6", "transfer-failed-retry": "6", diff --git a/tests/test_llm_provider_transport.py b/tests/test_llm_provider_transport.py index 43daa1880..5c66c42c8 100644 --- a/tests/test_llm_provider_transport.py +++ b/tests/test_llm_provider_transport.py @@ -99,7 +99,8 @@ def test_google_model_client_uses_adapter_args_and_closes_both_clients( display_name="Gemini Test", supported_actions=["generateContent"], ) - ] + ], + config={}, ) class FakeAsyncClient: diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 80b44c681..a6b75f7d6 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -1533,7 +1533,9 @@ def test_web_agent_stream_binds_session_to_agent_manager(): body = asyncio.run(scenario()) assert "状态正常" in body - assert MessageChain._user_sessions["1"][0] == session_id + assert MessageChain._user_sessions[MessageChain._session_user_key( + "1", NotificationChannel.WebAgent, "web-agent", + )][0] == session_id assert isinstance(agent_manager.active_agents[session_id], FakeWebAgent) finally: MessageChain._user_sessions.clear()