fix(plugin): close cancellable install lifecycle gaps

app/adapters/system/host.py

@@ -188,9 +188,11 @@ class SystemUtils:
             grace_seconds: float = 5,
     ) -> None:
         """终止安装子进程及其同组子进程，并确保句柄已回收。"""
+        process_tree = SystemUtils._process_tree(process.pid)
         if process.returncode is None:
             try:
                 if os.name == "nt":
+                    SystemUtils._signal_process_tree(process_tree)
                     process.terminate()
                 else:
                     os.killpg(process.pid, signal.SIGTERM)
@@ -205,9 +207,11 @@ class SystemUtils:
         except (asyncio.TimeoutError, asyncio.CancelledError):
             try:
                 if os.name == "nt":
+                    SystemUtils._signal_process_tree(process_tree, force=True)
                     process.kill()
                 else:
                     os.killpg(process.pid, signal.SIGKILL)
+                    SystemUtils._signal_process_tree(process_tree, force=True)
             except (ProcessLookupError, OSError):
                 pass
             try:
@@ -224,6 +228,27 @@ class SystemUtils:
             except (ProcessLookupError, OSError):
                 pass
 
+    @staticmethod
+    def _process_tree(pid: int) -> list[psutil.Process]:
+        """收集安装进程及其后代，避免构建进程继续写运行环境。"""
+        try:
+            parent = psutil.Process(pid)
+            return [parent, *parent.children(recursive=True)]
+        except (psutil.Error, OSError):
+            return []
+
+    @staticmethod
+    def _signal_process_tree(
+            processes: list[psutil.Process],
+            force: bool = False,
+    ) -> None:
+        """向进程树发送终止或强制结束信号。"""
+        for child in reversed(processes):
+            try:
+                (child.kill if force else child.terminate)()
+            except (psutil.Error, OSError):
+                continue
+
     @staticmethod
     async def execute_with_subprocess_async(
             command: list,

app/adapters/system/plugin/package.py

@@ -101,16 +101,18 @@ class PluginPackageManager:
     @staticmethod
     def rollback(checkpoint: PluginPackageCheckpoint) -> None:
         """删除当前包并把变更前文件快照恢复到运行目录。"""
-        if checkpoint.plugin_dir.exists():
-            shutil.rmtree(checkpoint.plugin_dir)
         snapshot_dir = checkpoint.transaction_dir / "package"
         if checkpoint.existed:
             if not snapshot_dir.is_dir():
                 raise FileNotFoundError(
                     f"插件 {checkpoint.plugin_id} 的补偿快照不存在：{snapshot_dir}"
                 )
+        if checkpoint.plugin_dir.exists():
+            shutil.rmtree(checkpoint.plugin_dir)
+        if checkpoint.existed:
             shutil.copytree(snapshot_dir, checkpoint.plugin_dir)
-        shutil.rmtree(checkpoint.transaction_dir, ignore_errors=False)
+        if checkpoint.transaction_dir.exists():
+            shutil.rmtree(checkpoint.transaction_dir, ignore_errors=False)
 
     async def async_rollback(self, checkpoint: PluginPackageCheckpoint) -> None:
         """在线程池中恢复插件包文件快照。"""

app/application/plugin/install.py

@@ -68,6 +68,7 @@ class _InstallState:
     checkpoint: Any = None
     stage: str = "package_checkpoint"
     package_installed: bool = False
+    installed_list_touched: bool = False
     installed_list_persisted: bool = False
     runtime_touched: bool = False
     registrations_touched: bool = False
@@ -210,6 +211,8 @@ class PluginInstallCommand:
         if plugin_id not in installed_plugins:
             updated_plugins = [*installed_plugins, plugin_id]
             try:
+                # 写入方可能在返回前已经提交；取消时按已触碰处理，恢复原清单是幂等的。
+                state.installed_list_touched = True
                 await self._installed_plugins_writer(updated_plugins)
                 installed_list_persisted = True
                 state.installed_list_persisted = True
@@ -267,9 +270,10 @@ class PluginInstallCommand:
 
         checkpoint_cleanup_error = ""
         state.stage = "checkpoint_commit"
+        # 运行态和注册已完成，后续只清理临时快照，不再把取消当作未提交安装回滚。
+        state.committed = True
         try:
             await self._package_committer(checkpoint)
-            state.committed = True
         except Exception as err:
             checkpoint_cleanup_error = str(err)
 
@@ -328,7 +332,7 @@ class PluginInstallCommand:
                 stage=state.stage,
                 message="插件安装已取消",
                 package_installed=state.package_installed,
-                installed_list_persisted=state.installed_list_persisted,
+                installed_list_persisted=state.installed_list_touched,
                 runtime_touched=state.runtime_touched,
                 registrations_touched=state.registrations_touched,
             )

