fix(plugin): close process tree cleanup gaps

app/adapters/system/host.py

@@ -205,28 +205,48 @@ class SystemUtils:
                 timeout=grace_seconds,
             )
         except (asyncio.TimeoutError, asyncio.CancelledError):
-            try:
-                if os.name == "nt":
-                    SystemUtils._signal_process_tree(process_tree, force=True)
+            pass
+
+        # 管道可能在父进程退出时立即关闭，而后代仍在运行或忽略终止信号。
+        # 通信任务完成只代表 stdout/stderr 已收口，不能作为进程树已收口的依据。
+        try:
+            known_pids = {item.pid for item in process_tree}
+            process_tree.extend(
+                process_item
+                for process_item in SystemUtils._process_tree(process.pid)
+                if process_item.pid not in known_pids
+            )
+        except (ProcessLookupError, OSError):
+            pass
+
+        alive_processes = SystemUtils._alive_processes(process_tree)
+        if alive_processes or process.returncode is None:
+            if os.name == "nt":
+                try:
+                    SystemUtils._signal_process_tree(alive_processes, force=True)
                     process.kill()
-                else:
+                except (ProcessLookupError, OSError):
+                    pass
+            else:
+                try:
                     os.killpg(process.pid, signal.SIGKILL)
-                    SystemUtils._signal_process_tree(process_tree, force=True)
-            except (ProcessLookupError, OSError):
-                pass
-            try:
-                await asyncio.wait_for(
-                    asyncio.shield(communication_task),
-                    timeout=grace_seconds,
-                )
-            except (asyncio.TimeoutError, asyncio.CancelledError):
-                communication_task.cancel()
-                await asyncio.gather(communication_task, return_exceptions=True)
-        finally:
-            try:
-                await process.wait()
-            except (ProcessLookupError, OSError):
-                pass
+                except (ProcessLookupError, OSError):
+                    pass
+                try:
+                    SystemUtils._signal_process_tree(alive_processes, force=True)
+                except (ProcessLookupError, OSError):
+                    pass
+            await SystemUtils._wait_process_tree(
+                process_tree,
+                grace_seconds,
+            )
+        if not communication_task.done():
+            communication_task.cancel()
+            await asyncio.gather(communication_task, return_exceptions=True)
+        try:
+            await process.wait()
+        except (ProcessLookupError, OSError):
+            pass
 
     @staticmethod
     def _process_tree(pid: int) -> list[psutil.Process]:
@@ -237,6 +257,38 @@ class SystemUtils:
         except (psutil.Error, OSError):
             return []
 
+    @staticmethod
+    def _alive_processes(processes: list[psutil.Process]) -> list[psutil.Process]:
+        """返回仍可能执行外部副作用的进程，忽略已退出和僵尸进程。"""
+        alive = []
+        seen_pids = set()
+        for process in processes:
+            if process.pid in seen_pids:
+                continue
+            seen_pids.add(process.pid)
+            try:
+                if process.is_running() and process.status() != psutil.STATUS_ZOMBIE:
+                    alive.append(process)
+            except (psutil.Error, OSError):
+                continue
+        return alive
+
+    @staticmethod
+    async def _wait_process_tree(
+            processes: list[psutil.Process],
+            timeout: float,
+    ) -> list[psutil.Process]:
+        """在事件循环中有界等待整棵进程树退出，并返回残留进程。"""
+        deadline = asyncio.get_running_loop().time() + max(timeout, 0)
+        while True:
+            alive = SystemUtils._alive_processes(processes)
+            if not alive:
+                return []
+            remaining = deadline - asyncio.get_running_loop().time()
+            if remaining <= 0:
+                return alive
+            await asyncio.sleep(min(0.05, remaining))
+
     @staticmethod
     def _signal_process_tree(
             processes: list[psutil.Process],

app/application/plugin/install.py

@@ -224,6 +224,7 @@ class PluginInstallCommand:
                     stage="installed_list_persistence",
                     message=str(err),
                     package_installed=True,
+                    installed_list_persisted=state.installed_list_touched,
                 )
                 if isinstance(err, DatabaseWorkerOverloadedError):
                     raise

tests/test_plugin_install_command.py

@@ -3,7 +3,6 @@ from unittest.mock import AsyncMock, Mock
 
 import pytest
 
-from app.application.database import DatabaseWorkerOverloadedError
 from app.application.plugin.install import PluginInstallCommand
 
 
@@ -177,12 +176,41 @@ async def test_persistence_failure_restores_package_without_touching_runtime():
     assert result.success is False
     assert result.failure_stage == "installed_list_persistence"
     assert result.rollback.file_restored is True
-    assert result.rollback.installed_list_attempted is False
+    assert result.rollback.installed_list_attempted is True
     assert result.rollback.runtime_attempted is False
     rollback.assert_awaited_once_with(checkpoint)
     reloader.assert_not_awaited()
 
 
+@pytest.mark.asyncio
+async def test_persistence_exception_after_write_restores_installed_list():
+    """清单写入已提交后抛异常时，文件和清单必须一起恢复。"""
+    persisted: list[list[str]] = []
+    checkpoint = object()
+    rollback = AsyncMock()
+
+    async def write(plugin_ids: list[str]) -> None:
+        persisted.append(list(plugin_ids))
+        if len(persisted) == 1:
+            raise RuntimeError("write acknowledgement lost")
+
+    result = await _command(
+        checkpointer=AsyncMock(return_value=checkpoint),
+        writer=write,
+        rollback=rollback,
+    ).execute(
+        plugin_id="DemoPlugin",
+        repo_url="https://github.com/demo/plugins",
+    )
+
+    assert result.success is False
+    assert result.failure_stage == "installed_list_persistence"
+    assert result.rollback.installed_list_attempted is True
+    assert result.rollback.installed_list_restored is True
+    assert persisted == [["DemoPlugin"], []]
+    rollback.assert_awaited_once_with(checkpoint)
+
+
 @pytest.mark.asyncio
 async def test_database_worker_overload_rolls_back_and_reaches_api_boundary():
     """配置 worker 背压完成补偿后继续抛出，交由 API 映射为 503。"""

tests/test_system_utils.py

@@ -251,6 +251,56 @@ async def test_async_subprocess_cancellation_reaps_process_tree(tmp_path):
         pytest.fail(f"进程树仍在运行：{alive}")
 
 
+@pytest.mark.skipif(os.name == "nt", reason="Windows 没有 POSIX 进程组信号语义")
+@pytest.mark.asyncio
+async def test_async_subprocess_reaps_descendant_after_early_pipe_close(tmp_path):
+    """父进程关闭管道后，忽略终止信号的后代也必须被强制回收。"""
+    marker = tmp_path / "pids"
+    child_code = (
+        "import os, signal, time; os.close(1); os.close(2); "
+        "signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(60)"
+    )
+    command = [
+        sys.executable,
+        "-c",
+        (
+            "from pathlib import Path; import os, signal, subprocess, time; "
+            f"child = subprocess.Popen([{sys.executable!r}, '-c', {child_code!r}], "
+            "start_new_session=True); "
+            f"Path({str(marker)!r}).write_text(str(os.getpid()) + ':' + str(child.pid)); "
+            "signal.signal(signal.SIGTERM, lambda *_: os._exit(0)); 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,