refactor: 迁移 ops/ 下 31 处 subprocess.run(check=True) 到 px.sh，统一错误处理

1. 调整storage.py里的batch方法返回类型为Generator
2. 格式化测试文件里的Graph.from_specs调用代码

.github/workflows/ci.yml

@@ -30,5 +30,5 @@ jobs:
       - name: Ruff check
         run: ruff check src tests
 
-      - name: Tox test (py38, py313)
-        run: uvx tox run -e py38,py313
+      - name: Tox test (py310, py314)
+        run: uvx tox run -e py310,py314

.trae/docs/iter-29-subprocess-migration.md

@@ -0,0 +1,63 @@
+# 迭代 29: subprocess.run → px.sh 迁移
+
+## 本轮目标
+
+将 `ops/` 目录下所有 `subprocess.run(cmd, check=True)` 调用迁移到 `px.sh` 封装，统一错误处理与中文输出。
+
+## 改动文件清单
+
+### 源文件 (9 个)
+
+| 文件 | 迁移调用数 | 保留调用数 | 保留原因 |
+|------|-----------|-----------|---------|
+| dev/gittool.py | 5 | 1 | `has_files()` 用 `check=False` 检查状态 |
+| dev/piptool.py | 8 | 0 | 全部迁移，移除 `import subprocess` |
+| dev/lscalc.py | 3 | 1 | Linux `pgrep` 用 `check=False` |
+| dev/bumpversion.py | 3 | 0 | 全部迁移，移除 `import subprocess` |
+| dev/packtool.py | 2 | 0 | 全部迁移，移除 `import subprocess` |
+| dev/autofmt.py | 4 | 1 | `sync_pyproject_config` 用 `check=False` (失败可忽略) |
+| infra/dockercmd.py | 1 | 0 | 全部迁移，移除 `import subprocess` |
+| infra/msdownload.py | 1 | 0 | 全部迁移，移除 `import subprocess` |
+| files/screenshot.py | 4 | 4 | Linux 回退模式用 `FileNotFoundError` 作为控制流 |
+
+### 未迁移文件
+
+- `infra/sshcopyid.py` — 使用 `timeout=` 参数（`px.sh` 不支持）
+- `infra/envdev.py` — 全部 `check=False`
+- `system/clr.py`, `system/reseticoncache.py`, `system/taskkill.py`, `system/which.py` — 全部 `check=False`
+
+### 测试文件 (8 个)
+
+- `test_piptool.py` — mock 目标从 `subprocess.run` 改为 `pyflowx.sh`，错误测试从 `side_effect` 改为 `return_value=None`
+- `test_lscalc.py` — 迁移调用改 mock `pyflowx.sh`，Linux `check=False` 调用保留 `subprocess.run`
+- `test_packtool.py` — mock 目标改为 `pyflowx.sh`
+- `test_autofmt.py` — 迁移调用改 mock `pyflowx.sh`，`sync_pyproject_config` 保留 `subprocess.run`
+- `test_screenshot.py` — Win/macOS 测试改 mock `pyflowx.sh`，Linux 回退测试保留 `subprocess.run`
+- `test_bumpversion.py` — `_mock_subprocess` 改为 `_mock_sh`，mock `pyflowx.sh`
+- `test_envdev.py` — 更新 dockercmd 错误消息断言（`px.sh` 消息格式与原自定义消息不同）
+- `test_llm.py` — 更新 msdownload 错误消息断言
+
+## 关键决策与依据
+
+1. **`check=False` 调用不迁移**：`px.sh` 内部 `check=True`，失败时返回 `None` + 打印错误。`check=False` 调用需要检查 `returncode` / `stdout`，语义不同。
+
+2. **screenshot Linux 回退不迁移**：`try: gnome-screenshot except FileNotFoundError: scrot` 模式中 `FileNotFoundError` 是控制流（回退到备用命令），不是错误。`px.sh` 会打印误导性错误信息。
+
+3. **sshcopyid 不迁移**：使用 `timeout=` 参数控制 SSH 超时，`px.sh` 不支持此参数。
+
+4. **测试 mock 策略**：迁移调用的测试从 mock `subprocess.run` 改为 mock `pyflowx.sh`，测试调用方的行为（处理 `None` 返回值）而非底层 `subprocess.run` 的异常。
+
+5. **错误消息断言放宽**：`px.sh` 的错误消息格式（`"{label}失败 (returncode=N): ..."`）与原自定义消息不同，测试断言改为检查关键词（`"失败"`、`"命令未找到"`）而非精确消息。
+
+## 验证结果
+
+- 全部 1731 个测试通过
+- 覆盖率 97.24%（门槛 95%）
+- ruff check + format 全部通过
+- 迁移后源文件覆盖率均为 100%
+
+## 遗留事项
+
+- `system/` 目录和 `infra/envdev.py` 的 `check=False` 调用未迁移（语义不同，不应迁移）
+- `infra/sshcopyid.py` 未迁移（需要 `timeout=` 支持）
+- 如需统一这些调用，可考虑扩展 `px.sh` 支持 `check=False` 和 `timeout` 参数

pyproject.toml

@@ -21,12 +21,13 @@ license = { text = "MIT" }
 name = "pyflowx"
 readme = "README.md"
 requires-python = ">=3.10"
-version = "0.4.9"
+version = "0.4.11"
 
 [project.scripts]
