bump version to 0.4.11

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

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/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/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_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:
@@ -190,3 +198,53 @@ class TestBumpProjectVersion:
 
         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_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/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.10"
 source = { editable = "." }
 dependencies = [
     { name = "pyyaml" },