fix(update): use managed progress execution

This commit is contained in:
jxxghp
2026-09-05 22:28:15 +08:00
parent 76174b11d6
commit b432354592
7 changed files with 53 additions and 17 deletions
+2 -1
View File
@@ -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.site import site_interaction_manager
from app.application.messaging.skill import skill_interaction_manager from app.application.messaging.skill import skill_interaction_manager
from app.application.messaging.subscribe import subscribe_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) @dataclass(frozen=True, slots=True)
@@ -158,6 +157,8 @@ class InteractionRouter:
def has_pending_interaction(user_id: Union[str, int]) -> bool: def has_pending_interaction(user_id: Union[str, int]) -> bool:
"""供 WebAgent 判断用户是否处于传统交互会话。""" """供 WebAgent 判断用户是否处于传统交互会话。"""
from app.application.messaging.update import update_interaction_manager
return any( return any(
manager.get_by_user(user_id) is not None manager.get_by_user(user_id) is not None
for manager in ( for manager in (
+11 -6
View File
@@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Callable, Coroutine from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass from dataclasses import dataclass
from threading import Lock from threading import Lock
from typing import Any, Optional, Protocol, Union from typing import Any, Optional, Protocol, Union
@@ -49,6 +49,7 @@ class SystemUpdateInteractionActions(Protocol):
UpdateMonitorSubmitter = Callable[[Coroutine[Any, Any, None]], Any] UpdateMonitorSubmitter = Callable[[Coroutine[Any, Any, None]], Any]
UpdateOperationRunner = Callable[..., Awaitable[Any]]
RestartMarker = Callable[[NotificationChannel, Union[str, int], Optional[str]], None] RestartMarker = Callable[[NotificationChannel, Union[str, int], Optional[str]], None]
RestartMarkerClearer = Callable[[], None] RestartMarkerClearer = Callable[[], None]
@@ -73,6 +74,7 @@ class SystemUpdateInteractionHandler:
messenger: MessageGateway, messenger: MessageGateway,
actions: SystemUpdateInteractionActions, actions: SystemUpdateInteractionActions,
submit_monitor: UpdateMonitorSubmitter, submit_monitor: UpdateMonitorSubmitter,
run_sync: UpdateOperationRunner,
mark_restart: RestartMarker, mark_restart: RestartMarker,
clear_restart_marker: RestartMarkerClearer, clear_restart_marker: RestartMarkerClearer,
poll_interval_seconds: float = _poll_interval_seconds, poll_interval_seconds: float = _poll_interval_seconds,
@@ -85,6 +87,7 @@ class SystemUpdateInteractionHandler:
self._renderer = _SystemUpdateRenderer(messenger=messenger, actions=actions) self._renderer = _SystemUpdateRenderer(messenger=messenger, actions=actions)
self._progress_monitor = _SystemUpdateProgressMonitor( self._progress_monitor = _SystemUpdateProgressMonitor(
actions=actions, renderer=self._renderer, submit_monitor=submit_monitor, actions=actions, renderer=self._renderer, submit_monitor=submit_monitor,
run_sync=run_sync,
poll_interval_seconds=poll_interval_seconds, poll_interval_seconds=poll_interval_seconds,
) )
@@ -505,12 +508,14 @@ class _SystemUpdateProgressMonitor:
def __init__( def __init__(
self, *, actions: SystemUpdateInteractionActions, self, *, actions: SystemUpdateInteractionActions,
renderer: _SystemUpdateRenderer, submit_monitor: UpdateMonitorSubmitter, renderer: _SystemUpdateRenderer, submit_monitor: UpdateMonitorSubmitter,
run_sync: UpdateOperationRunner,
poll_interval_seconds: float, poll_interval_seconds: float,
) -> None: ) -> None:
"""注入状态读取、消息渲染和后台任务提交能力。""" """注入状态读取、消息渲染和后台任务提交能力。"""
self._actions = actions self._actions = actions
self._renderer = renderer self._renderer = renderer
self._submit_monitor = submit_monitor self._submit_monitor = submit_monitor
self._run_sync = run_sync
self._poll_interval_seconds = max(0.0, poll_interval_seconds) self._poll_interval_seconds = max(0.0, poll_interval_seconds)
def schedule( def schedule(
@@ -567,7 +572,7 @@ class _SystemUpdateProgressMonitor:
request = update_interaction_manager.get_by_id(request_id, userid) request = update_interaction_manager.get_by_id(request_id, userid)
if request is None: if request is None:
return 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) item = self._renderer.application_item(status)
fingerprint = self._renderer.item_fingerprint(item) fingerprint = self._renderer.item_fingerprint(item)
if fingerprint != last_fingerprint: if fingerprint != last_fingerprint:
@@ -578,7 +583,7 @@ class _SystemUpdateProgressMonitor:
channel=channel, channel=channel,
) )
if original_message_id and original_chat_id and ChannelCapabilityManager.supports_editing(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, self._renderer.edit_view,
view=view, view=view,
channel=channel, channel=channel,
@@ -588,7 +593,7 @@ class _SystemUpdateProgressMonitor:
original_chat_id=original_chat_id, original_chat_id=original_chat_id,
) )
if not edited and (not edit_fallback_sent or item.state in self._terminal_download_states): 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, self._renderer.post_view,
view=view, view=view,
channel=channel, channel=channel,
@@ -602,7 +607,7 @@ class _SystemUpdateProgressMonitor:
else: else:
progress_bucket = item.progress // 10 progress_bucket = item.progress // 10
if progress_bucket != last_progress_bucket or item.state in self._terminal_download_states: 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, self._renderer.post_view,
view=view, view=view,
channel=channel, channel=channel,
@@ -621,7 +626,7 @@ class _SystemUpdateProgressMonitor:
logger.warning(f"监视 MoviePilot 更新下载进度失败:{error}") logger.warning(f"监视 MoviePilot 更新下载进度失败:{error}")
request = update_interaction_manager.get_by_id(request_id, userid) request = update_interaction_manager.get_by_id(request_id, userid)
if request is not None: if request is not None:
await asyncio.to_thread( await self._run_sync(
self._renderer.render_operation_failure, self._renderer.render_operation_failure,
request=request, request=request,
channel=channel, channel=channel,
+2 -1
View File
@@ -27,7 +27,6 @@ from app.application.messaging.session import MessageSessionService
from app.application.messaging.site import site_interaction_manager from app.application.messaging.site import site_interaction_manager
from app.application.messaging.skill import SkillInteractionHandler, skill_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.subscribe import subscribe_interaction_manager
from app.application.messaging.update import update_interaction_manager
from app.chain.base import ChainBase from app.chain.base import ChainBase
from app.chain.interaction import MediaInteractionChain as _MediaInteractionChain from app.chain.interaction import MediaInteractionChain as _MediaInteractionChain
from app.chain.site import SiteChain from app.chain.site import SiteChain
@@ -706,6 +705,8 @@ class MessageChain(ChainBase):
def _interaction_router(self) -> interaction_router.InteractionRouter: def _interaction_router(self) -> interaction_router.InteractionRouter:
"""构造交互路由器,文本会话按创建时间选择,回调路由注册顺序即优先级。""" """构造交互路由器,文本会话按创建时间选择,回调路由注册顺序即优先级。"""
from app.application.messaging.update import update_interaction_manager
session_routes = [ session_routes = [
interaction_router.SessionRoute( interaction_router.SessionRoute(
name="sites", name="sites",
+26 -4
View File
@@ -1,15 +1,15 @@
import asyncio
import errno import errno
import json import json
import re import re
import shutil import shutil
import threading import threading
import uuid import uuid
from collections.abc import Coroutine, Mapping from collections.abc import Callable, Coroutine, Mapping
from pathlib import Path 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.configuration import get_chain_runtime_config_snapshot
from app.application.messaging.update import SystemUpdateInteractionHandler
from app.chain.base import ChainBase from app.chain.base import ChainBase
from app.runtime import version as runtime_version from app.runtime import version as runtime_version
from app.runtime.log import logger 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.message import Message
from app.schemas.notification import NotificationChannel from app.schemas.notification import NotificationChannel
if TYPE_CHECKING:
from app.application.messaging.update import SystemUpdateInteractionHandler
class SystemResponsePort(Protocol): class SystemResponsePort(Protocol):
"""系统链查询发布版本所需的最小同步 HTTP 响应契约。""" """系统链查询发布版本所需的最小同步 HTTP 响应契约。"""
@@ -102,19 +105,38 @@ def _close_system_response(response: SystemResponsePort) -> None:
logger.debug(f"释放版本响应失败:{str(err)}") 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): class _SystemUpdateChain(ChainBase):
"""提供通知渠道主程序升级交互的 Chain 入口。""" """提供通知渠道主程序升级交互的 Chain 入口。"""
_update_restart_file = "__system_update_restart__" _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: if self.system_service is None:
raise RuntimeError("系统更新服务尚未由启动组合根装配") raise RuntimeError("系统更新服务尚未由启动组合根装配")
return SystemUpdateInteractionHandler( return SystemUpdateInteractionHandler(
messenger=self, messenger=self,
actions=self.system_service, actions=self.system_service,
submit_monitor=self._submit_update_monitor, submit_monitor=self._submit_update_monitor,
run_sync=_run_update_operation,
mark_restart=self._mark_update_restart, mark_restart=self._mark_update_restart,
clear_restart_marker=self._clear_update_restart_marker, clear_restart_marker=self._clear_update_restart_marker,
) )
+1 -1
View File
@@ -103,7 +103,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| Python 源码量 | 305,884 行 | 排除 `app/plugins/**`61 个文件超过 1,000 行,11 个超过 2,000 行 | | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`61 个文件超过 1,000 行,11 个超过 2,000 行 |
| 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`;65 个超过 150 行,21 个超过 250 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`;65 个超过 150 行,21 个超过 250 行 |
| 全量 mypy 历史债务 | 9,508 / 513 文件 | Agent API 重构后的现状基线;canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 | | 全量 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 未进入包级覆盖率门禁 | | 覆盖率固定基线 | Application 80.00%Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
### 3.3 热点文件 ### 3.3 热点文件
-3
View File
@@ -364,9 +364,6 @@
"F541": 1, "F541": 1,
"I001": 1 "I001": 1
}, },
"app/modules/slack/slack.py": {
"I001": 1
},
"app/modules/subtitle/__init__.py": { "app/modules/subtitle/__init__.py": {
"I001": 1 "I001": 1
}, },
+11 -1
View File
@@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Coroutine from collections.abc import Callable, Coroutine
from typing import Any from typing import Any
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
@@ -145,12 +145,22 @@ def _handler(
messenger=messenger, messenger=messenger,
actions=actions, actions=actions,
submit_monitor=submitted.append, submit_monitor=submitted.append,
run_sync=_run_sync,
mark_restart=mark_restart or Mock(), mark_restart=mark_restart or Mock(),
clear_restart_marker=clear_restart_marker or Mock(), clear_restart_marker=clear_restart_marker or Mock(),
poll_interval_seconds=0, 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: def test_update_command_prompts_for_download_when_release_is_available() -> None:
"""检测到新版本后应显示版本信息和确认升级按钮。""" """检测到新版本后应显示版本信息和确认升级按钮。"""
messenger = _Messenger() messenger = _Messenger()