refactor: declare stop signal and plugin settlement in lifecycle manifest

app/startup/lifecycle/__init__.py

@@ -309,6 +309,13 @@ def prepare_plugin_restore() -> None:
     SystemChain().restore_plugins()
 
 
+def schedule_plugin_settlement(app: FastAPI) -> None:
+    """调度插件同步与启动收尾任务，并把取消权交给最前置的 TaskRegistry owner。"""
+    task = asyncio.create_task(run_startup_step("插件同步与启动收尾", init_extra))
+    task_registry = app.state.task_registry
+    task_registry.register(task, owner="startup.plugin_settlement")
+
+
 def prepare_database_component(app: FastAPI) -> None:
     """完成数据库建表、迁移与 head 校验后发布数据库就绪状态。"""
     # Alembic 及全部 ORM 元数据只在 lifespan 真正启动时加载，create_app/import 阶段
@@ -531,6 +538,22 @@ def build_lifecycle_components(app: FastAPI) -> tuple[LifecycleComponent, ...]:
             stop_order=10,
             stop_timeout_seconds=300,
         ),
+        # 停止信号必须先于一切资源释放发出，让工作流、整理等长任务尽早感知停机。
+        LifecycleComponent(
+            name="停止信号",
+            stop=global_vars.stop_system,
+            stop_order=4,
+            stop_timeout_seconds=10,
+        ),
+        # 插件同步与启动收尾是最后一个启动阶段；任务本身由 TaskRegistry 在关停预算内取消等待。
+        LifecycleComponent(
+            name="插件同步与启动收尾",
+            dependencies=("工作流",),
+            start=lambda: schedule_plugin_settlement(app),
+            start_order=150,
+            start_timeout_seconds=30,
+            start_failure=LifecycleFailurePolicy.FAIL_FAST,
+        ),
     )
 
 
@@ -585,12 +608,6 @@ async def lifespan(app: FastAPI):
             active_start_component = None
         if settings.MOVIEPILOT_SAFE_MODE:
             print("MoviePilot safe mode enabled: skip plugins, scheduler, monitor, commands and workflow.")
-        # 插件同步到本地
-        sync_plugins_task = asyncio.create_task(
-            run_startup_step("插件同步与启动收尾", init_extra)
-        )
-        task_registry = app.state.task_registry
-        task_registry.register(sync_plugins_task, owner="startup.plugin_settlement")
         health.mark_ready()
     except BaseException:
         # Uvicorn 在 lifespan 抛错时不会开始接流量；状态仍需供嵌入式入口和测试诊断。
@@ -614,10 +631,10 @@ async def lifespan(app: FastAPI):
     finally:
         health.mark_stopping()
         print("Shutting down...")
-        global_vars.stop_system()
         try:
             # 插件 settlement 已登记到最前置 TaskRegistry。由该 FAIL_FAST owner
             # 在统一预算内取消/等待，不能在屏障之前无界 await 绕过停机预算。
+            # 停止信号与各资源 owner 的释放顺序由组件清单声明。
             await stop_lifecycle_components(enabled_components)
         finally:
             try:

docs/refactor/backend-architecture-review.md

@@ -89,6 +89,14 @@ chain 层零 `app.db` / `app.modules` 内部直连，domain 与 chain 层配置
 3. 停止标志设置；
 4. 日志关闭靠注释约定顺序。
 
+> 处理进展（2026-08-24）：第 2、3 项已组件化——"停止信号"（stop_order=4，先于一切
+> 资源释放发出停机通知，启动失败清理同样生效）与"插件同步与启动收尾"（start_order=150，
+> 依赖工作流，取消权仍归最前置 TaskRegistry owner）均已进入声明式清单并通过顺序快照测试。
+> 第 1、4 项经复核**不能直接进清单**：当前引擎的 FAIL_FAST break 会跳过更高 stop_order
+> 的组件，而主循环清除与日志关闭在现有嵌套 finally 中是无条件执行的"最外层保底"，
+> 直接搬移会让无关 owner 未收敛时跳过这两步（回归）。如需收口，须先做显式引擎决策：
+> 引入"最终保底 finalizer"概念或调整 FAIL_FAST 传播语义。
+
 另有两个组合根脆弱点：
 
 * `initializers/command.py`、`initializers/scheduler.py`、`initializers/agent.py`

tests/test_lifecycle_shutdown.py

@@ -553,8 +553,10 @@ def test_lifecycle_manifest_declares_normal_and_safe_mode_order() -> None:
         "待处理整理回放",
         "命令服务",
         "工作流",
+        "插件同步与启动收尾",
     ]
     assert normal_stop == [
+        "停止信号",
         "后台任务登记器",
         "插件变更监控",
         "插件备份",
@@ -583,6 +585,8 @@ def test_lifecycle_manifest_declares_normal_and_safe_mode_order() -> None:
         "AI智能体会话",
         "整理后台服务",
         "事件投递屏障",
+        "停止信号",
+        "插件同步与启动收尾",
     }
     assert all(item["start_failure"] == "fail_fast" for item in normal)
     assert {
@@ -840,7 +844,9 @@ def test_lifespan_cleans_started_owners_after_late_startup_failure(monkeypatch):
         asyncio.run(run_lifespan())
 
     assert raised.value is startup_error
-    lifecycle.global_vars.stop_system.assert_not_called()
+    # 停止信号是无依赖的 stop-only owner：启动失败清理同样要先发出停机通知，
+    # 让仍在运行的后台任务尽早感知进程即将退出。
+    lifecycle.global_vars.stop_system.assert_called_once_with()
     for name in (
         "stop_plugin_monitor",
         "backup_plugins",