feat(progress): 新增实时进度监控 — ProgressCallback 协议 + RichProgressMonitor, run(progress=) 支持 bool/自定义监控器, 与 on_event/verbose 共存, 提取 _resolve_progress 控制 run 分支数

.trae/docs/iter-03-progress-monitor.md

@@ -0,0 +1,57 @@
+# 迭代 03：实时进度监控
+
+## 本轮目标
+
+提供两种进度监控形式：
+1. `run(progress=True)` 开箱即用的 rich 进度条
+2. `ProgressCallback` 协议供程序化集成
+
+复用现有 `rich` 依赖与 `EventCallback`/`TaskEvent` 事件机制。
+
+## 改动文件清单
+
+- `src/pyflowx/progress.py` — 新增：ProgressCallback Protocol + RichProgressMonitor
+- `src/pyflowx/executors.py` — run() 新增 progress 参数，组合回调链
+- `src/pyflowx/__init__.py` — 导出 ProgressCallback、RichProgressMonitor
+- `tests/test_progress.py` — 新增测试
+
+## 关键决策与依据
+
+### 1. ProgressCallback Protocol 带生命周期
+
+普通 `EventCallback`（Callable[[TaskEvent], None]）无法表达 start/finish 生命周期，
+而 rich.Progress 需要显式启动/停止。因此 Protocol 定义三个方法：
+`start(total)` / `on_event(event)` / `finish()`。
+
+### 2. progress 参数接受 bool | ProgressCallback
+
+- `False`（默认）：无进度监控
+- `True`：使用默认 `RichProgressMonitor`
+- `ProgressCallback` 实例：自定义监控器
+
+与现有 `on_event` 独立——`on_event` 是简单逐事件函数，
+`progress` 是带生命周期的监控器。两者可共存。
+
+### 3. 回调组合
+
+用 `_compose_callbacks` 将 verbose 打印、monitor.on_event、用户 on_event
+组合为单个 effective_callback，避免多层包装。
+
+### 4. RichProgressMonitor 设计
+
+- 使用 rich.Progress 的 Live 显示，列：spinner + 描述 + 进度条 + M/N + 百分比 + 耗时
+- 运行中任务名显示在描述栏（最多 3 个 + 折叠）
+- SUCCESS/FAILED/SKIPPED 均推进进度（终态即完成）
+- console 可注入，便于测试
+
+## 验证结果
+
+- ruff / pyrefly / pytest 全通过
+- 覆盖率 ≥ 95%
+- run(progress=True) 在终端显示进度条
+- 自定义 ProgressCallback 可订阅事件
+
+## 遗留事项
+
+- P3 任务通知系统（下一迭代）
+- P4 结果流式获取

src/pyflowx/__init__.py