-emlman = "pyflowx.cli.legacy.emlmanager:main"
-pf     = "pyflowx.cli.pf:main"
-pxp    = "pyflowx.cli.legacy.profiler:main"
+emlman  = "pyflowx.cli.legacy.emlmanager:main"
+pf      = "pyflowx.cli.pf:main"
+pxp     = "pyflowx.cli.legacy.profiler:main"
+pyflowx = "pyflowx.cli.pf:main"
 
 [project.optional-dependencies]
 dev = [
@@ -105,7 +106,6 @@ target-version = "py310"
 ignore = [
     "E501",    # line too long (handled by formatter)
     "PLC0415", # import should be at top-level (intentional for lazy imports)
-    "PLR0913", # too many arguments
     "PLR0915", # too many statements (intentional for complex methods)
     "PLR2004", # magic value comparison
     "PTH119",  # os.path.basename (intentional for sys.argv)

src/pyflowx/__init__.py

@@ -112,7 +112,7 @@ from .task import (
 from .tools import ToolSpec, list_subcommands, list_tools, run_tool, tool
 from .yaml_loader import load_yaml, parse_yaml_string
 
-__version__ = "0.4.9"
+__version__ = "0.4.11"
 
 
 def graph(

src/pyflowx/executors.py

@@ -52,6 +52,7 @@ import queue
 import threading
 import time
 from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping
+from dataclasses import dataclass, field
 from dataclasses import replace as dc_replace
 from datetime import datetime
 from pathlib import Path
@@ -132,6 +133,24 @@ EventCallback = Callable[[TaskEvent], None]
 Strategy = Literal["sequential", "thread", "async", "dependency"]
 
 
+@dataclass(frozen=True)
+class _ExecContext:
+    """执行上下文：捆绑 run() 调用链中共享的状态，减少参数传递。
+
+    将 context/report/backend/on_event/concurrency_limits/cancel_event
+    打包为单一参数，使调用链中每个函数的参数数 ≤5（PLR0913）。
+    frozen=True 保证调用链中不可意外替换整体引用，但不阻止对
+    context/report 等可变属性的内部修改（如 ``ctx.context[name] = value``）。
+    """
+
+    context: dict[str, Any]
+    report: RunReport
+    backend: StateBackend
+    on_event: EventCallback | None
+    concurrency_limits: Mapping[str, int] = field(default_factory=dict)
+    cancel_event: threading.Event | CancelToken | None = None
+
+
 # ---------------------------------------------------------------------- #
 # 无状态公共辅助
 # ---------------------------------------------------------------------- #
@@ -208,32 +227,28 @@ def _build_context(
 
 
 def _apply_cached(
-    name: str,
     spec: TaskSpec[Any],
-    context: dict[str, Any],
-    report: RunReport,
-    backend: StateBackend,
-    on_event: EventCallback | None,
+    ctx: _ExecContext,
 ) -> bool:
-    """若 ``name`` 命中缓存，写入 context/report 并返回 True。
+    """若 ``spec.name`` 命中缓存，写入 context/report 并返回 True。
 
     单次 ``backend.get`` + ``KeyError`` 回退，避免 ``has`` + ``get`` 双重
     哈希查找与双重 TTL 判断。
     """
     # 快速路径：无 cache_key 时存储键即为任务名，避免函数调用开销。
-    storage_key = spec.name if spec.cache_key is None else spec.storage_key(context)
+    storage_key = spec.name if spec.cache_key is None else spec.storage_key(ctx.context)
     try:
-        cached = backend.get(storage_key)
+        cached = ctx.backend.get(storage_key)
     except KeyError:
         return False
-    context[name] = cached
+    ctx.context[spec.name] = cached
     result = TaskResult(spec=spec, status=TaskStatus.SKIPPED, value=cached, reason="缓存命中")
-    report.results[name] = result
-    _emit(on_event, result)
+    ctx.report.results[spec.name] = result
+    _emit(ctx.on_event, result)
     logger.info(
         "task %r skipped (cached)",
-        name,
-        extra={"run_id": report.run_id, "task_name": name, "status": TaskStatus.SKIPPED.value},
+        spec.name,
+        extra={"run_id": ctx.report.run_id, "task_name": spec.name, "status": TaskStatus.SKIPPED.value},
     )
     return True
 
@@ -351,27 +366,24 @@ def _mark_success(spec: TaskSpec[Any], result: TaskResult[Any], value: Any) -> N
 def _finalize_failure(
     result: TaskResult[Any],
     layer_idx: int | None,
-    on_event: EventCallback | None,
+    ctx: _ExecContext,
     continue_on_error: bool,
-    name: str,
-    report: RunReport | None,
 ) -> None:
     """标记任务为 FAILED。若 ``continue_on_error`` 为真则不抛出异常。
 
-    失败结果在抛出前写入 ``report.results[name]``，使流式 API
+    失败结果在抛出前写入 ``ctx.report.results``，使流式 API
     (:func:`run_iter`) 能在 re-raise 前yield 该结果。
     """
     result.status = TaskStatus.FAILED
     result.finished_at = datetime.now()
-    if report is not None:
-        report.results[name] = result
-    _emit(on_event, result)
+    ctx.report.results[result.spec.name] = result
+    _emit(ctx.on_event, result)
     if continue_on_error:
         logger.warning(
             "task %r failed but continue_on_error=True; continuing.",
             result.spec.name,
             extra={
-                "run_id": report.run_id if report is not None else "-",
+                "run_id": ctx.report.run_id,
                 "task_name": result.spec.name,
                 "status": TaskStatus.FAILED.value,
                 "attempts": result.attempts,
@@ -384,7 +396,7 @@ def _finalize_failure(
         cause=result.error if result.error is not None else RuntimeError("unknown"),
         attempts=result.attempts,
         layer=layer_idx,
-        report=report,
+        report=ctx.report,
     )
 
 
@@ -393,9 +405,7 @@ def _handle_failure(
     result: TaskResult[Any],
     exc: BaseException,
     layer_idx: int | None,
-    on_event: EventCallback | None,
-    name: str,
-    report: RunReport | None,
+    ctx: _ExecContext,
 ) -> bool:
     """统一处理失败：超时转换、重试决策、finalize。
 
@@ -405,7 +415,7 @@ def _handle_failure(
         ``True`` 表示已 finalize（不再重试）；``False`` 表示应继续重试。
     """
     # asyncio.TimeoutError → TaskTimeoutError（统一异常类型）
-    run_id = report.run_id if report is not None else "-"
+    run_id = ctx.report.run_id
     if isinstance(exc, asyncio.TimeoutError):
         exc = TaskTimeoutError(spec.name, spec.timeout or 0.0)
         logger.warning(
@@ -440,7 +450,7 @@ def _handle_failure(
     if _should_retry(spec, result.attempts, exc):
         return False
     _run_hooks(spec.hooks, "on_failure", spec, exc)
-    _finalize_failure(result, layer_idx, on_event, spec.continue_on_error, name, report)
+    _finalize_failure(result, layer_idx, ctx, spec.continue_on_error)
     return True
 
 
@@ -453,21 +463,20 @@ class SyncTaskRunner:
     @staticmethod
     def run(
         spec: TaskSpec[Any],
-        context: Mapping[str, Any],
+        task_ctx: Mapping[str, Any],
         layer_idx: int | None,
-        on_event: EventCallback | None = None,
-        report: RunReport | None = None,
+        ctx: _ExecContext,
     ) -> TaskResult[Any]:
-        skipped = _prepare_for_execution(spec, context, report, on_event)
+        skipped = _prepare_for_execution(spec, task_ctx, ctx.report, ctx.on_event)
         if skipped is not None:
             return skipped
 
         result: TaskResult[Any] = TaskResult(spec=spec)
         result.started_at = datetime.now()
-        args, kwargs = build_call_args(spec, context)
+        args, kwargs = build_call_args(spec, task_ctx)
 
         _run_hooks(spec.hooks, "pre_run", spec)
-        _emit_running(on_event, spec)
+        _emit_running(ctx.on_event, spec)
 
         while True:
             result.attempts += 1
@@ -481,7 +490,7 @@ class SyncTaskRunner:
                 _mark_success(spec, result, value)
                 return result
             except Exception as exc:
-                if _handle_failure(spec, result, exc, layer_idx, on_event, spec.name, report):
+                if _handle_failure(spec, result, exc, layer_idx, ctx):
                     return result
                 wait = spec.retry.wait_seconds(result.attempts)
                 if wait > 0:
@@ -494,24 +503,23 @@ class AsyncTaskRunner:
     @staticmethod
     async def run(
         spec: TaskSpec[Any],
-        context: Mapping[str, Any],
+        task_ctx: Mapping[str, Any],
         layer_idx: int | None,
-        on_event: EventCallback | None = None,
-        report: RunReport | None = None,
+        ctx: _ExecContext,
         semaphore: asyncio.Semaphore | None = None,
     ) -> TaskResult[Any]:
-        skipped = _prepare_for_execution(spec, context, report, on_event)
+        skipped = _prepare_for_execution(spec, task_ctx, ctx.report, ctx.on_event)
         if skipped is not None:
             return skipped
 
         async def _inner() -> TaskResult[Any]:
             result: TaskResult[Any] = TaskResult(spec=spec)
             result.started_at = datetime.now()
-            args, kwargs = build_call_args(spec, context)
+            args, kwargs = build_call_args(spec, task_ctx)
             loop = asyncio.get_running_loop()
 
             _run_hooks(spec.hooks, "pre_run", spec)
-            _emit_running(on_event, spec)
+            _emit_running(ctx.on_event, spec)
 
             while True:
                 result.attempts += 1
@@ -520,7 +528,7 @@ class AsyncTaskRunner:
                     _mark_success(spec, result, value)
                     return result
                 except Exception as exc:
-                    if _handle_failure(spec, result, exc, layer_idx, on_event, spec.name, report):
+                    if _handle_failure(spec, result, exc, layer_idx, ctx):
                         return result
                     wait = spec.retry.wait_seconds(result.attempts)
                     if wait > 0:
@@ -594,10 +602,7 @@ def _submit_sync_task(
 def _filter_and_sort(
     layer: list[str],
     graph: Graph,
-    context: dict[str, Any],
-    report: RunReport,
-    backend: StateBackend,
-    on_event: EventCallback | None,
+    ctx: _ExecContext,
 ) -> tuple[list[str], dict[str, TaskSpec[Any]]]:
     """过滤掉已命中缓存的任务，按优先级排序返回待运行列表与 specs 映射。
 
@@ -611,36 +616,32 @@ def _filter_and_sort(
         spec = graph.resolved_spec(name)
         specs[name] = spec
         # 检查点恢复：已在 report 中的 SUCCESS 任务直接跳过
-        if name in report.results and report.results[name].status == TaskStatus.SUCCESS:
-            context[name] = report.results[name].value
+        if name in ctx.report.results and ctx.report.results[name].status == TaskStatus.SUCCESS:
+            ctx.context[name] = ctx.report.results[name].value
             continue
-        if not _apply_cached(name, spec, context, report, backend, on_event):
+        if not _apply_cached(spec, ctx):
             to_run.append(name)
     return _sort_by_priority(to_run, specs), specs
 
 
 def _store_result(
-    name: str,
     result: TaskResult[Any],
     spec: TaskSpec[Any],
     task_ctx: dict[str, Any],
-    context: dict[str, Any],
-    report: RunReport,
-    backend: StateBackend,
-    on_event: EventCallback | None,
+    ctx: _ExecContext,
 ) -> None:
     """存储任务结果到 context/report/backend 并触发事件。
 
     ``spec`` 与 ``task_ctx`` 由调用方在执行前已构建，直接复用避免重复
     ``resolved_spec`` / ``_build_context`` 调用。
     """
-    context[name] = result.value
+    ctx.context[spec.name] = result.value
     if result.status == TaskStatus.SUCCESS:
         # 快速路径：无 cache_key 时存储键即为任务名，避免函数调用开销。
         key = spec.name if spec.cache_key is None else spec.storage_key(task_ctx)
-        backend.save(key, result.value)
-    report.results[name] = result
-    _emit(on_event, result)
+        ctx.backend.save(key, result.value)
+    ctx.report.results[spec.name] = result
+    _emit(ctx.on_event, result)
 
 
 def _build_semaphores(
@@ -676,18 +677,15 @@ class SequentialLayerRunner:
     def execute(
         layer: list[str],
         graph: Graph,
-        context: dict[str, Any],
-        report: RunReport,
-        backend: StateBackend,
+        ctx: _ExecContext,
         layer_idx: int,
-        on_event: EventCallback | None,
     ) -> None:
-        to_run, specs = _filter_and_sort(layer, graph, context, report, backend, on_event)
+        to_run, specs = _filter_and_sort(layer, graph, ctx)
         for name in to_run:
             spec = specs[name]
-            task_ctx = _build_context(spec, context, report)
-            result = SyncTaskRunner.run(spec, task_ctx, layer_idx, on_event, report)
-            _store_result(name, result, spec, task_ctx, context, report, backend, on_event)
+            task_ctx = _build_context(spec, ctx.context, ctx.report)
+            result = SyncTaskRunner.run(spec, task_ctx, layer_idx, ctx)
+            _store_result(result, spec, task_ctx, ctx)
 
 
 class ThreadedLayerRunner:
@@ -697,29 +695,25 @@ class ThreadedLayerRunner:
     def execute(
         layer: list[str],
         graph: Graph,
-        context: dict[str, Any],
-        report: RunReport,
-        backend: StateBackend,
+        ctx: _ExecContext,
         layer_idx: int,
-        on_event: EventCallback | None,
         pool: concurrent.futures.ThreadPoolExecutor,
-        concurrency_limits: Mapping[str, int],
     ) -> None:
-        to_run, specs = _filter_and_sort(layer, graph, context, report, backend, on_event)
+        to_run, specs = _filter_and_sort(layer, graph, ctx)
         if not to_run:
             return
-        semaphores = _build_semaphores(to_run, specs, threading.Semaphore, concurrency_limits)
-        context_snapshot = dict(context)
+        semaphores = _build_semaphores(to_run, specs, threading.Semaphore, ctx.concurrency_limits)
+        context_snapshot = dict(ctx.context)
         lock = threading.Lock()
 
         def _run_threaded_task(name: str) -> tuple[dict[str, Any], TaskResult[Any]]:
             spec = specs[name]
-            task_ctx = _build_context(spec, context_snapshot, report)
+            task_ctx = _build_context(spec, context_snapshot, ctx.report)
             sem = _get_sem(semaphores, spec)
             if sem is not None:
                 sem.acquire()
             try:
-                return task_ctx, SyncTaskRunner.run(spec, task_ctx, layer_idx, on_event, report)
+                return task_ctx, SyncTaskRunner.run(spec, task_ctx, layer_idx, ctx)
             finally:
                 if sem is not None:
                     sem.release()
@@ -735,7 +729,7 @@ class ThreadedLayerRunner:
         finally:
             with lock:
                 for name, (task_ctx, result) in completed.items():
-                    _store_result(name, result, specs[name], task_ctx, context, report, backend, on_event)
+                    _store_result(result, specs[name], task_ctx, ctx)
 
 
 class AsyncLayerRunner:
@@ -745,29 +739,25 @@ class AsyncLayerRunner:
     async def execute(
         layer: list[str],
         graph: Graph,
-        context: dict[str, Any],
-        report: RunReport,
-        backend: StateBackend,
+        ctx: _ExecContext,
         layer_idx: int,
-        on_event: EventCallback | None,
-        concurrency_limits: Mapping[str, int],
     ) -> None:
-        to_run, specs = _filter_and_sort(layer, graph, context, report, backend, on_event)
+        to_run, specs = _filter_and_sort(layer, graph, ctx)
         if not to_run:
             return
-        semaphores = _build_semaphores(to_run, specs, asyncio.Semaphore, concurrency_limits)
-        context_snapshot = dict(context)
+        semaphores = _build_semaphores(to_run, specs, asyncio.Semaphore, ctx.concurrency_limits)
+        context_snapshot = dict(ctx.context)
 
         async def _run_async_task(name: str) -> tuple[dict[str, Any], TaskResult[Any]]:
             spec = specs[name]
-            task_ctx = _build_context(spec, context_snapshot, report)
+            task_ctx = _build_context(spec, context_snapshot, ctx.report)
             sem = _get_sem(semaphores, spec)
-            result = await AsyncTaskRunner.run(spec, task_ctx, layer_idx, on_event, report, sem)
+            result = await AsyncTaskRunner.run(spec, task_ctx, layer_idx, ctx, sem)
             return task_ctx, result
 
         results = await asyncio.gather(*[_run_async_task(name) for name in to_run])
         for name, (task_ctx, result) in zip(to_run, results, strict=True):
-            _store_result(name, result, specs[name], task_ctx, context, report, backend, on_event)
+            _store_result(result, specs[name], task_ctx, ctx)
 
 
 def _extract_dynamic_specs(result: TaskResult[Any], spec: TaskSpec[Any]) -> list[TaskSpec[Any]]:
@@ -806,16 +796,11 @@ class DependencyRunner:
     @staticmethod
     async def execute(
         graph: Graph,
-        context: dict[str, Any],
-        report: RunReport,
-        backend: StateBackend,
-        on_event: EventCallback | None,
-        concurrency_limits: Mapping[str, int],
-        cancel_event: threading.Event | CancelToken | None = None,
+        ctx: _ExecContext,
     ) -> None:
         all_names = list(graph.all_specs().keys())
         all_specs: dict[str, TaskSpec[Any]] = {name: graph.resolved_spec(name) for name in all_names}
-        semaphores = _build_semaphores(all_names, all_specs, asyncio.Semaphore, concurrency_limits)
+        semaphores = _build_semaphores(all_names, all_specs, asyncio.Semaphore, ctx.concurrency_limits)
 
         # 事件驱动调度：跟踪 completed / in_flight / remaining。
         # 仅当任务依赖全部完成时才创建 asyncio Task，按优先级降序调度。
@@ -825,10 +810,10 @@ class DependencyRunner:
 
         # 检查点恢复：已在 report 中的 SUCCESS 任务直接标记完成，跳过调度。
         for name in list(remaining):
-            if name in report.results and report.results[name].status == TaskStatus.SUCCESS:
+            if name in ctx.report.results and ctx.report.results[name].status == TaskStatus.SUCCESS:
                 completed.add(name)
                 remaining.discard(name)
-                context[name] = report.results[name].value
+                ctx.context[name] = ctx.report.results[name].value
 
         # 增量就绪集：用 in_degree 计数器 + dependents 反向邻接表替代每轮 O(N) 扫描。
         # 每轮调度开销从 O(N*D) 降至 O(D_out)，大图（10k+ 任务）显著加速。
@@ -850,7 +835,7 @@ class DependencyRunner:
             all_specs[final_spec.name] = graph.resolved_spec(final_spec.name)
             remaining.add(final_spec.name)
             if final_spec.concurrency_key and final_spec.concurrency_key not in semaphores:
-                limit = concurrency_limits.get(final_spec.concurrency_key, 1)
+                limit = ctx.concurrency_limits.get(final_spec.concurrency_key, 1)
                 semaphores[final_spec.concurrency_key] = asyncio.Semaphore(limit)
             # 计算新任务的 in_degree（spawner 可能尚未标记 completed，计入 unsatisfied）
             deps = (*final_spec.depends_on, *final_spec.soft_depends_on)
@@ -867,34 +852,34 @@ class DependencyRunner:
             spec = all_specs[name]
 
             # 取消检查：依赖完成后、执行前检查；已取消则标记 SKIPPED
-            if _is_cancelled(cancel_event) and name not in report.results:
+            if _is_cancelled(ctx.cancel_event) and name not in ctx.report.results:
                 result: TaskResult[Any] = TaskResult(
                     spec=spec,
                     status=TaskStatus.SKIPPED,
                     finished_at=datetime.now(),
                     reason="执行被取消",
                 )
-                report.results[name] = result
-                _emit(on_event, result)
+                ctx.report.results[name] = result
+                _emit(ctx.on_event, result)
                 return result
 
-            task_ctx = _build_context(spec, context, report)
-            if _apply_cached(name, spec, context, report, backend, on_event):
-                return report.results[name]
+            task_ctx = _build_context(spec, ctx.context, ctx.report)
+            if _apply_cached(spec, ctx):
+                return ctx.report.results[spec.name]
 
             sem = _get_sem(semaphores, spec)
-            result = await AsyncTaskRunner.run(spec, task_ctx, None, on_event, report, sem)
+            result = await AsyncTaskRunner.run(spec, task_ctx, None, ctx, sem)
 
             # 动态任务生成：提取 specs，替换 result.value 为生成任务名列表
             spawned = _extract_dynamic_specs(result, spec)
             if spawned:
                 spawned_names = [s.name for s in spawned]
                 result = dc_replace(result, value=spawned_names)
-                context[name] = result.value
-                report.results[name] = result
-                _emit(on_event, result)
+                ctx.context[name] = result.value
+                ctx.report.results[name] = result
+                _emit(ctx.on_event, result)
             else:
-                _store_result(name, result, spec, task_ctx, context, report, backend, on_event)
+                _store_result(result, spec, task_ctx, ctx)
 
             # 注册动态生成的 specs，接入增量调度结构
             for raw_spec in spawned:
@@ -1018,13 +1003,8 @@ def _apply_subgraph_filter(
 def _dispatch_strategy(
     strategy: Strategy,
     graph: Graph,
-    context: dict[str, Any],
-    report: RunReport,
-    backend: StateBackend,
-    effective_callback: EventCallback | None,
+    ctx: _ExecContext,
     max_workers: int | None,
-    limits: Mapping[str, int],
-    cancel_event: threading.Event | CancelToken | None,
 ) -> None:
     """按策略派发执行，将策略分支从 :func:`run` 中提取出来控制 PLR0912。
 
@@ -1038,15 +1018,15 @@ def _dispatch_strategy(
         )
     if strategy == "sequential":
         layers = graph.layers()
-        _drive_sequential(graph, layers, context, report, backend, effective_callback, cancel_event)
+        _drive_sequential(graph, layers, ctx)
     elif strategy == "thread":
         layers = graph.layers()
-        _drive_threaded(graph, layers, context, report, backend, effective_callback, max_workers, limits, cancel_event)
+        _drive_threaded(graph, layers, ctx, max_workers)
     elif strategy == "async":
         layers = graph.layers()
-        asyncio.run(_async_drive(graph, layers, context, report, backend, effective_callback, limits, cancel_event))
+        asyncio.run(_async_drive(graph, layers, ctx))
     elif strategy == "dependency":
-        asyncio.run(DependencyRunner.execute(graph, context, report, backend, effective_callback, limits, cancel_event))
+        asyncio.run(DependencyRunner.execute(graph, ctx))
     else:  # pragma: no cover - Strategy Literal 已穷尽所有取值
         raise ValueError(f"Unknown strategy: {strategy!r}")
 
@@ -1105,7 +1085,7 @@ def _apply_resume(
         logger.info("从检查点恢复 %d 个任务", restored, extra={"restored": restored})
 
 
-def run(
+def run(  # noqa: PLR0913
     graph: Graph,
     strategy: Strategy = "dependency",
     *,
@@ -1221,15 +1201,24 @@ def run(
     internal_cancel = CancelToken()
     effective_cancel: CancelToken | threading.Event | None = cancel_event
 
+    # 打包执行上下文：将 context/report/backend/on_event/concurrency_limits/cancel_event
+    # 捆绑为单一参数传递给调用链，使每个内部函数参数数 ≤5（PLR0913）。
+    ctx = _ExecContext(
+        context=context,
+        report=report,
+        backend=backend,
+        on_event=effective_callback,
+        concurrency_limits=limits,
+        cancel_event=effective_cancel,
+    )
+
     # backend.batch()：将每任务一次落盘降为整次运行一次（JSONBackend）；
     # MemoryBackend 为 no-op。即使中途抛出 TaskFailedError，batch 退出时
     # 仍会 flush 一次，保留已成功任务的结果以便断点续跑。
     try:
         with backend.batch():
             try:
-                _dispatch_strategy(
-                    strategy, graph, context, report, backend, effective_callback, max_workers, limits, effective_cancel
-                )
+                _dispatch_strategy(strategy, graph, ctx, max_workers)
             except TaskFailedError:
                 report.success = False
                 raise
@@ -1267,7 +1256,7 @@ def run(
     return report
 
 
-def run_iter(
+def run_iter(  # noqa: PLR0913
     graph: Graph,
     strategy: Strategy = "dependency",
     *,
@@ -1369,53 +1358,39 @@ def _print_dry_run(graph: Graph, layers: list[list[str]]) -> None:
 def _drive_sequential(
     graph: Graph,
     layers: list[list[str]],
-    context: dict[str, Any],
-    report: RunReport,
-    backend: StateBackend,
-    on_event: EventCallback | None,
-    cancel_event: threading.Event | CancelToken | None = None,
+    ctx: _ExecContext,
 ) -> None:
     for idx, layer in enumerate(layers, 1):
-        if _is_cancelled(cancel_event):
+        if _is_cancelled(ctx.cancel_event):
             return
-        SequentialLayerRunner.execute(layer, graph, context, report, backend, idx, on_event)
+        SequentialLayerRunner.execute(layer, graph, ctx, idx)
 
 
 def _drive_threaded(
     graph: Graph,
     layers: list[list[str]],
-    context: dict[str, Any],
-    report: RunReport,
-    backend: StateBackend,
-    on_event: EventCallback | None,
+    ctx: _ExecContext,
     max_workers: int | None,
-    concurrency_limits: Mapping[str, int],
-    cancel_event: threading.Event | CancelToken | None = None,
 ) -> None:
     # 线程池在整个 run() 内复用，避免逐层创建/销毁线程的开销。
     max_layer_size = max((len(layer) for layer in layers), default=1)
     pool_workers = max_workers or max(1, min(32, max_layer_size))
     with concurrent.futures.ThreadPoolExecutor(max_workers=pool_workers) as pool:
         for idx, layer in enumerate(layers, 1):
-            if _is_cancelled(cancel_event):
+            if _is_cancelled(ctx.cancel_event):
                 return
-            ThreadedLayerRunner.execute(layer, graph, context, report, backend, idx, on_event, pool, concurrency_limits)
+            ThreadedLayerRunner.execute(layer, graph, ctx, idx, pool)
 
 
 async def _async_drive(
     graph: Graph,
     layers: list[list[str]],
-    context: dict[str, Any],
-    report: RunReport,
-    backend: StateBackend,
-    on_event: EventCallback | None,
-    concurrency_limits: Mapping[str, int],
-    cancel_event: threading.Event | CancelToken | None = None,
+    ctx: _ExecContext,
 ) -> None:
     for idx, layer in enumerate(layers, 1):
-        if _is_cancelled(cancel_event):
+        if _is_cancelled(ctx.cancel_event):
             return
-        await AsyncLayerRunner.execute(layer, graph, context, report, backend, idx, on_event, concurrency_limits)
+        await AsyncLayerRunner.execute(layer, graph, ctx, idx)
 
 
 def _mark_remaining_skipped(

src/pyflowx/fileops.py

@@ -70,7 +70,7 @@ __all__ = [
 # ---------------------------------------------------------------------- #
 # 查找与遍历
 # ---------------------------------------------------------------------- #
-def find(
+def find(  # noqa: PLR0913
     root: str | Path,
     pattern: str = "*",
     *,

src/pyflowx/ops/dev/autofmt.py

@@ -69,7 +69,7 @@ def format_with_ruff(target: Path, fix: bool = True) -> None:
     if fix:
         cmd.append("--fix")
 
-    subprocess.run(cmd, check=True)
+    px.sh(cmd, label="ruff format")
     print(f"ruff format 完成: {target}")
 
 
@@ -87,7 +87,7 @@ def lint_with_ruff(target: Path, fix: bool = True) -> None:
     if fix:
         cmd.extend(["--fix", "--unsafe-fixes"])
 
-    subprocess.run(cmd, check=True)
+    px.sh(cmd, label="ruff check")
     print(f"ruff check 完成: {target}")
 
 
@@ -223,6 +223,6 @@ def format_all(root_dir: Path) -> None:
     root_dir : Path
         根目录
     """
-    subprocess.run(["ruff", "format", str(root_dir)], check=True)
-    subprocess.run(["ruff", "check", "--fix", "--unsafe-fixes", str(root_dir)], check=True)
+    px.sh(["ruff", "format", str(root_dir)], label="ruff format")
+    px.sh(["ruff", "check", "--fix", "--unsafe-fixes", str(root_dir)], label="ruff check")
     print(f"格式化完成: {root_dir}")

src/pyflowx/ops/dev/bumpversion.py

@@ -15,7 +15,6 @@
 from __future__ import annotations
 
 import re
-import subprocess
 from pathlib import Path
 from typing import Literal
 
@@ -221,12 +220,12 @@ def bump_project_version(part: BumpVersionType = "patch", no_tag: bool = False)
 
     # 阶段 3: git add (按文件名) + commit + tag
     relative_files = [str(file.relative_to(cwd)) for file in sorted(all_files)]
-    subprocess.run(["git", "add", *relative_files], check=True)
-    subprocess.run(["git", "commit", "-m", f"bump version to {new_version}"], check=True)
+    px.sh(["git", "add", *relative_files], label="git add")
+    px.sh(["git", "commit", "-m", f"bump version to {new_version}"], label="git commit")
 
     if not no_tag:
         tag_name = f"v{new_version}"
-        subprocess.run(["git", "tag", "-a", tag_name, "-m", f"Release {tag_name}"], check=True)
+        px.sh(["git", "tag", "-a", tag_name, "-m", f"Release {tag_name}"], label="git tag")
         print(f"已创建标签: {tag_name}")
 
     return new_version

src/pyflowx/ops/dev/gittool.py

@@ -69,8 +69,8 @@ def git_add_commit(message: str = "chore: update") -> None:
     if not has_files():
         print("没有文件需要提交")
         return
-    subprocess.run(["git", "add", "."], check=True)
-    subprocess.run(["git", "commit", "-m", message], check=True)
+    px.sh(["git", "add", "."], label="git add")
+    px.sh(["git", "commit", "-m", message], label="git commit")
 
 
 @px.tool("gittool", subcommand="i", help="初始化并提交")
@@ -83,10 +83,10 @@ def git_init_add_commit(message: str = "init commit") -> None:
         提交信息 (默认: ``init commit``)
     """
     if not_has_git_repo():
-        subprocess.run(["git", "init"], check=True)
+        px.sh(["git", "init"], label="git init")
     if has_files():
-        subprocess.run(["git", "add", "."], check=True)
-        subprocess.run(["git", "commit", "-m", message], check=True)
+        px.sh(["git", "add", "."], label="git add")
+        px.sh(["git", "commit", "-m", message], label="git commit")
     else:
         print("没有文件需要提交")
 

src/pyflowx/ops/dev/lscalc.py

@@ -49,13 +49,8 @@ def run_ls_dyna(input_file: str, ncpu: int = _DEFAULT_NCPU) -> None:
         return
 
     cmd = get_ls_dyna_command(input_file, ncpu)
-    try:
-        subprocess.run(cmd, check=True)
+    if px.sh(cmd, label="LS-DYNA 计算") is not None:
         print(f"LS-DYNA 计算完成: {input_file}")
-    except FileNotFoundError:
-        print("未找到 ls-dyna_mpp 命令")
-    except subprocess.CalledProcessError as e:
-        print(f"LS-DYNA 计算失败: {e}")
 
 
 @px.tool("lscalc", subcommand="mpi", help="运行 LS-DYNA MPI 计算")
@@ -75,37 +70,29 @@ def run_ls_dyna_mpi(input_file: str, ncpu: int = _DEFAULT_NCPU) -> None:
         return
 
     cmd = ["mpirun", "-np", str(ncpu), "ls-dyna_mpp", f"i={input_file}"]
-    try:
-        subprocess.run(cmd, check=True)
+    if px.sh(cmd, label="LS-DYNA MPI 计算") is not None:
         print(f"LS-DYNA MPI 计算完成: {input_file}")
-    except FileNotFoundError:
-        print("未找到 mpirun 或 ls-dyna_mpp 命令")
-    except subprocess.CalledProcessError as e:
-        print(f"LS-DYNA MPI 计算失败: {e}")
 
 
 @px.tool("lscalc", subcommand="status", help="检查 LS-DYNA 进程状态")
 def check_ls_dyna_status() -> None:
     """检查 LS-DYNA 进程状态."""
-    try:
-        if Constants.IS_WINDOWS:
-            result = subprocess.run(
-                ["tasklist", "/fi", "imagename eq ls-dyna_mpp.exe"],
-                capture_output=True,
-                text=True,
-                check=True,
-            )
+    if Constants.IS_WINDOWS:
+        result = px.sh(
+            ["tasklist", "/fi", "imagename eq ls-dyna_mpp.exe"],
+            capture=True,
+            label="tasklist",
+        )
+        if result is not None:
             print(result.stdout)
+    else:
+        result = subprocess.run(
+            ["pgrep", "-f", "ls-dyna"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if result.stdout.strip():
+            print(f"运行中的 LS-DYNA 进程 PID: {result.stdout.strip()}")
         else:
-            result = subprocess.run(
-                ["pgrep", "-f", "ls-dyna"],
-                capture_output=True,
-                text=True,
-                check=False,
-            )
-            if result.stdout.strip():
-                print(f"运行中的 LS-DYNA 进程 PID: {result.stdout.strip()}")
-            else:
-                print("没有运行中的 LS-DYNA 进程")
-    except subprocess.CalledProcessError as e:
-        print(f"检查进程状态失败: {e}")
+            print("没有运行中的 LS-DYNA 进程")

src/pyflowx/ops/dev/packtool.py

@@ -7,7 +7,6 @@ from __future__ import annotations
 
 import platform
 import shutil
-import subprocess
 import urllib.request
 import zipfile
 from pathlib import Path
@@ -124,7 +123,7 @@ def pack_dependencies(lib_dir: Path = Path("libs"), dependencies: list[str] | No
     ]
     cmd.extend(dependencies)
 
-    subprocess.run(cmd, check=True)
+    px.sh(cmd, label="pip install")
     print(f"依赖打包完成: {lib_dir}")
 
 
@@ -150,7 +149,7 @@ def pack_wheel(project_dir: Path = Path(), output_dir: Path = Path("dist")) -> N
         str(project_dir),
     ]
 
-    subprocess.run(cmd, check=True)
+    px.sh(cmd, label="pip wheel")
     print(f"Wheel 打包完成: {output_dir}")
 
 

src/pyflowx/ops/dev/piptool.py

@@ -6,7 +6,6 @@
 from __future__ import annotations
 
 import fnmatch
-import subprocess
 from pathlib import Path
 
 import pyflowx as px
@@ -42,20 +41,14 @@ _PROTECTED_PACKAGES: frozenset[str] = frozenset(
 
 def _get_installed_packages() -> list[str]:
     """获取当前环境中所有已安装的包名."""
-    try:
-        result = subprocess.run(
-            ["pip", "list", "--format=freeze"],
-            capture_output=True,
-            text=True,
-            check=True,
-        )
-        packages: list[str] = []
-        for line in result.stdout.strip().split("\n"):
-            if line and "==" in line:
-                pkg_name = line.split("==")[0].strip()
-                packages.append(pkg_name)
-    except (subprocess.SubprocessError, OSError):
+    result = px.sh(["pip", "list", "--format=freeze"], capture=True, label="pip list")
+    if result is None:
         return []
+    packages: list[str] = []
+    for line in result.stdout.strip().split("\n"):
+        if line and "==" in line:
+            pkg_name = line.split("==")[0].strip()
+            packages.append(pkg_name)
     return packages
 
 
@@ -92,7 +85,7 @@ def pip_install(packages: list[str]) -> None:
     packages : list[str]
         包名列表
     """
-    subprocess.run(["pip", "install", *packages], check=True)
+    px.sh(["pip", "install", *packages], label="pip install")
     print(f"安装完成: {', '.join(packages)}")
 
 
@@ -114,7 +107,7 @@ def pip_uninstall(packages: list[str]) -> None:
     if not packages_to_uninstall:
         return
 
-    subprocess.run(["pip", "uninstall", "-y", *packages_to_uninstall], check=True)
+    px.sh(["pip", "uninstall", "-y", *packages_to_uninstall], label="pip uninstall")
 
 
 @px.tool("piptool", subcommand="r", help="重装包")
@@ -133,10 +126,10 @@ def pip_reinstall(packages: list[str], offline: bool = False) -> None:
         print("所有指定的包均为受保护包, 跳过重装")
         return
 
-    subprocess.run(["pip", "uninstall", "-y", *safe_ps], check=True)
+    px.sh(["pip", "uninstall", "-y", *safe_ps], label="pip uninstall")
 
     options = ["--no-index", "--find-links", "."] if offline else []
-    subprocess.run(["pip", "install", *options, *safe_ps], check=True)
+    px.sh(["pip", "install", *options, *safe_ps], label="pip install")
 
 
 @px.tool("piptool", subcommand="d", help="下载包")
@@ -151,27 +144,20 @@ def pip_download(packages: list[str], offline: bool = False) -> None:
         离线模式
     """
     options = ["--no-index", "--find-links", "."] if offline else []
-    subprocess.run(
-        ["pip", "download", *packages, *options, "-d", PACKAGE_DIR],
-        check=True,
-    )
+    px.sh(["pip", "download", *packages, *options, "-d", PACKAGE_DIR], label="pip download")
 
 
 @px.tool("piptool", subcommand="up", help="升级 pip")
 def pip_upgrade() -> None:
     """升级 pip 到最新版本."""
-    subprocess.run(["python", "-m", "pip", "install", "--upgrade", "pip"], check=True)
+    px.sh(["python", "-m", "pip", "install", "--upgrade", "pip"], label="pip 升级")
     print("pip 升级完成")
 
 
 @px.tool("piptool", subcommand="f", help="导出依赖")
 def pip_freeze() -> None:
     """冻结依赖到 requirements.txt."""
-    result = subprocess.run(
-        ["pip", "freeze", "--exclude-editable"],
-        capture_output=True,
-        text=True,
-        check=True,
-    )
-    Path(REQUIREMENTS_FILE).write_text(result.stdout)
+    result = px.sh(["pip", "freeze", "--exclude-editable"], capture=True, label="pip freeze")
+    if result is not None:
+        Path(REQUIREMENTS_FILE).write_text(result.stdout)
     print(f"依赖已导出到 {REQUIREMENTS_FILE}")

src/pyflowx/ops/files/imagetool.py

@@ -99,7 +99,7 @@ def image_resize(
 
 
 @px.tool("imagetool", subcommand="c", help="裁剪图片")
-def image_crop(
+def image_crop(  # noqa: PLR0913
     input_path: Path,
     output_path: Path,
     left: int,
@@ -213,7 +213,7 @@ def image_convert(
 
 
 @px.tool("imagetool", subcommand="wm", help="添加文字水印")
-def image_watermark(
+def image_watermark(  # noqa: PLR0913
     input_path: Path,
     output_path: Path,
     text: str,

src/pyflowx/ops/files/screenshot.py

@@ -59,9 +59,9 @@ $bitmap.Save('{output_path.as_posix()}')
 $graphics.Dispose()
 $bitmap.Dispose()
 """
-        subprocess.run(["powershell", "-Command", ps_script], check=True)
+        px.sh(["powershell", "-Command", ps_script], label="截图")
     elif Constants.IS_MACOS:
-        subprocess.run(["screencapture", "-x", str(output_path)], check=True)
+        px.sh(["screencapture", "-x", str(output_path)], label="截图")
     else:
         try:
             subprocess.run(["gnome-screenshot", "-f", str(output_path)], check=True)
@@ -104,9 +104,9 @@ $bitmap.Save('{output_path.as_posix()}')
 $graphics.Dispose()
 $bitmap.Dispose()
 """
-        subprocess.run(["powershell", "-Command", ps_script], check=True)
+        px.sh(["powershell", "-Command", ps_script], label="截图")
     elif Constants.IS_MACOS:
-        subprocess.run(["screencapture", "-i", str(output_path)], check=True)
+        px.sh(["screencapture", "-i", str(output_path)], label="截图")
     else:
         try:
             subprocess.run(["gnome-screenshot", "-a", "-f", str(output_path)], check=True)

src/pyflowx/ops/infra/dockercmd.py

@@ -6,7 +6,6 @@
 from __future__ import annotations
 
 import getpass
-import subprocess
 
 import pyflowx as px
 
@@ -23,15 +22,12 @@ def docker_login_tencent(username: str = "") -> None:
         Docker 用户名 (为空时由 docker 交互式提示输入)
     """
     user = username or getpass.getuser()
-    try:
-        subprocess.run(
+    if (
+        px.sh(
             ["docker", "login", "--username", user, _DOCKER_MIRROR_TENCENT],
-            check=True,
+            label="docker login",
         )
-    except subprocess.CalledProcessError as e:
-        print(f"docker login 失败 (returncode={e.returncode}): {e}")
-        return
-    except FileNotFoundError:
-        print("docker 命令未找到，请确认 Docker 已安装并在 PATH 中")
+        is None
+    ):
         return
     print(f"已登录腾讯云镜像仓库 (用户: {user})")

src/pyflowx/ops/infra/msdownload.py

@@ -5,7 +5,6 @@
 
 from __future__ import annotations
 
-import subprocess
 from pathlib import Path
 
 import pyflowx as px
@@ -38,9 +37,4 @@ def msdownload_run(name: str, target_type: str = "model", download_dir: str | No
 
     cmd = ["uvx", "modelscope", "download", f"--{target_type}", name, "--local_dir", str(out_dir)]
     print(f"下载 {target_type}: {name} -> {out_dir}")
-    try:
-        subprocess.run(cmd, check=True)
-    except subprocess.CalledProcessError as e:
-        print(f"下载失败 (returncode={e.returncode}): {e}")
-    except FileNotFoundError:
-        print("uvx 命令未找到，请确认 uv 已安装并在 PATH 中")
+    px.sh(cmd, label="下载")

src/pyflowx/ops/infra/sglang.py

@@ -32,7 +32,7 @@ def install_sglang() -> None:
 
 
 @px.tool("sglang", subcommand="run", help="启动 SGLang 服务", needs=["install"])
-def run_sglang(
+def run_sglang(  # noqa: PLR0913
     model: str = "~/.models/Qwen2.5-Coder-32B-Instruct-AWQ",
     port: int = 8000,
     ctx_len: int = 32768,

src/pyflowx/ops/infra/sshcopyid.py

@@ -15,7 +15,7 @@ __all__ = ["ssh_copy_id"]
 
 
 @px.tool("sshcopyid", help="部署 SSH 公钥到远程服务器")
-def ssh_copy_id(
+def ssh_copy_id(  # noqa: PLR0913
     hostname: str,
     username: str,
     password: str,

src/pyflowx/storage.py

@@ -19,7 +19,7 @@ import threading
 import time
 from abc import ABC, abstractmethod
 from collections import OrderedDict
-from collections.abc import Iterator, Mapping
+from collections.abc import Generator, Iterator, Mapping
 from contextlib import AbstractContextManager, contextmanager, nullcontext
 from pathlib import Path
 from typing import Any
@@ -328,7 +328,7 @@ class JSONBackend(_TTLStateBackendMixin):
 
     @override
     @contextmanager
-    def batch(self) -> Iterator[None]:
+    def batch(self) -> Generator[None, None, None]:
         """进入批量模式：``save`` 暂不落盘，退出时统一 flush 一次。
 
         将整次运行 N 个任务的 N 次全量落盘降为 1 次。
@@ -462,7 +462,7 @@ class SQLiteBackend(_TTLStateBackendMixin):
 
     @override
     @contextmanager
-    def batch(self) -> Iterator[None]:
+    def batch(self) -> Generator[None, None, None]:
         """进入批量模式：``save`` 暂不 commit，退出时统一 commit 一次。
 
         将整次运行 N 个任务的 N 次 commit 降为 1 次。

src/pyflowx/task.py

@@ -513,7 +513,7 @@ def _task_noop() -> None:
     return None
 
 
-def task(
+def task(  # noqa: PLR0913
     fn: TaskFn[Any] | None = None,
     *,
     cmd: TaskCmd | None = None,

src/pyflowx/tools.py

@@ -114,7 +114,7 @@ _TOOL_REGISTRY: dict[str, dict[str | None, ToolSpec]] = {}
 # ---------------------------------------------------------------------- #
 # @tool 装饰器 + 注册表
 # ---------------------------------------------------------------------- #
-def tool(
+def tool(  # noqa: PLR0913
     name: str,
     *,
     subcommand: str | None = None,

tests/cli/test_autofmt.py

@@ -16,14 +16,14 @@ class TestFormatWithRuff:
 
     def test_format_with_ruff(self, tmp_path: Path) -> None:
         """Should format with ruff."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.format_with_ruff(tmp_path, fix=True)
             assert mock_run.called
 
     def test_format_with_ruff_no_fix(self, tmp_path: Path) -> None:
         """Should format with ruff without fix."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.format_with_ruff(tmp_path, fix=False)
             # Should not include --fix flag
@@ -39,14 +39,14 @@ class TestLintWithRuff:
 
     def test_lint_with_ruff(self, tmp_path: Path) -> None:
         """Should lint with ruff."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.lint_with_ruff(tmp_path, fix=True)
             assert mock_run.called
 
     def test_lint_with_ruff_no_fix(self, tmp_path: Path) -> None:
         """Should lint with ruff without fix."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.lint_with_ruff(tmp_path, fix=False)
             # Should not include --fix flag
@@ -204,14 +204,14 @@ class TestFormatAll:
 
     def test_format_all_runs_ruff_format(self, tmp_path: Path) -> None:
         """Should run ruff format."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.format_all(tmp_path)
             assert mock_run.called
 
     def test_format_all_runs_ruff_check(self, tmp_path: Path) -> None:
         """Should run ruff check."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.format_all(tmp_path)
             # Should call ruff format and ruff check

tests/cli/test_bumpversion.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 import subprocess
 from pathlib import Path
 
@@ -9,6 +10,13 @@ import pytest
 
 from pyflowx.ops.dev import bumpversion
 
+_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]")
+
+
+def _strip_ansi(text: str) -> str:
+    """移除 ANSI 转义码，使断言在 CI(富终端) 与本地环境一致."""
+    return _ANSI_RE.sub("", text)
+
 
 @pytest.fixture(autouse=True)
 def auto_use_tmp_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -104,15 +112,15 @@ class TestBumpProjectVersion:
     """Test bump_project_version function."""
 
     @staticmethod
-    def _mock_subprocess(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
-        """Mock subprocess.run, 返回调用记录列表."""
+    def _mock_sh(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
+        """Mock px.sh, 返回调用记录列表."""
         calls: list[list[str]] = []
 
-        def fake_run(cmd: list[str], **_kwargs: object) -> subprocess.CompletedProcess[bytes]:
+        def fake_sh(cmd: list[str], **_kwargs: object) -> subprocess.CompletedProcess[bytes]:
             calls.append(cmd)
             return subprocess.CompletedProcess(cmd, 0, b"", b"")
 
-        monkeypatch.setattr(subprocess, "run", fake_run)
+        monkeypatch.setattr("pyflowx.sh", fake_sh)
         return calls
 
     def test_unsynced_files_synchronized(
@@ -129,7 +137,7 @@ class TestBumpProjectVersion:
         pyproj = tmp_path / "pyproject.toml"
         pyproj.write_text('version = "0.3.5"', encoding="utf-8")
 
-        calls = self._mock_subprocess(monkeypatch)
+        calls = self._mock_sh(monkeypatch)
 
         result = bumpversion.bump_project_version("patch")
 
@@ -163,7 +171,7 @@ class TestBumpProjectVersion:
         """有文件但所有文件都无版本号返回 None."""
         f = tmp_path / "__init__.py"
         f.write_text("# no version here", encoding="utf-8")
-        self._mock_subprocess(monkeypatch)
+        self._mock_sh(monkeypatch)
 
         assert bumpversion.bump_project_version("patch") is None
         assert "未能从任何文件读取版本号" in capsys.readouterr().out
@@ -173,7 +181,7 @@ class TestBumpProjectVersion:
         pyproj = tmp_path / "pyproject.toml"
         pyproj.write_text('version = "1.0.0"', encoding="utf-8")
 
-        calls = self._mock_subprocess(monkeypatch)
+        calls = self._mock_sh(monkeypatch)
 
         assert bumpversion.bump_project_version("patch", no_tag=True) == "1.0.1"
         assert not any(c[:2] == ["git", "tag"] for c in calls)
@@ -186,7 +194,57 @@ class TestBumpProjectVersion:
         pyproj = tmp_path / "pyproject.toml"
         pyproj.write_text('version = "1.0.0"', encoding="utf-8")
 
-        self._mock_subprocess(monkeypatch)
+        self._mock_sh(monkeypatch)
 
         assert bumpversion.bump_project_version("patch") == "1.0.1"
         assert venv_init.read_text(encoding="utf-8") == '__version__ = "0.1.0"'
+
+
+# ---------------------------------------------------------------------- #
+# CLI 调度测试 (pf bumpversion -h / pf bumpversion --part X --dry-run)
+# ---------------------------------------------------------------------- #
+class TestBumpversionCliDispatch:
+    """``pf bumpversion`` CLI 调度测试 (不执行真实 bump)."""
+
+    def test_bumpversion_h_shows_help(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """pf bumpversion -h 应返回 0 并显示 --part 选项."""
+        from pyflowx.cli.pf import PfApp
+
+        app = PfApp(["bumpversion", "-h"])
+        rc = app.run()
+        assert rc == 0
+        out = _strip_ansi(capsys.readouterr().out)
+        assert "--part" in out
+        assert "patch" in out
+
+    def test_bumpversion_part_patch_dry_run(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """pf bumpversion --part patch --dry-run 应打印执行计划, 不执行 bump."""
+        from pyflowx.cli.pf import PfApp
+
+        app = PfApp(["bumpversion", "--part", "patch", "--dry-run"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Dry run" in out
+        assert "bumpversion" in out
+
+    def test_bumpversion_part_minor_dry_run(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """pf bumpversion --part minor --dry-run 应打印执行计划."""
+        from pyflowx.cli.pf import PfApp
+
+        app = PfApp(["bumpversion", "--part", "minor", "--dry-run"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Dry run" in out
+
+    def test_bumpversion_default_part_is_patch(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """pf bumpversion --dry-run (无 --part) 默认使用 patch."""
+        from pyflowx.cli.pf import PfApp
+
+        app = PfApp(["bumpversion", "--dry-run"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Dry run" in out
+        assert "patch" in out

tests/cli/test_envdev.py

@@ -178,7 +178,8 @@ class TestDockerLoginTencent:
         monkeypatch.setattr(subprocess, "run", lambda *_, **__: (_ for _ in ()).throw(err))
         docker_login_tencent("myuser")
         captured = capsys.readouterr()
-        assert "docker login 失败" in captured.out
+        assert "docker login" in captured.out
+        assert "失败" in captured.out
         assert "returncode=1" in captured.out
         assert "已登录" not in captured.out
 
@@ -189,7 +190,7 @@ class TestDockerLoginTencent:
         monkeypatch.setattr(subprocess, "run", lambda *_, **__: (_ for _ in ()).throw(FileNotFoundError()))
         docker_login_tencent("myuser")
         captured = capsys.readouterr()
-        assert "docker 命令未找到" in captured.out
+        assert "命令未找到" in captured.out
         assert "已登录" not in captured.out
 
 

tests/cli/test_llm.py

@@ -90,7 +90,7 @@ class TestMsdownloadRun:
         monkeypatch.setattr(subprocess, "run", lambda *_, **__: (_ for _ in ()).throw(FileNotFoundError()))
         msdownload_run("foo/bar")
         captured = capsys.readouterr()
-        assert "uvx 命令未找到" in captured.out
+        assert "命令未找到" in captured.out
 
 
 # ---------------------------------------------------------------------- #

tests/cli/test_lscalc.py

@@ -43,7 +43,7 @@ class TestRunLsDyna:
         input_file = tmp_path / "input.k"
         input_file.write_text("LS-DYNA input")
 
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             system.run_ls_dyna(str(input_file), ncpu=4)
             assert mock_run.called
@@ -60,7 +60,7 @@ class TestRunLsDyna:
         input_file = tmp_path / "input.k"
         input_file.write_text("LS-DYNA input")
 
-        with patch("subprocess.run", side_effect=FileNotFoundError):
+        with patch("pyflowx.sh", return_value=None):
             system.run_ls_dyna(str(input_file), ncpu=4)
             # Should print error message
 
@@ -76,7 +76,7 @@ class TestRunLsDynaMpi:
         input_file = tmp_path / "input.k"
         input_file.write_text("LS-DYNA input")
 
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             system.run_ls_dyna_mpi(str(input_file), ncpu=8)
             assert mock_run.called
@@ -97,7 +97,7 @@ class TestCheckLsDynaStatus:
 
     def test_check_ls_dyna_status_windows(self) -> None:
         """Should check LS-DYNA status on Windows."""
-        with patch.object(Constants, "IS_WINDOWS", True), patch("subprocess.run") as mock_run:
+        with patch.object(Constants, "IS_WINDOWS", True), patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(stdout="ls-dyna_mpp.exe", returncode=0)
             system.check_ls_dyna_status()
             assert mock_run.called

tests/cli/test_packtool.py

@@ -67,7 +67,7 @@ class TestPackDependencies:
         """Should pack dependencies."""
         lib_dir = tmp_path / "libs"
 
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             system.pack_dependencies(lib_dir, ["numpy", "pandas"])
             assert mock_run.called
@@ -86,7 +86,7 @@ class TestPackWheel:
         (project_dir / "pyproject.toml").write_text("[project]\nname = 'test'")
         output_dir = tmp_path / "dist"
 
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             system.pack_wheel(project_dir, output_dir)
             assert mock_run.called

tests/cli/test_piptool.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import subprocess
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -17,7 +16,7 @@ class TestGetInstalledPackages:
 
     def test_get_installed_packages_success(self) -> None:
         """Should get installed packages."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(stdout="numpy==1.0.0\npandas==2.0.0\n", returncode=0)
             result = dev._get_installed_packages()
             assert "numpy" in result
@@ -25,20 +24,20 @@ class TestGetInstalledPackages:
 
     def test_get_installed_packages_empty(self) -> None:
         """Should handle empty output."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(stdout="", returncode=0)
             result = dev._get_installed_packages()
             assert result == []
 
     def test_get_installed_packages_error(self) -> None:
         """Should handle subprocess error."""
-        with patch("subprocess.run", side_effect=subprocess.SubprocessError):
+        with patch("pyflowx.sh", return_value=None):
             result = dev._get_installed_packages()
             assert result == []
 
     def test_get_installed_packages_oserror(self) -> None:
         """Should handle OSError."""
-        with patch("subprocess.run", side_effect=OSError):
+        with patch("pyflowx.sh", return_value=None):
             result = dev._get_installed_packages()
             assert result == []
 
@@ -106,14 +105,14 @@ class TestPipUninstall:
 
     def test_pip_uninstall_single_package(self) -> None:
         """Should uninstall single package."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.pip_uninstall(["numpy"])
             assert mock_run.called
 
     def test_pip_uninstall_multiple_packages(self) -> None:
         """Should uninstall multiple packages."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.pip_uninstall(["numpy", "pandas", "scipy"])
             # Should call pip uninstall
@@ -123,7 +122,7 @@ class TestPipUninstall:
         """Should handle wildcard in package name."""
         with (
             patch.object(dev, "_expand_wildcard_packages", return_value=["numpy", "numpy-core"]),
-            patch("subprocess.run") as mock_run,
+            patch("pyflowx.sh") as mock_run,
         ):
             mock_run.return_value = MagicMock(returncode=0)
             dev.pip_uninstall(["numpy*"])
@@ -149,7 +148,7 @@ class TestPipReinstall:
 
     def test_pip_reinstall_single_package(self) -> None:
         """Should reinstall single package."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.pip_reinstall(["numpy"])
             # Should call pip uninstall and pip install
@@ -157,7 +156,7 @@ class TestPipReinstall:
 
     def test_pip_reinstall_offline(self) -> None:
         """Should reinstall packages offline."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.pip_reinstall(["numpy"], offline=True)
             # Should call pip install with offline flags
@@ -177,14 +176,14 @@ class TestPipDownload:
 
     def test_pip_download_single_package(self) -> None:
         """Should download single package."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.pip_download(["numpy"])
             assert mock_run.called
 
     def test_pip_download_offline(self) -> None:
         """Should download packages offline."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(returncode=0)
             dev.pip_download(["numpy"], offline=True)
             # Should call pip download with offline flags
@@ -199,7 +198,7 @@ class TestPipFreeze:
 
     def test_pip_freeze(self, tmp_path: Path) -> None:
         """Should freeze dependencies."""
-        with patch("subprocess.run") as mock_run:
+        with patch("pyflowx.sh") as mock_run:
             mock_run.return_value = MagicMock(stdout="numpy==1.0.0\npandas==2.0.0", returncode=0)
             dev.pip_freeze()
             assert mock_run.called

tests/cli/test_profiler.py

@@ -27,7 +27,7 @@ def _fn() -> int:
 
 
 def _spec(name: str, deps: tuple[str, ...] = ()) -> TaskSpec[Any]:
-    return TaskSpec[Any](name, _fn, depends_on=deps)
+    return TaskSpec(name, _fn, depends_on=deps)
 
 
 def _result(
@@ -40,7 +40,7 @@ def _result(
 ) -> TaskResult[Any]:
     """构造带时间戳的 TaskResult."""
     end = start + timedelta(seconds=duration) if duration > 0 else start
-    return TaskResult[Any](
+    return TaskResult(
         spec=_spec(name),
         status=status,
         value=None,
@@ -139,7 +139,7 @@ class TestToHtml:
         start = datetime(2024, 1, 1, 0, 0, 0)
         report = px.RunReport()
         report.results["a"] = _result("a", start, 1.0)
-        report.results["b"] = TaskResult[Any](
+        report.results["b"] = TaskResult(
             spec=_spec("b"),
             status=TaskStatus.SKIPPED,
             reason="skip",

tests/cli/test_pymake_tool.py

@@ -1,9 +1,12 @@
-"""Tests for ops.pymake 模块 (@px.tool 注册验证)."""
+"""Tests for ops.pymake 模块 (@px.tool 注册验证 + CLI 调度)."""
 
 from __future__ import annotations
 
+import pytest
+
 import pyflowx as px
 import pyflowx.ops.dev.pymake
+from pyflowx.cli.pf import PfApp
 from pyflowx.tools import _TOOL_REGISTRY, get_tool
 
 
@@ -106,3 +109,58 @@ class TestPymakeRegistration:
         for name in ("ba", "bump", "cov", "tc", "p"):
             spec = get_tool("pymake", name)
             assert spec.cmd is None, f"{name} 应为聚合任务 (无 cmd)"
+
+    def test_bumpversion_cmd_uses_part_option(self) -> None:
+        """bumpversion 隐藏任务 cmd 应使用 --part patch (非位置参数)."""
+        spec = get_tool("pymake", "bumpversion")
+        assert spec.cmd is not None
+        assert "--part" in spec.cmd
+        assert "patch" in spec.cmd
+
+    def test_bumpmi_cmd_uses_part_option(self) -> None:
+        """bumpmi 子命令 cmd 应使用 --part minor (非位置参数)."""
+        spec = get_tool("pymake", "bumpmi")
+        assert spec.cmd is not None
+        assert "--part" in spec.cmd
+        assert "minor" in spec.cmd
+
+
+# ---------------------------------------------------------------------- #
+# CLI 调度测试 (pf pymake -h / pf pymake bump --dry-run)
+# ---------------------------------------------------------------------- #
+class TestPymakeCliDispatch:
+    """``pf pymake`` CLI 调度测试 (不执行真实命令)."""
+
+    def test_pymake_h_shows_help(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """pf pymake -h 应返回 0 并打印子命令列表."""
+        app = PfApp(["pymake", "-h"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Commands" in out
+        assert "bump" in out
+
+    def test_pymake_help_long_form(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """pf pymake --help 应与 -h 等价."""
+        app = PfApp(["pymake", "--help"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Commands" in out
+
+    def test_pymake_bump_dry_run(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """pf pymake bump --dry-run 应打印执行计划, 不执行真实命令."""
+        app = PfApp(["pymake", "bump", "--dry-run"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Dry run" in out
+        assert "bumpversion" in out
+
+    def test_pymake_bumpmi_dry_run(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """pf pymake bumpmi --dry-run 应打印执行计划."""
+        app = PfApp(["pymake", "bumpmi", "--dry-run"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Dry run" in out

tests/cli/test_screenshot.py

@@ -41,7 +41,7 @@ class TestTakeScreenshotFull:
             patch.object(Constants, "IS_WINDOWS", True),
             patch.object(Constants, "IS_MACOS", False),
             patch.object(Path, "home", return_value=tmp_path),
-            patch("subprocess.run") as mock_run,
+            patch("pyflowx.sh") as mock_run,
         ):
             mock_run.return_value = MagicMock(returncode=0)
             media.take_screenshot_full()
@@ -53,7 +53,7 @@ class TestTakeScreenshotFull:
             patch.object(Constants, "IS_WINDOWS", False),
             patch.object(Constants, "IS_MACOS", True),
             patch.object(Path, "home", return_value=tmp_path),
-            patch("subprocess.run") as mock_run,
+            patch("pyflowx.sh") as mock_run,
         ):
             mock_run.return_value = MagicMock(returncode=0)
             media.take_screenshot_full()
@@ -96,7 +96,7 @@ class TestTakeScreenshotArea:
             patch.object(Constants, "IS_WINDOWS", True),
             patch.object(Constants, "IS_MACOS", False),
             patch.object(Path, "home", return_value=tmp_path),
-            patch("subprocess.run") as mock_run,
+            patch("pyflowx.sh") as mock_run,
         ):
             mock_run.return_value = MagicMock(returncode=0)
             media.take_screenshot_area()
@@ -108,7 +108,7 @@ class TestTakeScreenshotArea:
             patch.object(Constants, "IS_WINDOWS", False),
             patch.object(Constants, "IS_MACOS", True),
             patch.object(Path, "home", return_value=tmp_path),
-            patch("subprocess.run") as mock_run,
+            patch("pyflowx.sh") as mock_run,
         ):
             mock_run.return_value = MagicMock(returncode=0)
             media.take_screenshot_area()

tests/test_cancellation.py

@@ -141,13 +141,11 @@ class TestMarkRemainingSkipped:
 
     def test_marks_all_unexecuted_tasks(self) -> None:
         """未执行的任务应被标记为 SKIPPED."""
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("a", lambda: 1),
-                px.TaskSpec("b", lambda: 2, depends_on=("a",)),
-                px.TaskSpec("c", lambda: 3, depends_on=("b",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("a", lambda: 1),
+            px.TaskSpec("b", lambda: 2, depends_on=("a",)),
+            px.TaskSpec("c", lambda: 3, depends_on=("b",)),
+        ])
         report = RunReport()
         # 模拟 a 已完成
         from datetime import datetime
@@ -171,12 +169,10 @@ class TestMarkRemainingSkipped:
 
     def test_marks_all_when_none_executed(self) -> None:
         """无已完成任务时，所有任务应被标记为 SKIPPED."""
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("a", lambda: 1),
-                px.TaskSpec("b", lambda: 2, depends_on=("a",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("a", lambda: 1),
+            px.TaskSpec("b", lambda: 2, depends_on=("a",)),
+        ])
         report = RunReport()
         _mark_remaining_skipped(graph, report, None)
         assert len(report.results) == 2
@@ -223,14 +219,12 @@ def _make_cancellable_graph() -> px.Graph:
         time.sleep(0.05)
         return c + 1
 
-    return px.Graph.from_specs(
-        [
-            px.TaskSpec("a", a),
-            px.TaskSpec("b", b, depends_on=("a",)),
-            px.TaskSpec("c", c, depends_on=("b",)),
-            px.TaskSpec("d", d, depends_on=("c",)),
-        ]
-    )
+    return px.Graph.from_specs([
+        px.TaskSpec("a", a),
+        px.TaskSpec("b", b, depends_on=("a",)),
+        px.TaskSpec("c", c, depends_on=("b",)),
+        px.TaskSpec("d", d, depends_on=("c",)),
+    ])
 
 
 class TestExternalCancel:
@@ -327,10 +321,12 @@ class TestKeyboardInterrupt:
         graph = _make_cancellable_graph()
 
         def _interrupt_after_first_layer(
-            graph: Any, layers: Any, context: Any, report: Any, backend: Any, on_event: Any, cancel_event: Any = None
+            graph: Any,
+            layers: Any,
+            ctx: Any,
         ) -> None:
             # 只运行第一层，然后抛 KeyboardInterrupt
-            executors.SequentialLayerRunner.execute(layers[0], graph, context, report, backend, 1, on_event)
+            executors.SequentialLayerRunner.execute(layers[0], graph, ctx, 1)
             raise KeyboardInterrupt
 
         with patch.object(executors, "_drive_sequential", _interrupt_after_first_layer):
@@ -350,12 +346,10 @@ class TestKeyboardInterrupt:
         def fn() -> int:
             return 1
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("a", fn),
-                px.TaskSpec("b", fn, depends_on=("a",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("a", fn),
+            px.TaskSpec("b", fn, depends_on=("a",)),
+        ])
 
         async def _interrupt(*args: Any, **kwargs: Any) -> None:
             raise KeyboardInterrupt

tests/test_diagnostics.py

@@ -16,7 +16,7 @@ def _fn() -> None:
     return None
 
 
-def _make_result(
+def _make_result(  # noqa: PLR0913
     name: str,
     status: TaskStatus = TaskStatus.SUCCESS,
     value: Any = None,
@@ -27,10 +27,10 @@ def _make_result(
     attempts: int = 1,
 ) -> TaskResult[Any]:
     """构造测试用 TaskResult。"""
-    spec: TaskSpec[Any] = TaskSpec[Any](name, _fn, depends_on=depends_on, soft_depends_on=soft_depends_on)
+    spec: TaskSpec[Any] = TaskSpec(name, _fn, depends_on=depends_on, soft_depends_on=soft_depends_on)
     start = datetime(2024, 1, 1, 0, 0, 0)
     end = datetime(2024, 1, 1, 0, 0, 1) if status != TaskStatus.PENDING else None
-    return TaskResult[Any](
+    return TaskResult(
         spec=spec,
         status=status,
         value=value,
@@ -507,12 +507,10 @@ class TestRunReportIntegration:
             return 1
 
         # continue_on_error=True 使失败不抛异常而写入报告（success 保持 True）
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("boom", boom, continue_on_error=True),
-                px.TaskSpec("downstream", downstream, depends_on=("boom",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("boom", boom, continue_on_error=True),
+            px.TaskSpec("downstream", downstream, depends_on=("boom",)),
+        ])
         report = px.run(graph, strategy="sequential")
         # continue_on_error 不影响 success 标志，手动标记以触发诊断
         report.success = False

tests/test_fileops.py

@@ -565,12 +565,10 @@ class TestDagComposition:
         def read_all(find_logs: list[Path]) -> list[str]:
             return [read_text(p) for p in find_logs]
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("find_logs", fn=find_logs, outputs={"paths": "$"}),
-                px.TaskSpec("read_all", fn=read_all, depends_on=("find_logs",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("find_logs", fn=find_logs, outputs={"paths": "$"}),
+            px.TaskSpec("read_all", fn=read_all, depends_on=("find_logs",)),
+        ])
         report = px.run(graph, strategy="sequential")
 
         assert report.success
@@ -592,12 +590,10 @@ class TestDagComposition:
         def copy_all(find_sources: list[Path]) -> list[Path]:
             return [copy(p, tmp_path / "dst") for p in find_sources]
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("find_sources", fn=find_sources, outputs={"files": "$"}),
-                px.TaskSpec("copy_all", fn=copy_all, depends_on=("find_sources",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("find_sources", fn=find_sources, outputs={"files": "$"}),
+            px.TaskSpec("copy_all", fn=copy_all, depends_on=("find_sources",)),
+        ])
         report = px.run(graph, strategy="sequential")
 
         assert report.success
@@ -619,12 +615,10 @@ class TestDagComposition:
         def checksums(find_bins: list[Path]) -> dict[str, str]:
             return {p.name: sha256(p) for p in find_bins}
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("find_bins", fn=find_bins, outputs={"paths": "$"}),
-                px.TaskSpec("checksums", fn=checksums, depends_on=("find_bins",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("find_bins", fn=find_bins, outputs={"paths": "$"}),
+            px.TaskSpec("checksums", fn=checksums, depends_on=("find_bins",)),
+        ])
         report = px.run(graph, strategy="sequential")
 
         assert report.success
@@ -643,12 +637,10 @@ class TestDagComposition:
             assert write_data == len("dag-content")
             return read_text(target)
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("write_data", fn=write_data, outputs={"bytes": "$"}),
-                px.TaskSpec("read_back", fn=read_back, depends_on=("write_data",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("write_data", fn=write_data, outputs={"bytes": "$"}),
+            px.TaskSpec("read_back", fn=read_back, depends_on=("write_data",)),
+        ])
         report = px.run(graph, strategy="sequential")
 
         assert report.success
@@ -671,13 +663,11 @@ class TestDagComposition:
             content = f"files: {summarize['count']}, size: {summarize['total_size']}"
             return write_text(tmp_path / "summary.txt", content)
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("find_files", fn=find_files, outputs={"paths": "$"}),
-                px.TaskSpec("summarize", fn=summarize, depends_on=("find_files",)),
-                px.TaskSpec("write_summary", fn=write_summary, depends_on=("summarize",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("find_files", fn=find_files, outputs={"paths": "$"}),
+            px.TaskSpec("summarize", fn=summarize, depends_on=("find_files",)),
+            px.TaskSpec("write_summary", fn=write_summary, depends_on=("summarize",)),
+        ])
         report = px.run(graph, strategy="sequential")
 
         assert report.success

tests/test_profiling.py

@@ -21,7 +21,7 @@ def _fn() -> int:
 
 
 def _spec(name: str, deps: tuple[str, ...] = ()) -> TaskSpec[Any]:
-    return TaskSpec[Any](name, _fn, depends_on=deps)
+    return TaskSpec(name, _fn, depends_on=deps)
 
 
 def _result(
@@ -34,7 +34,7 @@ def _result(
 ) -> TaskResult[Any]:
     """构造带时间戳的 TaskResult."""
     end = start + timedelta(seconds=duration) if duration > 0 else start
-    return TaskResult[Any](
+    return TaskResult(
         spec=_spec(name),
         status=status,
         value=None,
@@ -46,7 +46,7 @@ def _result(
 
 def _skipped_result(name: str, reason: str = "skip") -> TaskResult[Any]:
     """构造 SKIPPED 结果（无时间戳）."""
-    return TaskResult[Any](
+    return TaskResult(
         spec=_spec(name),
         status=TaskStatus.SKIPPED,
         reason=reason,
@@ -95,13 +95,11 @@ class TestProfileReportConstruction:
         report.results["a"] = _result("a", start, 1.0)
         report.results["b"] = _result("b", start + timedelta(seconds=1), 2.0)
         report.results["c"] = _result("c", start + timedelta(seconds=3), 1.5)
-        graph = px.Graph.from_specs(
-            [
-                _spec("a"),
-                _spec("b", deps=("a",)),
-                _spec("c", deps=("b",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            _spec("a"),
+            _spec("b", deps=("a",)),
+            _spec("c", deps=("b",)),
+        ])
 
         profile = ProfileReport.from_report(report, graph)
 
@@ -138,13 +136,11 @@ class TestProfileReportConstruction:
         report.results["a"] = _result("a", start, 1.0)
         report.results["b"] = _result("b", start, 3.0)
         report.results["c"] = _result("c", start + timedelta(seconds=3), 1.0)
-        graph = px.Graph.from_specs(
-            [
-                _spec("a"),
-                _spec("b"),
-                _spec("c", deps=("a", "b")),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            _spec("a"),
+            _spec("b"),
+            _spec("c", deps=("a", "b")),
+        ])
 
         profile = ProfileReport.from_report(report, graph)
 
@@ -192,12 +188,10 @@ class TestWaitTime:
         report = px.RunReport()
         report.results["a"] = _result("a", start, 1.0)
         report.results["b"] = _result("b", start + timedelta(seconds=1.5), 1.0)
-        graph = px.Graph.from_specs(
-            [
-                _spec("a"),
-                _spec("b", deps=("a",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            _spec("a"),
+            _spec("b", deps=("a",)),
+        ])
 
         profile = ProfileReport.from_report(report, graph)
 
@@ -210,12 +204,10 @@ class TestWaitTime:
         report.results["a"] = _result("a", start, 2.0)
         # b 在 a 还没完成时就开始（不应该但可能发生）
         report.results["b"] = _result("b", start + timedelta(seconds=1), 1.0)
-        graph = px.Graph.from_specs(
-            [
-                _spec("a"),
-                _spec("b", deps=("a",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            _spec("a"),
+            _spec("b", deps=("a",)),
+        ])
 
         profile = ProfileReport.from_report(report, graph)
 
@@ -228,12 +220,10 @@ class TestWaitTime:
         report = px.RunReport()
         report.results["a"] = _result("a", start, 1.0)
         report.results["b"] = _skipped_result("b")
-        graph = px.Graph.from_specs(
-            [
-                _spec("a"),
-                _spec("b", deps=("a",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            _spec("a"),
+            _spec("b", deps=("a",)),
+        ])
 
         profile = ProfileReport.from_report(report, graph)
 
@@ -251,14 +241,12 @@ class TestCriticalPath:
         report.results["b"] = _result("b", start + timedelta(seconds=1), 3.0)
         report.results["c"] = _result("c", start + timedelta(seconds=1), 1.0)
         report.results["d"] = _result("d", start + timedelta(seconds=4), 1.0)
-        graph = px.Graph.from_specs(
-            [
-                _spec("a"),
-                _spec("b", deps=("a",)),
-                _spec("c", deps=("a",)),
-                _spec("d", deps=("b", "c")),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            _spec("a"),
+            _spec("b", deps=("a",)),
+            _spec("c", deps=("a",)),
+            _spec("d", deps=("b", "c")),
+        ])
 
         profile = ProfileReport.from_report(report, graph)
 
@@ -291,7 +279,7 @@ class TestParallelism:
     def test_no_timestamps_zero_parallelism(self) -> None:
         """所有任务无时间戳：并行度为 0."""
         report = px.RunReport()
-        report.results["a"] = TaskResult[Any](spec=_spec("a"), status=TaskStatus.SUCCESS)
+        report.results["a"] = TaskResult(spec=_spec("a"), status=TaskStatus.SUCCESS)
         graph = px.Graph.from_specs([_spec("a")])
 
         profile = ProfileReport.from_report(report, graph)
@@ -405,14 +393,12 @@ class TestQueries:
         report.results["b"] = _result("b", start + timedelta(seconds=1), 3.0)
         report.results["c"] = _result("c", start + timedelta(seconds=1), 1.0)
         report.results["d"] = _result("d", start + timedelta(seconds=4), 1.0)
-        graph = px.Graph.from_specs(
-            [
-                _spec("a"),
-                _spec("b", deps=("a",)),
-                _spec("c", deps=("a",)),
-                _spec("d", deps=("b", "c")),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            _spec("a"),
+            _spec("b", deps=("a",)),
+            _spec("c", deps=("a",)),
+            _spec("d", deps=("b", "c")),
+        ])
 
         profile = ProfileReport.from_report(report, graph)
 
@@ -551,13 +537,11 @@ class TestIntegrationWithRun:
             time.sleep(0.01)  # 确保任务有实际耗时，避免 duration 极小导致并行度计算为 0
             return 1
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("a", slow),
-                px.TaskSpec("b", slow, depends_on=("a",)),
-                px.TaskSpec("c", slow, depends_on=("a",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("a", slow),
+            px.TaskSpec("b", slow, depends_on=("a",)),
+            px.TaskSpec("c", slow, depends_on=("a",)),
+        ])
         report = px.run(graph, strategy="sequential")
 
         profile = ProfileReport.from_report(report, graph)
@@ -576,13 +560,11 @@ class TestIntegrationWithRun:
             time.sleep(0.05)
             return 1
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("a", slow),
-                px.TaskSpec("b", slow),
-                px.TaskSpec("c", slow),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("a", slow),
+            px.TaskSpec("b", slow),
+            px.TaskSpec("c", slow),
+        ])
         report = px.run(graph, strategy="thread", max_workers=3)
 
         profile = ProfileReport.from_report(report, graph)

tests/test_report.py

@@ -20,7 +20,7 @@ def _fn() -> int:
     return 1
 
 
-def _make_result(
+def _make_result(  # noqa: PLR0913
     name: str = "a",
     status: TaskStatus = TaskStatus.SUCCESS,
     value: Any = 42,
@@ -32,11 +32,11 @@ def _make_result(
     reason: str | None = None,
 ) -> TaskResult[Any]:
     """构造测试用 TaskResult 实例."""
-    spec: TaskSpec[Any] = TaskSpec[Any](name, _fn, tags=tags, depends_on=depends_on)
+    spec: TaskSpec[Any] = TaskSpec(name, _fn, tags=tags, depends_on=depends_on)
     start = datetime(2024, 1, 1, 0, 0, 0)
     # 用 timedelta 精确表达秒数，避免 int() 截断小数
     end = start + timedelta(seconds=duration) if duration else None
-    return TaskResult[Any](
+    return TaskResult(
         spec=spec,
         status=status,
         value=value,
@@ -98,7 +98,7 @@ class TestRunReportSummary:
     def test_summary_with_none_duration(self) -> None:
         """未开始/未结束的任务 duration 为 None，不应计入总时长."""
         report = px.RunReport()
-        spec: TaskSpec[Any] = TaskSpec[Any]("a", _fn)  # type: ignore[arg-type]
+        spec: TaskSpec[Any] = TaskSpec("a", _fn)  # type: ignore[arg-type]
         report.results["a"] = TaskResult(spec=spec, status=TaskStatus.FAILED)
         s = report.summary()
         assert s["total_duration_seconds"] == 0.0
@@ -135,8 +135,8 @@ class TestRunReportDescribe:
     def test_describe_no_duration(self) -> None:
         """无耗时的任务应显示为 '-'."""
         report = px.RunReport()
-        spec: TaskSpec[Any] = TaskSpec[Any]("a", _fn)  # type: ignore[arg-type]
-        report.results["a"] = TaskResult[Any](spec=spec, status=TaskStatus.PENDING)
+        spec: TaskSpec[Any] = TaskSpec("a", _fn)  # type: ignore[arg-type]
+        report.results["a"] = TaskResult(spec=spec, status=TaskStatus.PENDING)
         desc = report.describe()
         assert "-" in desc  # duration 显示为 "-"
 
@@ -181,8 +181,8 @@ class TestRunReportQueries:
     def test_durations_no_duration(self) -> None:
         """无时长的任务应返回 0.0."""
         report = px.RunReport()
-        spec: TaskSpec[Any] = TaskSpec[Any]("a", _fn)  # type: ignore[arg-type]
-        report.results["a"] = TaskResult[Any](spec=spec, status=TaskStatus.PENDING)
+        spec: TaskSpec[Any] = TaskSpec("a", _fn)  # type: ignore[arg-type]
+        report.results["a"] = TaskResult(spec=spec, status=TaskStatus.PENDING)
         durs = report.durations()
         assert durs["a"] == 0.0
 
@@ -257,8 +257,8 @@ class TestRunReportToDict:
     def test_to_dict_no_timestamps(self) -> None:
         """未开始/未结束的任务 started_at/finished_at/duration 应为 None。"""
         report = px.RunReport()
-        spec: TaskSpec[Any] = TaskSpec[Any]("a", _fn)  # type: ignore[arg-type]
-        report.results["a"] = TaskResult[Any](spec=spec, status=TaskStatus.PENDING)
+        spec: TaskSpec[Any] = TaskSpec("a", _fn)  # type: ignore[arg-type]
+        report.results["a"] = TaskResult(spec=spec, status=TaskStatus.PENDING)
         d = report.to_dict()
         entry = d["results"][0]
         assert entry["started_at"] is None
@@ -439,7 +439,7 @@ class TestRunReportToHtml:
         """任务名含 HTML 特殊字符应被转义。"""
         report = px.RunReport()
         malicious = "<script>evil()</script>"
-        spec: TaskSpec[Any] = TaskSpec[Any](malicious, _fn)
+        spec: TaskSpec[Any] = TaskSpec(malicious, _fn)
         result: TaskResult[Any] = TaskResult(
             spec=spec,
             status=TaskStatus.SUCCESS,
@@ -538,7 +538,7 @@ class TestRunReportToHtml:
         """无时间戳时应渲染占位符。"""
         report = px.RunReport()
         # 构造无时间戳的结果
-        spec: TaskSpec[Any] = TaskSpec[Any]("a", _fn)
+        spec: TaskSpec[Any] = TaskSpec("a", _fn)
         result: TaskResult[Any] = TaskResult(
             spec=spec,
             status=TaskStatus.SUCCESS,
@@ -652,12 +652,10 @@ class TestRunReportSerializationIntegration:
         def double(extract: list[int]) -> list[int]:
             return [x * 2 for x in extract]
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("extract", extract, tags=("ingest",)),
-                px.TaskSpec("double", double, depends_on=("extract",), tags=("transform",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("extract", extract, tags=("ingest",)),
+            px.TaskSpec("double", double, depends_on=("extract",), tags=("transform",)),
+        ])
         report = px.run(graph, strategy="sequential")
 
         # 序列化 → 反序列化
@@ -753,12 +751,10 @@ class TestRunId:
         def task_b(a: int) -> int:
             return a * 2
 
-        graph = px.Graph.from_specs(
-            [
-                px.TaskSpec("a", task_a),
-                px.TaskSpec("b", task_b, depends_on=("a",)),
-            ]
-        )
+        graph = px.Graph.from_specs([
+            px.TaskSpec("a", task_a),
+            px.TaskSpec("b", task_b, depends_on=("a",)),
+        ])
         report = px.run(graph, strategy="sequential")
         profile = report.profile(graph)
         assert profile.total_duration >= 0

tests/test_taskspec_commands.py

@@ -354,7 +354,7 @@ class TestTaskSpecVerbose:
 
     def test_verbose_default_is_false(self) -> None:
         """verbose 默认应为 False."""
-        spec: px.TaskSpec[Any] = px.TaskSpec[Any]("a", cmd=[*ECHO_CMD, "hi"])
+        spec: px.TaskSpec[Any] = px.TaskSpec("a", cmd=[*ECHO_CMD, "hi"])
         assert spec.verbose is False
 
     def test_verbose_true_prints_command(self, capsys: pytest.CaptureFixture[str]) -> None:
@@ -367,7 +367,7 @@ class TestTaskSpecVerbose:
 
     def test_verbose_false_silent(self, capsys: pytest.CaptureFixture[str]) -> None:
         """verbose=False 时不应打印命令信息."""
-        graph = px.Graph.from_specs([px.TaskSpec[Any]("echo", cmd=[*ECHO_CMD, "silent"], verbose=False)])
+        graph = px.Graph.from_specs([px.TaskSpec("echo", cmd=[*ECHO_CMD, "silent"], verbose=False)])
         _ = px.run(graph, strategy="sequential")
         captured = capsys.readouterr()
         assert "执行命令" not in captured.out
@@ -390,7 +390,7 @@ class TestTaskSpecVerbose:
         import tempfile
 
         with tempfile.TemporaryDirectory() as tmpdir:
-            graph = px.Graph.from_specs([px.TaskSpec[Any]("ls", cmd=ECHO_CMD, cwd=Path(tmpdir), verbose=True)])
+            graph = px.Graph.from_specs([px.TaskSpec("ls", cmd=ECHO_CMD, cwd=Path(tmpdir), verbose=True)])
             _ = px.run(graph, strategy="sequential")
             captured = capsys.readouterr()
             assert "工作目录" in captured.out

tox.ini

@@ -8,7 +8,7 @@ skipsdist = true
 [testenv]
 uv_sync = true
 deps =
-    .[dev]
+    .[dev,docs,fast,office]
 commands =
     pytest -m "not slow" {posargs}
 passenv =

uv.lock

@@ -1298,7 +1298,7 @@ wheels = [
 
 [[package]]
 name = "pyflowx"
-version = "0.4.8"
+version = "0.4.11"
 source = { editable = "." }
 dependencies = [
     { name = "pyyaml" },