feat(executors): 新增 run_iter() 流式执行 API — 生成器逐个 yield (name, TaskResult), 后台线程+队列实现, 支持 4 种策略; _finalize_failure 在抛出前写入 report.results 使失败任务可被 yield; run() 新增内部 _report 参数支持结果共享

.trae/docs/iter-05-run-iter.md

@@ -0,0 +1,44 @@
+# 迭代 05：结果流式获取（run_iter）
+
+## 目标
+
+新增 `run_iter()` 函数，以生成器形式逐个 yield 完成任务的结果，
+适用于大量任务逐个处理、内存敏感场景。保留 `run()` 返回 `RunReport`，
+完全向后兼容。
+
+## 设计
+
+### 方案：线程 + 队列
+
+`run()` 是阻塞函数，内部可能调用 `asyncio.run()`，无法直接改为生成器。
+采用线程 + 队列方案：
+
+1. `run()` 新增内部参数 `_report: RunReport | None`（下划线前缀，内部用）
+2. `run_iter()` 创建 `RunReport`，传给 `run(_report=report)`
+3. `on_event` 回调将终态事件（SUCCESS/FAILED/SKIPPED）的任务名放入队列
+4. 后台线程执行 `run()`，生成器从队列取名，读 `report.results[name]` yield
+5. `None` 哨兵标记执行结束
+
+### 关键点
+
+- **结果可见性**：`_store_result` 先写 `report.results[name]` 再 `_emit`，
+  队列提供内存屏障，yield 时结果必然可见
+- **去重**：skipped 任务可能 double-emit（`_prepare_for_execution` + `_store_result`），
+  用 `yielded` set 去重
+- **错误传播**：`run()` 抛异常时存入 `error_box`，生成器耗尽后 re-raise
+- **早退**：daemon 线程，生成器 `finally` 中 join(timeout=0) 不阻塞
+
+## 改动文件
+
+- `src/pyflowx/executors.py` — 新增 `run_iter()`，`run()` 加 `_report` 参数
+- `src/pyflowx/__init__.py` — 导出 `run_iter`
+- `tests/test_run_iter.py` — 新增测试
+
+## 验收标准
+
+- `run_iter()` 逐个 yield 所有任务的 `(name, TaskResult)`
+- 支持 sequential/thread/async/dependency 四种策略
+- 失败任务也 yield（FAILED 状态），之后 re-raise `TaskFailedError`
+- 缓存命中的 SKIPPED 任务也 yield
+- `run()` 行为不变，现有测试全过
+- 覆盖率 ≥ 95%

src/pyflowx/__init__.py

@@ -72,7 +72,7 @@ from .errors import (
     TaskFailedError,
     TaskTimeoutError,
 )
