diff --git a/app/application/messaging/router.py b/app/application/messaging/router.py index d03dcfa09..83643f6e8 100644 --- a/app/application/messaging/router.py +++ b/app/application/messaging/router.py @@ -16,7 +16,6 @@ from app.application.messaging.media import media_interaction_manager from app.application.messaging.site import site_interaction_manager from app.application.messaging.skill import skill_interaction_manager from app.application.messaging.subscribe import subscribe_interaction_manager -from app.application.messaging.update import update_interaction_manager @dataclass(frozen=True, slots=True) @@ -158,6 +157,8 @@ class InteractionRouter: def has_pending_interaction(user_id: Union[str, int]) -> bool: """供 WebAgent 判断用户是否处于传统交互会话。""" + from app.application.messaging.update import update_interaction_manager + return any( manager.get_by_user(user_id) is not None for manager in ( diff --git a/app/application/messaging/update.py b/app/application/messaging/update.py index 0c7670029..fb637782c 100644 --- a/app/application/messaging/update.py +++ b/app/application/messaging/update.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio -from collections.abc import Callable, Coroutine +from collections.abc import Awaitable, Callable, Coroutine from dataclasses import dataclass from threading import Lock from typing import Any, Optional, Protocol, Union @@ -49,6 +49,7 @@ class SystemUpdateInteractionActions(Protocol): UpdateMonitorSubmitter = Callable[[Coroutine[Any, Any, None]], Any] +UpdateOperationRunner = Callable[..., Awaitable[Any]] RestartMarker = Callable[[NotificationChannel, Union[str, int], Optional[str]], None] RestartMarkerClearer = Callable[[], None] @@ -73,6 +74,7 @@ class SystemUpdateInteractionHandler: messenger: MessageGateway, actions: SystemUpdateInteractionActions, submit_monitor: UpdateMonitorSubmitter, + run_sync: UpdateOperationRunner, mark_restart: RestartMarker, clear_restart_marker: RestartMarkerClearer, poll_interval_seconds: float = _poll_interval_seconds, @@ -85,6 +87,7 @@ class SystemUpdateInteractionHandler: self._renderer = _SystemUpdateRenderer(messenger=messenger, actions=actions) self._progress_monitor = _SystemUpdateProgressMonitor( actions=actions, renderer=self._renderer, submit_monitor=submit_monitor, + run_sync=run_sync, poll_interval_seconds=poll_interval_seconds, ) @@ -505,12 +508,14 @@ class _SystemUpdateProgressMonitor: def __init__( self, *, actions: SystemUpdateInteractionActions, renderer: _SystemUpdateRenderer, submit_monitor: UpdateMonitorSubmitter, + run_sync: UpdateOperationRunner, poll_interval_seconds: float, ) -> None: """注入状态读取、消息渲染和后台任务提交能力。""" self._actions = actions self._renderer = renderer self._submit_monitor = submit_monitor + self._run_sync = run_sync self._poll_interval_seconds = max(0.0, poll_interval_seconds) def schedule( @@ -567,7 +572,7 @@ class _SystemUpdateProgressMonitor: request = update_interaction_manager.get_by_id(request_id, userid) if request is None: return - status = await asyncio.to_thread(self._actions.update_status) + status = await self._run_sync(self._actions.update_status) item = self._renderer.application_item(status) fingerprint = self._renderer.item_fingerprint(item) if fingerprint != last_fingerprint: @@ -578,7 +583,7 @@ class _SystemUpdateProgressMonitor: channel=channel, ) if original_message_id and original_chat_id and ChannelCapabilityManager.supports_editing(channel): - edited = await asyncio.to_thread( + edited = await self._run_sync( self._renderer.edit_view, view=view, channel=channel, @@ -588,7 +593,7 @@ class _SystemUpdateProgressMonitor: original_chat_id=original_chat_id, ) if not edited and (not edit_fallback_sent or item.state in self._terminal_download_states): - await asyncio.to_thread( + await self._run_sync( self._renderer.post_view, view=view, channel=channel, @@ -602,7 +607,7 @@ class _SystemUpdateProgressMonitor: else: progress_bucket = item.progress // 10 if progress_bucket != last_progress_bucket or item.state in self._terminal_download_states: - await asyncio.to_thread( + await self._run_sync( self._renderer.post_view, view=view, channel=channel, @@ -621,7 +626,7 @@ class _SystemUpdateProgressMonitor: logger.warning(f"监视 MoviePilot 更新下载进度失败：{error}") request = update_interaction_manager.get_by_id(request_id, userid) if request is not None: - await asyncio.to_thread( + await self._run_sync( self._renderer.render_operation_failure, request=request, channel=channel, diff --git a/app/chain/message.py b/app/chain/message.py index eac479eba..f444006da 100644 --- a/app/chain/message.py +++ b/app/chain/message.py @@ -27,7 +27,6 @@ from app.application.messaging.session import MessageSessionService from app.application.messaging.site import site_interaction_manager from app.application.messaging.skill import SkillInteractionHandler, skill_interaction_manager from app.application.messaging.subscribe import subscribe_interaction_manager -from app.application.messaging.update import update_interaction_manager from app.chain.base import ChainBase from app.chain.interaction import MediaInteractionChain as _MediaInteractionChain from app.chain.site import SiteChain @@ -706,6 +705,8 @@ class MessageChain(ChainBase): def _interaction_router(self) -> interaction_router.InteractionRouter: """构造交互路由器，文本会话按创建时间选择，回调路由注册顺序即优先级。""" + from app.application.messaging.update import update_interaction_manager + session_routes = [ interaction_router.SessionRoute( name="sites", diff --git a/app/chain/system.py b/app/chain/system.py index 0eba6acff..0681440eb 100644 --- a/app/chain/system.py +++ b/app/chain/system.py @@ -1,15 +1,15 @@ +import asyncio import errno import json import re import shutil import threading import uuid -from collections.abc import Coroutine, Mapping +from collections.abc import Callable, Coroutine, Mapping from pathlib import Path -from typing import Any, Optional, Protocol, Union +from typing import TYPE_CHECKING, Any, Optional, Protocol, Union from app.application.configuration import get_chain_runtime_config_snapshot -from app.application.messaging.update import SystemUpdateInteractionHandler from app.chain.base import ChainBase from app.runtime import version as runtime_version from app.runtime.log import logger @@ -19,6 +19,9 @@ from app.runtime.tasks import get_task_registry from app.schemas.message import Message from app.schemas.notification import NotificationChannel +if TYPE_CHECKING: + from app.application.messaging.update import SystemUpdateInteractionHandler + class SystemResponsePort(Protocol): """系统链查询发布版本所需的最小同步 HTTP 响应契约。""" @@ -102,19 +105,38 @@ def _close_system_response(response: SystemResponsePort) -> None: logger.debug(f"释放版本响应失败：{str(err)}") +async def _run_update_operation( + function: Callable[..., Any], + /, + *args: Any, + **kwargs: Any, +) -> Any: + """通过任务登记器在线程池执行同步更新操作并等待真实终态。""" + task = get_task_registry().create_sync( + function, + *args, + owner="chain.system.update_progress.operation", + **kwargs, + ) + return await asyncio.shield(task) + + class _SystemUpdateChain(ChainBase): """提供通知渠道主程序升级交互的 Chain 入口。""" _update_restart_file = "__system_update_restart__" - def _update_interaction_handler(self) -> SystemUpdateInteractionHandler: + def _update_interaction_handler(self) -> "SystemUpdateInteractionHandler": """构造复用当前消息网关和系统应用服务的更新交互控制器。""" + from app.application.messaging.update import SystemUpdateInteractionHandler + if self.system_service is None: raise RuntimeError("系统更新服务尚未由启动组合根装配") return SystemUpdateInteractionHandler( messenger=self, actions=self.system_service, submit_monitor=self._submit_update_monitor, + run_sync=_run_update_operation, mark_restart=self._mark_update_restart, clear_restart_marker=self._clear_update_restart_marker, ) diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 167b0b739..3108e9f4e 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -103,7 +103,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 | | 全量 mypy 历史债务 | 9,508 / 513 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | -| Ruff 历史诊断 | 547 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | +| Ruff 历史诊断 | 546 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | ### 3.3 热点文件 diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index 8ecaadab4..2f951aac4 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -364,9 +364,6 @@ "F541": 1, "I001": 1 }, - "app/modules/slack/slack.py": { - "I001": 1 - }, "app/modules/subtitle/__init__.py": { "I001": 1 }, diff --git a/tests/test_system_update_interaction.py b/tests/test_system_update_interaction.py index c8778ba21..95a293770 100644 --- a/tests/test_system_update_interaction.py +++ b/tests/test_system_update_interaction.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio -from collections.abc import Coroutine +from collections.abc import Callable, Coroutine from typing import Any from unittest.mock import Mock, patch @@ -145,12 +145,22 @@ def _handler( messenger=messenger, actions=actions, submit_monitor=submitted.append, + run_sync=_run_sync, mark_restart=mark_restart or Mock(), clear_restart_marker=clear_restart_marker or Mock(), poll_interval_seconds=0, ) +async def _run_sync( + function: Callable[..., Any], + *args: Any, + **kwargs: Any, +) -> Any: + """在测试事件循环内直接执行同步更新操作。""" + return function(*args, **kwargs) + + def test_update_command_prompts_for_download_when_release_is_available() -> None: """检测到新版本后应显示版本信息和确认升级按钮。""" messenger = _Messenger()