app/application/plugin/lifecycle.py

@@ -4,8 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import threading
-from contextlib import asynccontextmanager, contextmanager
-from typing import Iterator
+from contextlib import asynccontextmanager
 
 
 class PluginLifecycleCoordinator:
@@ -58,21 +57,6 @@ class PluginLifecycleCoordinator:
         finally:
             self._release_plugin(plugin_id)
 
-    @contextmanager
-    def hold_sync(self, plugin_id: str) -> Iterator[None]:
-        """同步持有单个插件的生命周期资格。"""
-        normalized_id = self._normalize(plugin_id)
-        if not normalized_id:
-            raise ValueError("插件ID不能为空")
-        with self._condition:
-            while self._startup_active or normalized_id in self._active_plugins:
-                self._condition.wait()
-            self._active_plugins.add(normalized_id)
-        try:
-            yield
-        finally:
-            self._release_plugin(normalized_id)
-
     @asynccontextmanager
     async def hold_startup(self):
         """异步持有启动同步的全局资格，阻止安装请求穿过启动收口。"""
@@ -85,4 +69,3 @@ class PluginLifecycleCoordinator:
 
 
 plugin_lifecycle = PluginLifecycleCoordinator()
-

app/runtime/extensions/plugin/dependency.py

@@ -63,6 +63,27 @@ class PluginDependencyService:
         """安装当前环境缺失的插件依赖并保持历史列表返回合同。"""
         return self.install_missing_with_status().missing
 
+    async def async_install_missing_with_status(self) -> PluginDependencyInstallResult:
+        """在异步启动链中恢复缺失依赖，确保安装子进程可取消。"""
+        installer = self._system().dependency
+        missing = await installer.async_find_missing()
+        if not missing:
+            return PluginDependencyInstallResult(missing=[], success=True)
+        self._logger.debug(f"检测到缺失的依赖项: {missing}")
+        self._logger.info(f"开始安装缺失的依赖项，共 {len(missing)} 个...")
+        started = time.time()
+        success, _message = await installer.async_install(missing)
+        elapsed = time.time() - started
+        if success:
+            self._logger.info(
+                f"已完成 {len(missing)} 个依赖项安装，总耗时：{elapsed:.2f} 秒"
+            )
+        else:
+            self._logger.warning(
+                f"存在缺失依赖项安装失败，请尝试手动安装，总耗时：{elapsed:.2f} 秒"
+            )
+        return PluginDependencyInstallResult(missing=missing, success=success)
+
     def classify_plugins(self) -> PluginDependencyClassification:
         """返回启动编排使用的轻量插件分类。"""
         ready, missing_dependencies, missing_source = (

app/runtime/extensions/plugin_manager.py

@@ -603,6 +603,14 @@ class PluginManager(ConfigReloadMixin, metaclass=Singleton):
             log=logger,
         ).install_missing_with_status()
 