@@ -75,6 +75,7 @@ from .errors import (
 from .executors import Strategy, run
 from .graph import Graph, GraphDefaults
 from .profiling import ProfileReport, TaskProfile
+from .progress import ProgressCallback, RichProgressMonitor
 from .report import RunReport
 from .runner import CliExitCode, CliRunner
 from .storage import JSONBackend, MemoryBackend, SQLiteBackend, StateBackend
@@ -117,8 +118,10 @@ __all__ = [
     "MemoryBackend",
     "MissingDependencyError",
     "ProfileReport",
+    "ProgressCallback",
     "PyFlowXError",
     "RetryPolicy",
+    "RichProgressMonitor",
     "RunReport",
     "SQLiteBackend",
     "StateBackend",

src/pyflowx/executors.py

@@ -57,6 +57,7 @@ from rich.console import Console
 from .context import build_call_args, describe_injection
 from .errors import TaskFailedError, TaskTimeoutError
 from .graph import Graph
+from .progress import ProgressCallback, RichProgressMonitor
 from .report import RunReport
 from .storage import StateBackend, resolve_backend
 from .task import TaskEvent, TaskHooks, TaskResult, TaskSpec, TaskStatus
@@ -728,6 +729,46 @@ class DependencyRunner:
 # ---------------------------------------------------------------------- #
 # 公共 API
 # ---------------------------------------------------------------------- #
+def _compose_callbacks(*callbacks: EventCallback | None) -> EventCallback | None:
+    """组合多个回调为一个，按顺序调用；全为 None 时返回 None。"""
+    fns = [c for c in callbacks if c is not None]
+    if not fns:
+        return None
+
+    def _composed(event: TaskEvent) -> None:
+        for fn in fns:
+            fn(event)
+
+    return _composed
+
+
+def _resolve_progress(
+    progress: bool | ProgressCallback,
+    verbose: bool,
+    on_event: EventCallback | None,
+) -> tuple[ProgressCallback | None, EventCallback | None]:
+    """解析 ``progress`` 参数，返回 (monitor, effective_callback)。
+
+    将 progress/verbose/on_event 的回调组合逻辑从 :func:`run` 中提取出来，
+    控制 ``run`` 的分支数在 ruff PLR0912 阈值内。
+    """
+    monitor: ProgressCallback | None = None
+    if progress is True:
+        monitor = RichProgressMonitor()
+    elif progress is not False:
+        monitor = progress
+
+    callbacks: list[EventCallback] = []
+    if verbose:
+        callbacks.append(_make_verbose_callback(None))
+    if monitor is not None:
+        callbacks.append(monitor.on_event)
+    if on_event is not None:
+        callbacks.append(on_event)
+    effective_callback = _compose_callbacks(*callbacks) if callbacks else None
+    return monitor, effective_callback
+
+
 def _make_verbose_callback(on_event: EventCallback | None) -> EventCallback:
     """包装 on_event 回调, 在 verbose 模式下用 rich 打印任务生命周期。"""
 
@@ -761,6 +802,7 @@ def run(
     on_event: EventCallback | None = None,
     state: StateBackend | None = None,
     concurrency_limits: Mapping[str, int] | None = None,
+    progress: bool | ProgressCallback = False,
 ) -> RunReport:
     """执行图并返回 :class:`RunReport`。
 
@@ -784,6 +826,11 @@ def run(
     concurrency_limits:
         ``{concurrency_key: max_concurrent}`` 映射。具有相同
         ``concurrency_key`` 的任务共享信号量，限制同时运行实例数。
+    progress:
+        进度监控：``False``（默认）无监控；``True`` 使用
+        :class:`~pyflowx.progress.RichProgressMonitor` 显示终端进度条；
+        传入 :class:`~pyflowx.progress.ProgressCallback` 实例使用自定义监控器。
+        与 ``on_event`` 独立且可共存。
 
     抛出
     ----
@@ -808,37 +855,46 @@ def run(
     # 各自重复调用 validate()。
     graph.validate()
 
-    effective_callback: EventCallback | None = _make_verbose_callback(on_event) if verbose else on_event
+    # 组合回调链：verbose 打印 + 进度监控 + 用户回调
+    monitor, effective_callback = _resolve_progress(progress, verbose, on_event)
+
     backend = resolve_backend(state)
     report = RunReport()
     context: dict[str, Any] = {}
     limits = concurrency_limits or {}
 
+    if monitor is not None:
+        monitor.start(len(graph))
+
     # backend.batch()：将每任务一次落盘降为整次运行一次（JSONBackend）；
     # MemoryBackend 为 no-op。即使中途抛出 TaskFailedError，batch 退出时
     # 仍会 flush 一次，保留已成功任务的结果以便断点续跑。
-    with backend.batch():
-        try:
-            if strategy == "sequential":
-                layers = graph.layers()
-                _drive_sequential(graph, layers, context, report, backend, effective_callback)
-            elif strategy == "thread":
-                layers = graph.layers()
-                _drive_threaded(graph, layers, context, report, backend, effective_callback, max_workers, limits)
-            elif strategy == "async":
-                layers = graph.layers()
-                asyncio.run(_async_drive(graph, layers, context, report, backend, effective_callback, limits))
-            elif strategy == "dependency":
-                asyncio.run(DependencyRunner.execute(graph, context, report, backend, effective_callback, limits))
-            else:
-                raise ValueError(f"Unknown strategy: {strategy!r}")
-        except TaskFailedError:
-            report.success = False
-            raise
-        finally:
-            # 关闭进程池：通知管理线程退出 + kill 工作进程，避免
-            # threading._shutdown 等待管理线程 join 工作进程导致 ~7s 阻塞。
-            _shutdown_process_pool()
+    try:
+        with backend.batch():
+            try:
+                if strategy == "sequential":
+                    layers = graph.layers()
+                    _drive_sequential(graph, layers, context, report, backend, effective_callback)
+                elif strategy == "thread":
+                    layers = graph.layers()
+                    _drive_threaded(graph, layers, context, report, backend, effective_callback, max_workers, limits)
+                elif strategy == "async":
+                    layers = graph.layers()
+                    asyncio.run(_async_drive(graph, layers, context, report, backend, effective_callback, limits))
+                elif strategy == "dependency":
+                    asyncio.run(DependencyRunner.execute(graph, context, report, backend, effective_callback, limits))
+                else:
+                    raise ValueError(f"Unknown strategy: {strategy!r}")
+            except TaskFailedError:
+                report.success = False
+                raise
+            finally:
+                # 关闭进程池：通知管理线程退出 + kill 工作进程，避免
+                # threading._shutdown 等待管理线程 join 工作进程导致 ~7s 阻塞。
+                _shutdown_process_pool()
+    finally:
+        if monitor is not None:
+            monitor.finish()
 
     return report
 

src/pyflowx/progress.py

@@ -0,0 +1,144 @@
+"""实时进度监控。
+
+提供两种监控形式：
+
+* :class:`RichProgressMonitor` —— 基于 ``rich.Progress`` 的终端进度条，
+  通过 ``run(graph, progress=True)`` 开箱即用。
+* :class:`ProgressCallback` —— 协议，定义 ``start`` / ``on_event`` / ``finish``
+  三个生命周期方法，供程序化集成（如接入 Web 仪表盘、日志聚合系统）。
+
+复用 :class:`pyflowx.task.TaskEvent` 事件机制，与现有 ``on_event`` 回调独立
+且可共存——``on_event`` 是简单逐事件函数，``ProgressCallback`` 是带生命周期的
+监控器（需要知道总任务数以渲染进度条）。
+
+快速上手
+--------
+    import pyflowx as px
+
+    # 1. 开箱即用
+    report = px.run(graph, progress=True)
+
+    # 2. 自定义监控器
+    class MyMonitor(px.ProgressCallback):
+        def start(self, total: int) -> None:
+            print(f"开始执行 {total} 个任务")
+        def on_event(self, event: px.TaskEvent) -> None:
+            print(f"{event.task}: {event.status.value}")
+        def finish(self) -> None:
+            print("执行完成")
+
+    report = px.run(graph, progress=MyMonitor())
+"""
+
+from __future__ import annotations
+
+__all__ = [
+    "ProgressCallback",
+    "RichProgressMonitor",
+]
+
+from typing import Any, Protocol, runtime_checkable
+
+from rich.console import Console
+from rich.progress import (
+    BarColumn,
+    MofNCompleteColumn,
+    Progress,
+    SpinnerColumn,
+    TaskID,
+    TaskProgressColumn,
+    TextColumn,
+    TimeElapsedColumn,
+)
+
+from .task import TaskEvent, TaskStatus
+
+
+@runtime_checkable
+class ProgressCallback(Protocol):
+    """进度监控回调协议。
+
+    生命周期：``start`` → 多次 ``on_event`` → ``finish``。
+    实现者可在 ``start`` 中初始化资源（如打开进度条），在 ``finish`` 中清理。
+    """
+
+    def start(self, total: int) -> None:
+        """执行开始时调用，``total`` 为图中的任务总数。"""
+        ...  # pragma: no cover
+
+    def on_event(self, event: TaskEvent) -> None:
+        """每次任务状态转换时调用。"""
+        ...  # pragma: no cover
+
+    def finish(self) -> None:
+        """执行结束（无论成功/失败）时调用。"""
+        ...  # pragma: no cover
+
+
+class RichProgressMonitor:
+    """基于 :class:`rich.progress.Progress` 的默认终端进度监控器。
+
+    显示运行中任务名、完成数/总数、进度条、百分比与耗时。
+    终态事件（SUCCESS/FAILED/SKIPPED）均推进进度。
+
+    Parameters
+    ----------
+    console:
+        可选的 :class:`rich.console.Console`，用于测试注入。
+        默认创建新 Console。
+    """
+
+    def __init__(self, console: Console | None = None) -> None:
+        self._progress = Progress(
+            SpinnerColumn(),
+            TextColumn("[bold blue]{task.description}"),
+            BarColumn(),
+            MofNCompleteColumn(),
+            TaskProgressColumn(),
+            TimeElapsedColumn(),
+            console=console,
+        )
+        self._task_id: TaskID | None = None
+        self._completed: int = 0
+        self._total: int = 0
+        self._running: set[str] = set()
+
+    def start(self, total: int) -> None:
+        """启动进度条。"""
+        self._total = total
+        self._progress.start()
+        self._task_id = self._progress.add_task("工作流执行中", total=total)
+
+    def on_event(self, event: TaskEvent) -> None:
+        """处理任务事件，更新进度条。"""
+        if event.status == TaskStatus.RUNNING:
+            self._running.add(event.task)
+            self._update_description()
+        elif event.status in (TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.SKIPPED):
+            self._running.discard(event.task)
+            self._completed += 1
+            if self._task_id is not None:
+                self._progress.advance(self._task_id)
+            self._update_description()
+
+    def finish(self) -> None:
+        """停止进度条。"""
+        self._progress.stop()
+
+    def _update_description(self) -> None:
+        """更新进度条描述：显示运行中任务名或完成统计。"""
+        if self._task_id is None:
+            return
+        if self._running:
+            shown = sorted(self._running)[:3]
+            suffix = f" (+{len(self._running) - 3})" if len(self._running) > 3 else ""
+            desc = f"运行中: {', '.join(shown)}{suffix}"
+        else:
+            desc = f"已完成 {self._completed}/{self._total}"
+        self._progress.update(self._task_id, description=desc)
+
+    def __enter__(self) -> RichProgressMonitor:
+        return self
+
+    def __exit__(self, *exc: Any) -> None:
+        self.finish()

tests/test_progress.py

@@ -0,0 +1,269 @@
+"""进度监控测试。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+import pyflowx as px
+from pyflowx.progress import ProgressCallback, RichProgressMonitor
+from pyflowx.task import TaskEvent, TaskStatus
+
+
+# ---------------------------------------------------------------------- #
+# 测试用监控器
+# ---------------------------------------------------------------------- #
+class _RecordingMonitor:
+    """记录所有事件的测试用监控器。"""
+
+    def __init__(self) -> None:
+        self.events: list[TaskEvent] = []
+        self.started_total: int | None = None
+        self.finished: bool = False
+
+    def start(self, total: int) -> None:
+        self.started_total = total
+
+    def on_event(self, event: TaskEvent) -> None:
+        self.events.append(event)
+
+    def finish(self) -> None:
+        self.finished = True
+
+
+# ---------------------------------------------------------------------- #
+# ProgressCallback Protocol
+# ---------------------------------------------------------------------- #
+def test_progress_callback_protocol_satisfied() -> None:
+    """_RecordingMonitor 满足 ProgressCallback 协议。"""
+    monitor = _RecordingMonitor()
+    assert isinstance(monitor, ProgressCallback)
+
+
+def test_rich_progress_monitor_satisfies_protocol() -> None:
+    """RichProgressMonitor 满足 ProgressCallback 协议。"""
+    assert isinstance(RichProgressMonitor(), ProgressCallback)
+
+
+# ---------------------------------------------------------------------- #
+# RichProgressMonitor
+# ---------------------------------------------------------------------- #
+def test_rich_monitor_start_finish_lifecycle() -> None:
+    """start/finish 生命周期正常。"""
+    monitor = RichProgressMonitor()
+    monitor.start(5)
+    assert monitor._total == 5
+    assert monitor._task_id is not None
+    monitor.finish()
+
+
+def test_rich_monitor_on_event_running() -> None:
+    """RUNNING 事件将任务加入 running 集合。"""
+    monitor = RichProgressMonitor()
+    monitor.start(3)
+    monitor.on_event(TaskEvent(task="a", status=TaskStatus.RUNNING))
+    assert "a" in monitor._running
+    monitor.finish()
+
+
+def test_rich_monitor_on_event_success_advances() -> None:
+    """SUCCESS 事件推进进度并从 running 移除。"""
+    monitor = RichProgressMonitor()
+    monitor.start(2)
+    monitor.on_event(TaskEvent(task="a", status=TaskStatus.RUNNING))
+    monitor.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=0.1))
+    assert "a" not in monitor._running
+    assert monitor._completed == 1
+    monitor.finish()
+
+
+def test_rich_monitor_on_event_failed_advances() -> None:
+    """FAILED 事件也推进进度。"""
+    monitor = RichProgressMonitor()
+    monitor.start(1)
+    monitor.on_event(TaskEvent(task="a", status=TaskStatus.FAILED, error="boom"))
+    assert monitor._completed == 1
+    monitor.finish()
+
+
+def test_rich_monitor_on_event_skipped_advances() -> None:
+    """SKIPPED 事件也推进进度。"""
+    monitor = RichProgressMonitor()
+    monitor.start(1)
+    monitor.on_event(TaskEvent(task="a", status=TaskStatus.SKIPPED, reason="缓存命中"))
+    assert monitor._completed == 1
+    monitor.finish()
+
+
+def test_rich_monitor_context_manager() -> None:
+    """__enter__/__exit__ 正常工作。"""
+    with RichProgressMonitor() as monitor:
+        monitor.start(1)
+        monitor.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS))
+    # 退出后不再报错
+
+
+def test_rich_monitor_on_event_before_start() -> None:
+    """start 前调用 on_event 不报错（task_id 为 None 时早返回）。"""
+    monitor = RichProgressMonitor()
+    monitor.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS))
+    # 未 start 时 advance 不执行，_update_description 早返回
+    assert monitor._completed == 1
+    assert monitor._task_id is None
+
+
+def test_rich_monitor_pending_event_noop() -> None:
+    """PENDING 事件不是 RUNNING 也不是终态，应被忽略。"""
+    monitor = RichProgressMonitor()
+    monitor.start(2)
+    monitor.on_event(TaskEvent(task="a", status=TaskStatus.PENDING))
+    assert monitor._completed == 0
+    assert len(monitor._running) == 0
+    monitor.finish()
+
+
+def test_rich_monitor_update_description_running() -> None:
+    """运行中时描述应包含任务名。"""
+    monitor = RichProgressMonitor()
+    monitor.start(5)
+    monitor.on_event(TaskEvent(task="task_a", status=TaskStatus.RUNNING))
+    monitor.on_event(TaskEvent(task="task_b", status=TaskStatus.RUNNING))
+    # 无法直接断言 rich 内部状态，但 _running 应包含两个任务
+    assert len(monitor._running) == 2
+    monitor.finish()
+
+
+def test_rich_monitor_update_description_many_running_folds() -> None:
+    """超过 3 个运行中任务时描述应折叠。"""
+    monitor = RichProgressMonitor()
+    monitor.start(10)
+    for i in range(5):
+        monitor.on_event(TaskEvent(task=f"task_{i}", status=TaskStatus.RUNNING))
+    assert len(monitor._running) == 5
+    # 不报错即可（描述更新逻辑内部处理折叠）
+    monitor.finish()
+
+
+# ---------------------------------------------------------------------- #
+# run() 集成测试
+# ---------------------------------------------------------------------- #
+def _make_simple_graph() -> px.Graph:
+    """构建简单测试图。"""
+
+    def task_a() -> int:
+        return 1
+
+    def task_b(task_a: int) -> int:
+        return task_a + 1
+
+    return px.Graph.from_specs([
+        px.TaskSpec("task_a", task_a),
+        px.TaskSpec("task_b", task_b, depends_on=("task_a",)),
+    ])
+
+
+def test_run_with_progress_true() -> None:
+    """progress=True 使用 RichProgressMonitor 不报错。"""
+    graph = _make_simple_graph()
+    report = px.run(graph, strategy="sequential", progress=True)
+    assert report.success
+    assert report["task_b"] == 2
+
+
+def test_run_with_custom_monitor() -> None:
+    """progress=自定义监控器，事件被正确捕获。"""
+    graph = _make_simple_graph()
+    monitor = _RecordingMonitor()
+    report = px.run(graph, strategy="sequential", progress=monitor)
+    assert report.success
+    assert monitor.started_total == 2
+    assert monitor.finished
+    # 至少有 RUNNING + SUCCESS 事件
+    statuses = [e.status for e in monitor.events]
+    assert TaskStatus.RUNNING in statuses
+    assert TaskStatus.SUCCESS in statuses
+
+
+def test_run_with_monitor_and_on_event_coexist() -> None:
+    """progress 监控器与 on_event 回调可共存。"""
+    graph = _make_simple_graph()
+    monitor = _RecordingMonitor()
+    on_event_events: list[TaskEvent] = []
+
+    def capture(event: TaskEvent) -> None:
+        on_event_events.append(event)
+
+    report = px.run(
+        graph,
+        strategy="sequential",
+        progress=monitor,
+        on_event=capture,
+    )
+    assert report.success
+    # 两者都应收到事件
+    assert len(monitor.events) > 0
+    assert len(on_event_events) > 0
+    assert len(monitor.events) == len(on_event_events)
+
+
+def test_run_with_monitor_finish_called_on_failure() -> None:
+    """任务失败时 monitor.finish 仍被调用。"""
+
+    def fail_task() -> Any:
+        raise RuntimeError("boom")
+
+    graph = px.Graph.from_specs([px.TaskSpec("fail", fail_task)])
+    monitor = _RecordingMonitor()
+    with pytest.raises(px.TaskFailedError):
+        px.run(graph, strategy="sequential", progress=monitor)
+    assert monitor.finished
+    assert monitor.started_total == 1
+
+
+def test_run_progress_false_no_monitor() -> None:
+    """progress=False 不创建监控器，行为不变。"""
+    graph = _make_simple_graph()
+    report = px.run(graph, strategy="sequential", progress=False)
+    assert report.success
+
+
+def test_run_progress_with_verbose_coexist() -> None:
+    """progress 与 verbose 可同时使用。"""
+    graph = _make_simple_graph()
+    monitor = _RecordingMonitor()
+    report = px.run(graph, strategy="sequential", progress=monitor, verbose=True)
+    assert report.success
+    assert monitor.finished
+
+
+def test_run_progress_with_thread_strategy() -> None:
+    """progress 在 thread 策略下正常工作。"""
+    graph = _make_simple_graph()
+    monitor = _RecordingMonitor()
+    report = px.run(graph, strategy="thread", progress=monitor)
+    assert report.success
+    assert monitor.finished
+
+
+def test_run_progress_with_dependency_strategy() -> None:
+    """progress 在 dependency 策略下正常工作。"""
+    graph = _make_simple_graph()
+    monitor = _RecordingMonitor()
+    report = px.run(graph, strategy="dependency", progress=monitor)
+    assert report.success
+    assert monitor.finished
+
+
+def test_run_progress_with_async_strategy() -> None:
+    """progress 在 async 策略下正常工作。"""
+
+    async def atask() -> int:
+        return 42
+
+    graph = px.Graph.from_specs([px.TaskSpec("atask", atask)])
+    monitor = _RecordingMonitor()
+    report = px.run(graph, strategy="async", progress=monitor)
+    assert report.success
+    assert monitor.finished
+    assert monitor.started_total == 1