From 6c2817e383d6ec90e532dfb0ff80482612970aa8 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 06:56:44 +0800 Subject: [PATCH] fix(plugin): close cancellable install lifecycle gaps --- app/adapters/system/host.py | 25 +++++++++ app/adapters/system/plugin/package.py | 8 +-- app/application/plugin/install.py | 8 ++- app/application/plugin/lifecycle.py | 19 +------ app/runtime/extensions/plugin/dependency.py | 21 ++++++++ app/runtime/extensions/plugin_manager.py | 8 +++ app/startup/plugins_initializer.py | 6 +-- .../architecture/dependency-baseline.json | 21 ++++---- tests/test_plugin_dependency_service.py | 30 ++++++++++- tests/test_plugin_install_command.py | 52 +++++++++++++++++++ tests/test_plugin_monitor_lifecycle.py | 23 +++++--- tests/test_plugin_package_manager.py | 20 +++++++ tests/test_system_utils.py | 46 ++++++++++++++++ 13 files changed, 240 insertions(+), 47 deletions(-) diff --git a/app/adapters/system/host.py b/app/adapters/system/host.py index e120840f5..ec1d9ace9 100644 --- a/app/adapters/system/host.py +++ b/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, diff --git a/app/adapters/system/plugin/package.py b/app/adapters/system/plugin/package.py index f69de831b..c01e05fe1 100644 --- a/app/adapters/system/plugin/package.py +++ b/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: """在线程池中恢复插件包文件快照。""" diff --git a/app/application/plugin/install.py b/app/application/plugin/install.py index ff844934b..14ef8ebf4 100644 --- a/app/application/plugin/install.py +++ b/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, ) diff --git a/app/application/plugin/lifecycle.py b/app/application/plugin/lifecycle.py index 587979448..04b85a551 100644 --- a/app/application/plugin/lifecycle.py +++ b/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() - diff --git a/app/runtime/extensions/plugin/dependency.py b/app/runtime/extensions/plugin/dependency.py index 691c9c6b1..5a026f4b2 100644 --- a/app/runtime/extensions/plugin/dependency.py +++ b/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 = ( diff --git a/app/runtime/extensions/plugin_manager.py b/app/runtime/extensions/plugin_manager.py index d593e6e71..746b3a00d 100644 --- a/app/runtime/extensions/plugin_manager.py +++ b/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( diff --git a/app/startup/plugins_initializer.py b/app/startup/plugins_initializer.py index 583feafb3..f82a6f26b 100644 --- a/app/startup/plugins_initializer.py +++ b/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 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index ce2f2bbf3..abf4f749c 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/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", diff --git a/tests/test_plugin_dependency_service.py b/tests/test_plugin_dependency_service.py index b18981756..c762ffad6 100644 --- a/tests/test_plugin_dependency_service.py +++ b/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"]) diff --git a/tests/test_plugin_install_command.py b/tests/test_plugin_install_command.py index 7ae5c4691..f7c90c9b8 100644 --- a/tests/test_plugin_install_command.py +++ b/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: """启动同步持有全局资格时，插件安装不得穿过启动收口。""" diff --git a/tests/test_plugin_monitor_lifecycle.py b/tests/test_plugin_monitor_lifecycle.py index c76952a14..11a621ed1 100644 --- a/tests/test_plugin_monitor_lifecycle.py +++ b/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()) diff --git a/tests/test_plugin_package_manager.py b/tests/test_plugin_package_manager.py index 03fa7dbdc..ad48f91c4 100644 --- a/tests/test_plugin_package_manager.py +++ b/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) diff --git a/tests/test_system_utils.py b/tests/test_system_utils.py index d7fe0d67b..d899989e7 100644 --- a/tests/test_system_utils.py +++ b/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,