+    @staticmethod
+    async def async_install_plugin_missing_dependencies_with_status() -> PluginDependencyInstallResult:
+        """在异步启动链中恢复插件依赖并保留取消语义。"""
+        return await PluginDependencyService(
+            system=get_plugin_system,
+            log=logger,
+        ).async_install_missing_with_status()
+
     def classify_plugins(self) -> PluginDependencyClassification:
         """按源码依赖状态分类物理插件，并把结果映射到虚拟实例。"""
         source_classification = PluginDependencyService(

app/startup/plugins_initializer.py

@@ -148,10 +148,8 @@ async def sync_plugins() -> bool:
         plugin_manager.set_plugin_settling(True)
 
         sync_result = await execute_task(loop, plugin_manager.sync, "插件同步到本地")
-        dependency_result = await execute_task(
-            loop,
-            plugin_manager.install_plugin_missing_dependencies_with_status,
-            "缺失依赖项安装",
+        dependency_result = await (
+            plugin_manager.async_install_plugin_missing_dependencies_with_status()
         )
         if dependency_result is None:
             return False

tests/fixtures/architecture/dependency-baseline.json

@@ -13,8 +13,8 @@
     "runtime_to_db": [],
     "workflow_to_db": []
   },
-  "edge_count": 6444,
-  "edge_sha256": "b5db7b31c7ea4dd7311fcd9e11a49eb32feb6939f896ed85703b3573408752a8",
+  "edge_count": 6440,
+  "edge_sha256": "93995005b1d5d1da95e9ec73a39c1d92e8e8e3e3ce3751ce1035e3f31103350d",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -237,8 +237,6 @@
     "app.agent.mcp -> app.schemas.types",
     "app.agent.memory -> app.application",
     "app.agent.memory -> app.application.agentdata",
-    "app.agent.memory -> app.application.messaging",
-    "app.agent.memory -> app.application.messaging.chat",
     "app.agent.memory -> app.runtime",
     "app.agent.memory -> app.runtime.log",
     "app.agent.memory -> app.runtime.settings",
@@ -347,8 +345,6 @@
     "app.agent.orchestrator -> app.agent.tools.impl.query_system_settings",
     "app.agent.orchestrator -> app.application",
     "app.agent.orchestrator -> app.application.agentdata",
-    "app.agent.orchestrator -> app.application.messaging",
-    "app.agent.orchestrator -> app.application.messaging.chat",
     "app.agent.orchestrator -> app.application.plugin",
     "app.agent.orchestrator -> app.application.plugin.runtime",
     "app.agent.orchestrator -> app.chain",
@@ -2557,10 +2553,6 @@
     "app.application.mediaserver -> app.schemas.types",
     "app.application.messaging.agent -> app.schemas",
     "app.application.messaging.agent -> app.schemas.types",
-    "app.application.messaging.chat -> app.application",
-    "app.application.messaging.chat -> app.application.database",
-    "app.application.messaging.chat -> app.runtime",
-    "app.application.messaging.chat -> app.runtime.observability",
     "app.application.messaging.chat -> app.schemas",
     "app.application.messaging.chat -> app.schemas.agent",
     "app.application.messaging.interaction -> app.schemas",
@@ -2650,6 +2642,10 @@
     "app.application.plugin.folders -> app.schemas.types",
     "app.application.plugin.install -> app.application",
     "app.application.plugin.install -> app.application.database",
+    "app.application.plugin.install -> app.application.plugin",
+    "app.application.plugin.install -> app.application.plugin.lifecycle",
+    "app.application.plugin.install -> app.runtime",
+    "app.application.plugin.install -> app.runtime.log",
     "app.application.recognition -> app.application",
     "app.application.recognition -> app.application.configuration",
     "app.application.recognition -> app.schemas",
@@ -6039,6 +6035,7 @@
     "app.startup.lifecycle -> app.adapters.network.http",
     "app.startup.lifecycle -> app.application",
     "app.startup.lifecycle -> app.application.plugin",
+    "app.startup.lifecycle -> app.application.plugin.lifecycle",
     "app.startup.lifecycle -> app.application.plugin.runtime",
     "app.startup.lifecycle -> app.chain",
     "app.startup.lifecycle -> app.chain.system",
@@ -6097,7 +6094,6 @@
     "app.startup.modules_initializer -> app.application.history",
     "app.startup.modules_initializer -> app.application.image",
     "app.startup.modules_initializer -> app.application.messaging",
-    "app.startup.modules_initializer -> app.application.messaging.agent",
     "app.startup.modules_initializer -> app.application.messaging.chat",
     "app.startup.modules_initializer -> app.application.messaging.message",
     "app.startup.modules_initializer -> app.application.module",
@@ -6461,7 +6457,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 798,
+  "module_count": 799,
   "modules": [
     "app",
     "app.adapters",
@@ -6753,6 +6749,7 @@
     "app.application.plugin.data",
     "app.application.plugin.folders",
     "app.application.plugin.install",
+    "app.application.plugin.lifecycle",
     "app.application.plugin.routes",
     "app.application.plugin.runtime",
     "app.application.recognition",

tests/test_plugin_dependency_service.py

@@ -1,7 +1,13 @@
+from unittest.mock import AsyncMock
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
-from app.runtime.extensions.plugin.dependency import PluginDependencyService
+import pytest
+
+from app.runtime.extensions.plugin.dependency import (
+    PluginDependencyInstallResult,
+    PluginDependencyService,
+)
 
 
 def test_install_missing_skips_installer_when_environment_is_satisfied() -> None:
@@ -35,3 +41,25 @@ def test_install_missing_preserves_list_return_contract() -> None:
 
     assert service.install_missing() == ["demo>=1"]
     installer.install.assert_called_once_with(["demo>=1"])
+
+
+@pytest.mark.asyncio
+async def test_async_install_missing_uses_async_installer() -> None:
+    """异步启动恢复必须调用可取消的依赖安装入口。"""
+    installer = SimpleNamespace(
+        async_find_missing=AsyncMock(return_value=["demo>=1"]),
+        async_install=AsyncMock(return_value=(True, "")),
+    )
+    service = PluginDependencyService(
+        system=lambda: SimpleNamespace(dependency=installer),
+        log=MagicMock(),
+    )
+
+    result = await service.async_install_missing_with_status()
+
+    assert result == PluginDependencyInstallResult(
+        missing=["demo>=1"],
+        success=True,
+    )
+    installer.async_find_missing.assert_awaited_once()
+    installer.async_install.assert_awaited_once_with(["demo>=1"])

tests/test_plugin_install_command.py

@@ -371,6 +371,58 @@ async def test_cancelled_install_waits_for_rollback_before_releasing_lifecycle()
     rollback.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_cancelled_persisted_list_is_restored_conservatively() -> None:
+    """清单写入已产生副作用但尚未返回时取消，也必须恢复原清单。"""
+    persisted: list[list[str]] = []
+    writer_started = asyncio.Event()
+    rollback = AsyncMock()
+
+    async def writer(plugin_ids: list[str]) -> None:
+        persisted.append(list(plugin_ids))
+        if len(persisted) == 1:
+            writer_started.set()
+            await asyncio.Event().wait()
+
+    task = asyncio.create_task(
+        _command(writer=writer, rollback=rollback).execute(
+            plugin_id="DemoPlugin",
+            repo_url="https://github.com/demo/plugins",
+        )
+    )
+    await writer_started.wait()
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    assert persisted == [["DemoPlugin"], []]
+    rollback.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_cancelled_snapshot_cleanup_does_not_rollback_committed_plugin() -> None:
+    """运行态提交后清理快照期间取消，不得删除已生效插件。"""
+    cleanup_started = asyncio.Event()
+    rollback = AsyncMock()
+
+    async def committer(_checkpoint) -> None:
+        cleanup_started.set()
+        await asyncio.Event().wait()
+
+    task = asyncio.create_task(
+        _command(committer=committer, rollback=rollback).execute(
+            plugin_id="DemoPlugin",
+            repo_url="https://github.com/demo/plugins",
+        )
+    )
+    await cleanup_started.wait()
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    rollback.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_startup_lifecycle_lock_blocks_plugin_install_until_settlement() -> None:
     """启动同步持有全局资格时，插件安装不得穿过启动收口。"""

tests/test_plugin_monitor_lifecycle.py

@@ -2,7 +2,7 @@ import asyncio
 import threading
 import time
 from types import SimpleNamespace
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
@@ -139,6 +139,12 @@ def _patch_sync_plugins(monkeypatch, manager: MagicMock) -> MagicMock:
     monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager)
     monkeypatch.setattr(plugins_initializer, "execute_task", execute)
     monkeypatch.setattr(plugins_initializer, "register_plugin_api", register)
+    dependency_result = (
+        manager.async_install_plugin_missing_dependencies_with_status.return_value
+    )
+    manager.async_install_plugin_missing_dependencies_with_status = AsyncMock(
+        return_value=dependency_result,
+    )
     manager.get_plugin_runtime_statuses.return_value = {}
     return register
 
@@ -150,7 +156,7 @@ async def test_sync_plugins_activates_ready_plugins_when_dependencies_fail(
     """依赖恢复失败时仍激活无关的已就绪插件。"""
     manager = MagicMock()
     manager.sync.return_value = ["demo"]
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
         PluginDependencyInstallResult(missing=["demo>=1"], success=False)
     )
     manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -175,7 +181,7 @@ async def test_sync_plugins_loads_only_plugins_that_become_ready(
     """后台依赖恢复后只启动尚未运行且当前已就绪的插件。"""
     manager = MagicMock()
     manager.sync.return_value = []
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
         PluginDependencyInstallResult(missing=["demo>=1"], success=True)
     )
     manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -204,7 +210,7 @@ async def test_sync_plugins_reloads_only_updated_running_plugins(monkeypatch) ->
     """源码同步只重载对应运行实例，不重启其他插件。"""
     manager = MagicMock()
     manager.sync.return_value = ["UpdatedPlugin"]
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
         PluginDependencyInstallResult(missing=[], success=True)
     )
     manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -232,7 +238,7 @@ async def test_sync_plugins_reloads_running_plugin_after_dependency_recovery(
     """依赖恢复后，已运行的旧实例必须切换到新源码。"""
     manager = MagicMock()
     manager.sync.return_value = []
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
         PluginDependencyInstallResult(missing=["demo>=1"], success=True)
     )
     manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -258,7 +264,7 @@ async def test_sync_plugins_keeps_runtime_when_nothing_changed(monkeypatch) -> N
     """源码和依赖均无变化时保留首次初始化结果。"""
     manager = MagicMock()
     manager.sync.return_value = []
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
         PluginDependencyInstallResult(missing=[], success=True)
     )
     manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -283,7 +289,7 @@ async def test_sync_plugins_keeps_event_loop_responsive_during_activation(
     """插件初始化运行在线程池时，Web 事件循环仍可继续调度。"""
     manager = MagicMock()
     manager.sync.return_value = []
-    manager.install_plugin_missing_dependencies_with_status.return_value = (
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
         PluginDependencyInstallResult(missing=[], success=True)
     )
     manager.classify_plugins.return_value = PluginDependencyClassification(
@@ -299,6 +305,9 @@ async def test_sync_plugins_keeps_event_loop_responsive_during_activation(
         time.sleep(0.1)
 
     manager.start.side_effect = slow_start
+    manager.async_install_plugin_missing_dependencies_with_status = AsyncMock(
+        return_value=PluginDependencyInstallResult(missing=[], success=True),
+    )
     monkeypatch.setattr(plugins_initializer, "configure_plugin_services", lambda: None)
     monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager)
     monkeypatch.setattr(plugins_initializer, "register_plugin_api", MagicMock())

tests/test_plugin_package_manager.py

@@ -1,7 +1,10 @@
+import shutil
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import Mock
 
+import pytest
+
 from app.adapters.system.plugin.package import PluginPackageManager
 
 
@@ -46,6 +49,23 @@ def test_checkpoint_rollback_removes_new_package(monkeypatch, tmp_path):
     assert not checkpoint.transaction_dir.exists()
 
 
+def test_rollback_does_not_delete_package_when_snapshot_is_missing(monkeypatch, tmp_path):
+    """补偿快照损坏时先失败，不能先删除当前可用插件。"""
+    manager = _manager(monkeypatch, tmp_path)
+    plugin_dir = tmp_path / "app" / "plugins" / "demoplugin"
+    plugin_dir.mkdir(parents=True)
+    (plugin_dir / "__init__.py").write_text("old", encoding="utf-8")
+
+    checkpoint = manager.checkpoint("DemoPlugin")
+    shutil.rmtree(checkpoint.transaction_dir / "package")
+    (plugin_dir / "__init__.py").write_text("new", encoding="utf-8")
+
+    with pytest.raises(FileNotFoundError):
+        manager.rollback(checkpoint)
+
+    assert (plugin_dir / "__init__.py").read_text(encoding="utf-8") == "new"
+
+
 def test_local_sync_failure_restores_previous_runtime_copy(monkeypatch, tmp_path):
     """本地来源不可复制时不得丢失已经运行的插件副本。"""
     manager = _manager(monkeypatch, tmp_path)

tests/test_system_utils.py

@@ -11,6 +11,7 @@ from pathlib import Path
 from unittest import TestCase
 from unittest.mock import MagicMock, call, patch
 
+import psutil
 import pytest
 
 from app.runtime.state import SystemHelper
@@ -205,6 +206,51 @@ async def test_async_subprocess_cancellation_reaps_process(tmp_path):
         pytest.fail(f"子进程仍在运行：{pid}")
 
 
+@pytest.mark.asyncio
+async def test_async_subprocess_cancellation_reaps_process_tree(tmp_path):
+    """取消安装命令时，子进程派生的构建进程也不得继续运行。"""
+    marker = tmp_path / "pids"
+    child_code = "import time; time.sleep(60)"
+    command = [
+        sys.executable,
+        "-c",
+        (
+            "from pathlib import Path; import subprocess, os, time; "
+            f"child = subprocess.Popen([{sys.executable!r}, '-c', {child_code!r}]); "
+            f"Path({str(marker)!r}).write_text(str(os.getpid()) + ':' + str(child.pid)); "
+            "time.sleep(60)"
+        ),
+    ]
+    task = asyncio.create_task(
+        SystemUtils.execute_with_subprocess_async(command, timeout=30)
+    )
+    deadline = time.monotonic() + 2
+    while not marker.exists() and time.monotonic() < deadline:
+        await asyncio.sleep(0.01)
+    assert marker.exists()
+
+    pids = [int(value) for value in marker.read_text().split(":")]
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    for _ in range(100):
+        alive = []
+        for pid in pids:
+            try:
+                process = psutil.Process(pid)
+                if not process.is_running() or process.status() == psutil.STATUS_ZOMBIE:
+                    continue
+            except (psutil.Error, OSError):
+                continue
+            alive.append(pid)
+        if not alive:
+            break
+        await asyncio.sleep(0.01)
+    else:
+        pytest.fail(f"进程树仍在运行：{alive}")
+
+
 def test_execute_with_subprocess_redacts_userinfo_from_stdout_and_stderr():
     error = subprocess.CalledProcessError(
         returncode=1,