-from .executors import Strategy, run
+from .executors import Strategy, run, run_iter
 from .graph import Graph, GraphDefaults
 from .notification import (
     ALL_LEVELS,
@@ -163,6 +163,7 @@ __all__ = [
     "list_tools",
     "run",
     "run_command",
+    "run_iter",
     "run_tool",
     "task",
     "task_template",

src/pyflowx/executors.py

@@ -46,9 +46,10 @@ import concurrent.futures
 import contextlib
 import inspect
 import logging
+import queue
 import threading
 import time
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Awaitable, Callable, Iterator, Mapping
 from datetime import datetime
 from typing import Any, Literal, cast
 
@@ -305,10 +306,18 @@ def _finalize_failure(
     layer_idx: int | None,
     on_event: EventCallback | None,
     continue_on_error: bool,
+    name: str,
+    report: RunReport | None,
 ) -> None:
-    """标记任务为 FAILED。若 ``continue_on_error`` 为真则不抛出异常。"""
+    """标记任务为 FAILED。若 ``continue_on_error`` 为真则不抛出异常。
+
+    失败结果在抛出前写入 ``report.results[name]``，使流式 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)
     if continue_on_error:
         logger.warning(
@@ -330,6 +339,8 @@ def _handle_failure(
     exc: BaseException,
     layer_idx: int | None,
     on_event: EventCallback | None,
+    name: str,
+    report: RunReport | None,
 ) -> bool:
     """统一处理失败：超时转换、重试决策、finalize。
 
@@ -359,7 +370,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)
+    _finalize_failure(result, layer_idx, on_event, spec.continue_on_error, name, report)
     return True
 
 
@@ -396,7 +407,7 @@ class SyncTaskRunner:
                 _mark_success(spec, result, value)
                 return result
             except Exception as exc:
-                if _handle_failure(spec, result, exc, layer_idx, on_event):
+                if _handle_failure(spec, result, exc, layer_idx, on_event, spec.name, report):
                     return result
                 wait = spec.retry.wait_seconds(result.attempts)
                 if wait > 0:
@@ -435,7 +446,7 @@ class AsyncTaskRunner:
                     _mark_success(spec, result, value)
                     return result
                 except Exception as exc:
-                    if _handle_failure(spec, result, exc, layer_idx, on_event):
+                    if _handle_failure(spec, result, exc, layer_idx, on_event, spec.name, report):
                         return result
                     wait = spec.retry.wait_seconds(result.attempts)
                     if wait > 0:
@@ -820,6 +831,7 @@ def run(
     concurrency_limits: Mapping[str, int] | None = None,
     progress: bool | ProgressCallback = False,
     notifiers: Notifier | list[Notifier] | None = None,
+    _report: RunReport | None = None,
 ) -> RunReport:
     """执行图并返回 :class:`RunReport`。
 
@@ -880,7 +892,7 @@ def run(
     monitor, notifier_list, effective_callback = _resolve_progress(progress, verbose, on_event, notifiers)
 
     backend = resolve_backend(state)
-    report = RunReport()
+    report = _report if _report is not None else RunReport()
     context: dict[str, Any] = {}
     limits = concurrency_limits or {}
 
@@ -919,6 +931,96 @@ def run(
     return report
 
 
+def run_iter(
+    graph: Graph,
+    strategy: Strategy = "dependency",
+    *,
+    max_workers: int | None = None,
+    verbose: bool = False,
+    state: StateBackend | None = None,
+    concurrency_limits: Mapping[str, int] | None = None,
+) -> Iterator[tuple[str, TaskResult[Any]]]:
+    """流式执行图，逐个 yield 完成任务的 ``(name, TaskResult)``。
+
+    与 :func:`run` 相同的执行语义，但以生成器形式逐个返回结果而非等待
+    全部完成后返回 :class:`RunReport`。适用于：
+
+    * 大量任务需要逐个处理结果（如写入数据库、发送通知）
+    * 内存敏感场景（不需要同时持有所有结果）
+    * 需要在任务完成时即时响应
+
+    内部通过后台线程执行 :func:`run` + 队列传递结果实现，不改变 ``run``
+    的执行逻辑。生成器耗尽后若执行过程抛异常则 re-raise。
+
+    参数与 :func:`run` 一致（不含 ``dry_run``/``on_event``/``progress``/
+    ``notifiers``/``_report``，这些由流式机制内部管理）。
+
+    Yields
+    ------
+    tuple[str, TaskResult]
+        ``(task_name, result)``，按任务完成顺序 yield。终态事件
+        （SUCCESS/FAILED/SKIPPED）均会 yield。
+
+    Raises
+    ------
+    ValueError
+        ``strategy`` 不被识别时。
+    TaskFailedError
+        任何任务耗尽重试后仍失败时（在 yield 完已失败任务后 re-raise）。
+
+    Example
+    -------
+        for name, result in px.run_iter(graph):
+            print(f"{name}: {result.status.value}")
+    """
+    report = RunReport()
+    ready: queue.Queue[str | None] = queue.Queue()
+    error_box: list[BaseException | None] = [None]
+    yielded: set[str] = set()
+
+    _terminal = {TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.SKIPPED}
+
+    def _on_event(event: TaskEvent) -> None:
+        if event.status in _terminal:
+            ready.put(event.task)
+
+    def _worker() -> None:
+        try:
+            run(
+                graph,
+                strategy=strategy,
+                max_workers=max_workers,
+                verbose=verbose,
+                on_event=_on_event,
+                state=state,
+                concurrency_limits=concurrency_limits,
+                _report=report,
+            )
+        except BaseException as exc:
+            error_box[0] = exc
+        finally:
+            ready.put(None)
+
+    thread = threading.Thread(target=_worker, daemon=True)
+    thread.start()
+
+    try:
+        while True:
+            name = ready.get()
+            if name is None:
+                break
+            if name in yielded:
+                continue
+            yielded.add(name)
+            if name in report.results:
+                yield name, report.results[name]
+    finally:
+        thread.join(timeout=0)
+
+    if error_box[0] is not None:
+        raise error_box[0]
+
+
 def _print_dry_run(graph: Graph, layers: list[list[str]]) -> None:
     """打印执行计划但不运行任何任务。"""
     _verbose_console.print(f"[bold]Dry run:[/bold] [cyan]{len(graph)}[/cyan] tasks, [cyan]{len(layers)}[/cyan] layers")

tests/test_run_iter.py

@@ -0,0 +1,251 @@
+"""run_iter() 流式执行测试。"""
+
+from __future__ import annotations
+
+import pytest
+
+import pyflowx as px
+from pyflowx.task import TaskStatus
+
+
+# ---------------------------------------------------------------------- #
+# 基础流式执行
+# ---------------------------------------------------------------------- #
+def test_run_iter_yields_all_results() -> None:
+    """run_iter 逐个 yield 所有任务结果。"""
+
+    @px.task
+    def a() -> int:
+        return 1
+
+    @px.task(depends_on=("a",))
+    def b(a: int) -> int:
+        return a + 1
+
+    graph = px.Graph.from_specs([a, b])
+    results = dict(px.run_iter(graph))
+    assert len(results) == 2
+    assert results["a"].status == TaskStatus.SUCCESS
+    assert results["a"].value == 1
+    assert results["b"].status == TaskStatus.SUCCESS
+    assert results["b"].value == 2
+
+
+def test_run_iter_returns_iterator() -> None:
+    """run_iter 返回迭代器。"""
+
+    @px.task
+    def t() -> int:
+        return 42
+
+    graph = px.Graph.from_specs([t])
+    it = px.run_iter(graph)
+    assert hasattr(it, "__next__")
+    name, result = next(it)
+    assert name == "t"
+    assert result.value == 42
+    with pytest.raises(StopIteration):
+        next(it)
+
+
+def test_run_iter_single_task() -> None:
+    """单任务图流式执行。"""
+
+    @px.task
+    def solo() -> str:
+        return "hello"
+
+    graph = px.Graph.from_specs([solo])
+    results = list(px.run_iter(graph))
+    assert len(results) == 1
+    assert results[0][0] == "solo"
+    assert results[0][1].value == "hello"
+
+
+# ---------------------------------------------------------------------- #
+# 四种策略
+# ---------------------------------------------------------------------- #
+@pytest.mark.parametrize("strategy", ["sequential", "thread", "async", "dependency"])
+def test_run_iter_all_strategies(strategy: str) -> None:
+    """run_iter 支持所有执行策略。"""
+
+    @px.task
+    def x() -> int:
+        return 10
+
+    @px.task(depends_on=("x",))
+    def y(x: int) -> int:
+        return x * 2
+
+    graph = px.Graph.from_specs([x, y])
+    results = dict(px.run_iter(graph, strategy=strategy))  # type: ignore[arg-type]
+    assert results["x"].value == 10
+    assert results["y"].value == 20
+
+
+# ---------------------------------------------------------------------- #
+# 任务完成顺序
+# ---------------------------------------------------------------------- #
+def test_run_iter_yields_in_completion_order_dependency() -> None:
+    """dependency 策略下，x 在 y 之前完成。"""
+
+    @px.task
+    def x() -> int:
+        return 1
+
+    @px.task(depends_on=("x",))
+    def y(x: int) -> int:
+        return x + 1
+
+    graph = px.Graph.from_specs([x, y])
+    names = [name for name, _ in px.run_iter(graph, strategy="dependency")]
+    assert names.index("x") < names.index("y")
+
+
+# ---------------------------------------------------------------------- #
+# 失败传播
+# ---------------------------------------------------------------------- #
+def test_run_iter_yields_failed_then_reraises() -> None:
+    """失败任务先 yield（FAILED），然后 re-raise TaskFailedError。"""
+
+    @px.task
+    def boom() -> None:
+        raise RuntimeError("explode")
+
+    graph = px.Graph.from_specs([boom])
+    results = []
+    with pytest.raises(px.TaskFailedError):
+        for name, result in px.run_iter(graph):
+            results.append((name, result))
+    assert len(results) == 1
+    assert results[0][0] == "boom"
+    assert results[0][1].status == TaskStatus.FAILED
+
+
+def test_run_iter_failure_with_continue_on_error() -> None:
+    """continue_on_error=True 时，失败任务 yield 后继续执行其他任务。"""
+
+    def fail() -> None:
+        raise ValueError("err")
+
+    graph = px.Graph.from_specs([
+        px.TaskSpec("fail", fn=fail, continue_on_error=True),
+        px.TaskSpec("ok", fn=lambda: 1),
+    ])
+    results = dict(px.run_iter(graph))
+    assert results["fail"].status == TaskStatus.FAILED
+    assert results["ok"].status == TaskStatus.SUCCESS
+
+
+# ---------------------------------------------------------------------- #
+# 跳过与缓存
+# ---------------------------------------------------------------------- #
+def test_run_iter_yields_skipped_tasks() -> None:
+    """条件跳过的任务以 SKIPPED 状态 yield。"""
+    never_true = lambda _ctx: False  # noqa: E731
+
+    graph = px.Graph.from_specs([
+        px.TaskSpec("normal", fn=lambda: 1),
+        px.TaskSpec("skipped", fn=lambda: 2, conditions=(never_true,)),
+    ])
+    results = dict(px.run_iter(graph))
+    assert results["normal"].status == TaskStatus.SUCCESS
+    assert results["skipped"].status == TaskStatus.SKIPPED
+
+
+def test_run_iter_yields_cached_tasks() -> None:
+    """缓存命中的任务以 SKIPPED 状态 yield。"""
+
+    @px.task
+    def cached() -> int:
+        return 42
+
+    graph = px.Graph.from_specs([cached])
+    backend = px.MemoryBackend()
+    backend.save("cached", 42)
+
+    results = dict(px.run_iter(graph, state=backend))
+    assert results["cached"].status == TaskStatus.SKIPPED
+    assert results["cached"].value == 42
+    assert results["cached"].reason is not None
+
+
+# ---------------------------------------------------------------------- #
+# 早退（生成器未耗尽）
+# ---------------------------------------------------------------------- #
+def test_run_iter_early_exit_does_not_block() -> None:
+    """提前 break 不会阻塞生成器销毁。"""
+
+    @px.task
+    def a() -> int:
+        return 1
+
+    @px.task(depends_on=("a",))
+    def b(a: int) -> int:
+        return a + 1
+
+    @px.task(depends_on=("b",))
+    def c(b: int) -> int:
+        return b + 1
+
+    graph = px.Graph.from_specs([a, b, c])
+    count = 0
+    for _name, _result in px.run_iter(graph):
+        count += 1
+        if count >= 1:
+            break
+    assert count == 1
+
+
+# ---------------------------------------------------------------------- #
+# 不可识别策略
+# ---------------------------------------------------------------------- #
+def test_run_iter_unknown_strategy_raises() -> None:
+    """不可识别的策略在生成器迭代时抛 ValueError。"""
+
+    @px.task
+    def t() -> int:
+        return 1
+
+    graph = px.Graph.from_specs([t])
+    with pytest.raises(ValueError, match="Unknown strategy"):
+        list(px.run_iter(graph, strategy="invalid"))  # type: ignore[arg-type]
+
+
+# ---------------------------------------------------------------------- #
+# 多任务图
+# ---------------------------------------------------------------------- #
+def test_run_iter_diamond_graph() -> None:
+    """菱形依赖图：a → b,c → d，所有结果均 yield。"""
+
+    @px.task
+    def a() -> int:
+        return 1
+
+    @px.task(depends_on=("a",))
+    def b(a: int) -> int:
+        return a + 10
+
+    @px.task(depends_on=("a",))
+    def c(a: int) -> int:
+        return a + 20
+
+    @px.task(depends_on=("b", "c"))
+    def d(b: int, c: int) -> int:
+        return b + c
+
+    graph = px.Graph.from_specs([a, b, c, d])
+    results = dict(px.run_iter(graph))
+    assert results["a"].value == 1
+    assert results["b"].value == 11
+    assert results["c"].value == 21
+    assert results["d"].value == 32
+
+
+# ---------------------------------------------------------------------- #
+# 公共 API 导出
+# ---------------------------------------------------------------------- #
+def test_run_iter_exported() -> None:
+    """run_iter 从 pyflowx 顶层导出。"""
+    assert hasattr(px, "run_iter")
+    assert callable(px.run_iter)