feat: P9 功能扩展 — 任务编排(group/dynamic)/数据流(output_of/pipeline)/持久化(resume_from/RunHistory)/监控导出(MetricsCollector/health_check/HTTP server)；1525 测试全绿，覆盖率 97.58%

.trae/docs/iter-21-p9-feature-extension.md

@@ -0,0 +1,93 @@
+# 迭代 21 — P9 功能扩展（任务编排/数据流/持久化/监控）
+
+## 本轮目标
+
+延续 P8 性能优化后，扩展 PyFlowX 在四个方向的核心能力：
+
+1. **P9.1 任务编排增强**：任务分组（`Graph.group`）+ 动态任务生成（`TaskSpec(dynamic=True)`）。
+2. **P9.2 数据流增强**：`RunReport.output_of` 命名输出提取 + `Graph.pipeline` 语义别名。
+3. **P9.3 持久化与恢复**：`run(resume_from=...)` 检查点恢复 + `RunHistory` 历史管理。
+4. **P9.4 监控导出**：`MetricsCollector`（Prometheus 文本）+ `health_check` + `start_metrics_server`（零依赖 HTTP）。
+
+## 改动文件清单
+
+### 新增
+
+- `src/pyflowx/monitoring.py` — `MetricsCollector` / `health_check` / `start_metrics_server`。
+- `src/pyflowx/history.py` — `RunHistory` 文件系统历史管理。
+- `tests/test_group.py` — 13 用例（P9.1.2 任务分组）。
+- `tests/test_dynamic.py` — 11 用例（P9.1.3 动态任务）。
+- `tests/test_dataflow.py` — 12 用例（P9.2 数据流）。
+- `tests/test_persistence.py` — 15 用例（P9.3 检查点/历史）。
+- `tests/test_monitoring.py` — 16 用例（P9.4 监控导出）。
+
+### 修改
+
+- `src/pyflowx/task.py` — `TaskSpec` 新增 `dynamic: bool = False` 字段；`task()` 工厂新增 `dynamic` 参数。
+- `src/pyflowx/graph.py` — 新增 `group()` 方法；`_rewrite_deps_for_loops` 改名 `_rewrite_deps`，统一处理 `_loop_groups` + `_groups`；新增 `pipeline()` 语义别名。
+- `src/pyflowx/report.py` — 新增 `_resolve_output_path` 辅助 + `RunReport.output_of(name, output)` 方法。
+- `src/pyflowx/executors.py` —
+  - `DependencyRunner.execute` 增加 while 循环 + `f.exception()` 手动传播（替代 `asyncio.gather`）。
+  - 新增 `_extract_dynamic_specs` 辅助；`_run_task` 检测动态生成、注册派生 spec、启动派生 task。
+  - `_dispatch_strategy` 增加 `has_dynamic` 校验（仅 `dependency` 策略支持）。
+  - 新增 `_load_resume_report` + `_apply_resume` 辅助；`run()` 接受 `resume_from` 参数。
+  - `_filter_and_sort` + `_run_task` 检查点跳过逻辑。
+- `src/pyflowx/__init__.py` — 导出 `RunHistory` / `MetricsCollector` / `health_check` / `start_metrics_server`。
+
+## 关键决策与依据
+
+### 1. 动态任务仅支持 `dependency` 策略
+
+`dynamic=True` 允许任务 `fn` 运行时返回 `TaskSpec | list[TaskSpec]`，DependencyRunner 中途注册并调度派生任务。层模型（sequential/thread/async）的层屏障与动态扩展语义冲突，因此 `_dispatch_strategy` 显式拒绝其他策略。
+
+依据：层模型按拓扑分层执行，动态任务可能跨层；依赖驱动模型无层屏障，可自然容纳运行时新增节点。
+
+### 2. `asyncio.wait` 替代 `asyncio.gather` + 手动异常传播
+
+`asyncio.gather` 在 first-failure 时取消其他任务并 re-raise，但所有 task 必须在调用前确定。动态任务场景下 `futures` 字典会随执行增长，必须用 while 循环 + `asyncio.wait(return_when=FIRST_COMPLETED)` 增量调度。
+
+陷阱：`asyncio.wait` **不会**自动 re-raise task 异常。需手动 `f.exception()` 检查并显式 `raise`，同时取消未完成任务，才能与 `gather` 的 fail-fast 行为一致。该陷阱导致 7 个测试失败后定位修复。
+
+### 3. 检查点恢复：仅恢复 SUCCESS 任务
+
+`resume_from` 接受 `RunReport | str | Path`，仅恢复 `status == SUCCESS` 且名称在当前图中的任务。FAILED/SKIPPED 任务会重新执行。`_apply_resume` 抽取为独立函数以控制 `run()` 的 PLR0912 分支数（≤12）。
+
+### 4. `metrics_text` 按 section 条件化输出
+
+最初实现总是输出 `# HELP` / `# TYPE` 行（即便无数据）。这导致 `reset()` 后文本仍包含 `pyflowx_task_total` 字样，违反测试预期。改为所有 section（含 task_total/duration/duration_sum）都按数据存在性条件化，与 retries/run_total 一致。
+
+### 5. `start_metrics_server` 用 `http.server` 零依赖
+
+未引入 `prometheus_client` 等三方库，仅用标准库 `http.server.HTTPServer` + `BaseHTTPRequestHandler`。生产环境可叠加反向代理或替换为 `prometheus_client`。`@override` 装饰器按 Python 版本条件导入（`typing` 3.12+ / `typing_extensions` <3.12）。
+
+### 6. `RunHistory` 文件命名与 `__contains__`/`__len__`
+
+按 `{run_id}.json` 命名存储；`__contains__` / `__len__` 委托文件存在性检查，避免维护内存索引。`latest()` 按 mtime 排序取最新。
+
+## 验证结果
+
+```
+ruff check .         → All checks passed!
+ruff format --check  → 126 files already formatted
+pyrefly check .      → 0 errors (68 suppressed)
+pytest --cov=pyflowx → 1525 passed in 11.56s
+coverage             → 97.58%（branch；门槛 95%）
+```
+
+各模块覆盖率（P9 新增部分）：
+- `monitoring.py` 99%（仅 `typing_extensions` 回退分支未覆盖）
+- `history.py` 95%
+- `executors.py` 97%（dynamic/resume 分支已覆盖）
+- `report.py` 99%（`output_of` 路径解析已覆盖）
+- `graph.py` 98%（`group`/`pipeline`/`_rewrite_deps` 已覆盖）
+
+## 遗留事项
+
+- 监控导出未实现 OpenTelemetry 协议（当前仅 Prometheus 文本格式）。如需 OTLP，可在后续迭代引入 `opentelemetry-sdk`（违反零依赖原则，需用户确认）。
+- `RunHistory` 未实现压缩/归档；大量历史运行可能占用较多磁盘。后续可加 `max_entries` 或滚动归档。
+- 动态任务当前仅在 DependencyRunner 中实现；如层模型用户也需要，需额外设计跨层调度策略。
+- `_apply_resume` 仅恢复 `report.results`，未恢复 `backend` 缓存；如需检查点+缓存联合恢复，需额外设计。
+
+## 归档说明
+
+iter-21 暂不归档（前 5 次 iter-16~20 已归档至 `.trae/skills/pyflowx-development/SKILL.md`）。下次清理窗口为 iter-26 前夕。

pyproject.toml

@@ -74,7 +74,7 @@ packages = ["src/pyflowx"]
 pyflowx = { workspace = true }
 
 [dependency-groups]
-dev = ["pyflowx[dev,docs,office]"]
+dev = ["pyflowx[dev,docs,fast,office]"]
 
 [tool.coverage.run]
 branch      = true

src/pyflowx/__init__.py

@@ -76,6 +76,8 @@ from .errors import (
 )
 from .executors import Strategy, run, run_iter
 from .graph import Graph, GraphDefaults
+from .history import RunHistory
+from .monitoring import MetricsCollector, health_check, start_metrics_server
 from .notification import (
     ALL_LEVELS,
     CallbackNotifier,
@@ -96,6 +98,7 @@ from .runner import CliExitCode, CliRunner
 from .storage import JSONBackend, MemoryBackend, SQLiteBackend, StateBackend
 from .task import (
     CacheKeyFn,
+    LoopSpec,
     RetryPolicy,
     TaskCmd,
     TaskEvent,
@@ -140,7 +143,9 @@ __all__ = [
     "GraphDefaults",
     "InjectionError",
     "JSONBackend",
+    "LoopSpec",
     "MemoryBackend",
+    "MetricsCollector",
     "MissingDependencyError",
     "NotificationLevel",
     "Notifier",
@@ -149,6 +154,7 @@ __all__ = [
     "PyFlowXError",
     "RetryPolicy",
     "RichProgressMonitor",
+    "RunHistory",
     "RunReport",
     "SQLiteBackend",
     "SlackNotifier",
@@ -173,6 +179,7 @@ __all__ = [
     "compose",
     "describe_injection",
     "diagnose",
+    "health_check",
     "list_subcommands",
     "list_tools",
     "load_yaml",
@@ -181,6 +188,7 @@ __all__ = [
     "run_command",
     "run_iter",
     "run_tool",
+    "start_metrics_server",
     "task",
     "task_template",
     "tool",

src/pyflowx/executors.py

@@ -50,7 +50,9 @@ import queue
 import threading
 import time
 from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping
+from dataclasses import replace as dc_replace
 from datetime import datetime
+from pathlib import Path
 from typing import Any, Literal, cast
 
 from rich.console import Console
@@ -570,6 +572,10 @@ def _filter_and_sort(
     for name in layer:
         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
+            continue
         if not _apply_cached(name, spec, context, report, backend, on_event):
             to_run.append(name)
     return _sort_by_priority(to_run, specs)
@@ -724,6 +730,26 @@ class AsyncLayerRunner:
             _store_result(name, result, graph.resolved_spec(name), task_ctx, context, report, backend, on_event)
 
 
+def _extract_dynamic_specs(result: TaskResult[Any], spec: TaskSpec[Any]) -> list[TaskSpec[Any]]:
+    """从 ``dynamic`` 任务的返回值中提取新 specs。
+
+    仅当 ``spec.dynamic`` 为 ``True`` 且任务成功时提取。返回值可以是单个
+    :class:`TaskSpec` 或其列表。提取后，原结果的 ``value`` 被替换为生成的
+    任务名列表（便于查询），specs 列表返回给调用方注册调度。
+    """
+    if not spec.dynamic or result.status != TaskStatus.SUCCESS:
+        return []
+    value = result.value
+    spawned: list[TaskSpec[Any]] = []
+    if isinstance(value, TaskSpec):
+        spawned = [value]
+    elif isinstance(value, list) and value and all(isinstance(x, TaskSpec) for x in value):
+        spawned = value
+    if not spawned:
+        return []
+    return spawned
+
+
 class DependencyRunner:
     """依赖驱动调度：任务在硬/软依赖完成后立即启动，无层屏障。
 
@@ -746,10 +772,13 @@ class DependencyRunner:
     ) -> None:
         all_names = list(graph.all_specs().keys())
         semaphores = _build_semaphores(all_names, graph, asyncio.Semaphore, concurrency_limits)
-        futures: dict[str, asyncio.Future[TaskResult[Any]]] = {}
+        futures: dict[str, asyncio.Task[TaskResult[Any]]] = {}
 
         async def _run_task(name: str) -> TaskResult[Any]:
             spec = graph.resolved_spec(name)
+            # 检查点恢复：已在 report 中的 SUCCESS 任务直接返回
+            if name in report.results and report.results[name].status == TaskStatus.SUCCESS:
+                return report.results[name]
             # 等待所有硬依赖完成
             for dep in spec.depends_on:
                 if dep in futures:
@@ -777,13 +806,53 @@ class DependencyRunner:
 
             sem = _get_sem(semaphores, spec)
             result = await AsyncTaskRunner.run(spec, task_ctx, None, on_event, report, sem)
-            _store_result(name, result, spec, task_ctx, context, report, backend, on_event)
+
+            # 动态任务生成：提取 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)
+                # 重新存储已修改的 result
+                context[name] = result.value
+                report.results[name] = result
+                _emit(on_event, result)
+            else:
+                _store_result(name, result, spec, task_ctx, context, report, backend, on_event)
+
+            # 注册并调度动态生成的 specs
+            for raw_spec in spawned:
+                # 确保新 spec 依赖生成方（保证顺序与上下文可见性）
+                final_spec = raw_spec
+                if name not in raw_spec.depends_on and name not in raw_spec.soft_depends_on:
+                    final_spec = dc_replace(raw_spec, depends_on=(*raw_spec.depends_on, name))
+                graph._register_single(final_spec)
+                # 为新 spec 的 concurrency_key 创建信号量
+                if final_spec.concurrency_key and final_spec.concurrency_key not in semaphores:
+                    limit = concurrency_limits.get(final_spec.concurrency_key, 1)
+                    semaphores[final_spec.concurrency_key] = asyncio.Semaphore(limit)
+                futures[final_spec.name] = loop.create_task(_run_task(final_spec.name))
+
             return result
 
         loop = asyncio.get_running_loop()
         for name in all_names:
             futures[name] = loop.create_task(_run_task(name))
-        await asyncio.gather(*futures.values())
+
+        # 循环等待所有 futures（含动态生成的），直到全部完成。
+        # asyncio.wait 不像 gather 自动传播异常，需手动检查并 re-raise，
+        # 匹配 gather 的 fail-fast 行为（首个异常即取消剩余任务并抛出）。
+        while futures:
+            await asyncio.wait(futures.values(), return_when=asyncio.FIRST_COMPLETED)
+            completed_names = [n for n, f in futures.items() if f.done()]
+            for done_name in completed_names:
+                f = futures[done_name]
+                exc = f.exception()
+                if exc is not None:
+                    for remaining in futures.values():
+                        if not remaining.done():
+                            remaining.cancel()
+                    raise exc
+                del futures[done_name]
 
 
 # ---------------------------------------------------------------------- #
@@ -878,7 +947,16 @@ def _dispatch_strategy(
     limits: Mapping[str, int],
     cancel_event: threading.Event | CancelToken | None,
 ) -> None:
-    """按策略派发执行，将策略分支从 :func:`run` 中提取出来控制 PLR0912。"""
+    """按策略派发执行，将策略分支从 :func:`run` 中提取出来控制 PLR0912。
+
+    ``dynamic=True`` 任务仅在 ``dependency`` 策略下支持（需运行时扩展图）。
+    """
+    has_dynamic = any(s.dynamic for s in graph.all_specs().values())
+    if has_dynamic and strategy != "dependency":
+        raise ValueError(
+            f"dynamic=True 任务仅支持 strategy='dependency'，当前策略: {strategy!r}。"
+            "动态任务生成需要运行时扩展图，层屏障模型（sequential/thread/async）不支持。"
+        )
     if strategy == "sequential":
         layers = graph.layers()
         _drive_sequential(graph, layers, context, report, backend, effective_callback, cancel_event)
@@ -917,6 +995,37 @@ def _make_verbose_callback(on_event: EventCallback | None) -> EventCallback:
     return _verbose_callback
 
 
+def _load_resume_report(source: RunReport | str | Path) -> RunReport:
+    """从 RunReport 或 JSON 文件路径加载前次运行报告。"""
+    if isinstance(source, RunReport):
+        return source
+    return RunReport.from_json(Path(source).read_text())
+
+
+def _apply_resume(
+    resume_from: RunReport | str | Path | None,
+    graph: Graph,
+    report: RunReport,
+    context: dict[str, Any],
+) -> None:
+    """从前次报告中预填充 SUCCESS 任务的 result 和 context。
+
+    执行器在遇到已存在 result 时自动跳过（_filter_and_sort / _run_task）。
+    """
+    if resume_from is None:
+        return
+    prev_report = _load_resume_report(resume_from)
+    all_specs = graph.all_specs()
+    restored = 0
+    for name, result in prev_report.results.items():
+        if result.status == TaskStatus.SUCCESS and name in all_specs:
+            report.results[name] = dc_replace(result, reason="从检查点恢复")
+            context[name] = result.value
+            restored += 1
+    if restored:
+        logger.info("从检查点恢复 %d 个任务", restored, extra={"restored": restored})
+
+
 def run(
     graph: Graph,
     strategy: Strategy = "dependency",
@@ -932,6 +1041,7 @@ def run(
     cancel_event: threading.Event | CancelToken | None = None,
     only: Iterable[str] | None = None,
     tags: Iterable[str] | None = None,
+    resume_from: RunReport | str | Path | None = None,
     _report: RunReport | None = None,
 ) -> RunReport:
     """执行图并返回 :class:`RunReport`。
@@ -974,6 +1084,10 @@ def run(
         只运行指定任务名及其传递依赖。与 ``tags`` 取并集。
     tags:
         只运行匹配任意标签的任务及其传递依赖。与 ``only`` 取并集。
+    resume_from:
+        从之前的运行报告恢复：传入 :class:`RunReport` 或 JSON 文件路径。
+        前次运行中 SUCCESS 的任务会被跳过（标记 reason="从检查点恢复"），
+        FAILED/SKIPPED 的任务及其下游会被重新执行。
 
     抛出
     ----
@@ -1010,6 +1124,9 @@ def run(
     context: dict[str, Any] = {}
     limits = concurrency_limits or {}
 
+    # 检查点恢复：预填充 SUCCESS 任务的 result 和 context，执行器自动跳过。
+    _apply_resume(resume_from, graph, report, context)
+
     logger.info(
         "运行开始: run_id=%s strategy=%s tasks=%d",
         report.run_id,

src/pyflowx/graph.py

@@ -196,6 +196,14 @@ class Graph:
     # 在 specs 变更时失效。
     _layers_cache: list[list[str]] | None = field(default=None)
 
+    # loop 组映射：原 spec 名 → 展开后的实例名列表。
+    # 用于依赖重写：其他任务引用 loop 原名时，自动展开为依赖所有实例。
+    _loop_groups: dict[str, list[str]] = field(default_factory=dict)
+
+    # 任务分组映射：组名 → 组内任务名元组。
+    # 用于依赖重写：引用组名等价于依赖组内所有任务。
+    _groups: dict[str, tuple[str, ...]] = field(default_factory=dict)
+
     # ------------------------------------------------------------------ #
     # 构建
     # ------------------------------------------------------------------ #
@@ -227,7 +235,70 @@ class Graph:
             prev_name = current.name
         return self
 
+    def pipeline(self, *specs: TaskSpec[Any]) -> Graph:
+        """创建数据流水线：每个任务的结果通过上下文注入传给下一个。
+
+        与 :meth:`chain` 行为一致（依赖关系相同），但语义上强调数据流：
+        前驱任务的返回值会自动注入后继任务的同名参数。
+
+        后继任务的 ``fn`` 参数名需匹配前驱任务名才能接收数据。
+        ``outputs`` 字段声明的命名输出可通过
+        :meth:`RunReport.output_of` 查询。
+
+        Examples
+        --------
+        >>> graph = px.Graph().pipeline(extract_spec, transform_spec, load_spec)
+        >>> report = px.run(graph)
+        >>> report.output_of("extract", "items")
+        """
+        return self.chain(*specs)
+
+    def group(self, name: str, tasks: Sequence[str]) -> Graph:
+        """声明任务分组，引用组名等价于依赖组内所有任务。
+
+        组名必须不与任何已注册任务名冲突。组内任务必须已注册（或在本批
+        ``from_specs`` 收集阶段先于引用方注册）。声明后，后续 ``add`` 或
+        ``from_specs`` 中 ``depends_on``/``soft_depends_on`` 引用组名时，
+        会被重写为依赖组内全部任务。
+
+        参数
+        ----
+        name:
+            组名。不能与 ``self.specs`` 中任务名重复，也不能与已有组名重复。
+        tasks:
+            组内任务名序列。所有任务必须已在图中注册。
+
+        返回 ``self`` 支持链式调用。
+
+        Raises
+        ------
+        ValueError
+            组名与任务名冲突、组名已存在、或组内任务未注册时。
+        """
+        if not name:
+            raise ValueError("group.name 必须为非空字符串。")
+        if name in self.specs:
+            raise ValueError(f"组名 {name!r} 与已注册任务名冲突。")
+        if name in self._groups:
+            raise ValueError(f"组名 {name!r} 已存在。")
+        missing = [t for t in tasks if t not in self.specs]
+        if missing:
+            raise ValueError(f"组 {name!r} 引用了未注册的任务: {missing}")
+        if not tasks:
+            raise ValueError(f"组 {name!r} 不能为空。")
+        self._groups[name] = tuple(tasks)
+        return self
+
     def _register(self, spec: TaskSpec[Any]) -> None:
+        """注册单个 spec。若 ``spec.loop`` 非 ``None``，展开为多实例。"""
+        if spec.loop is not None:
+            for expanded in self._expand_loop_to_specs(spec):
+                self._register_single(expanded)
+            return
+        self._register_single(self._rewrite_deps(spec))
+
+    def _register_single(self, spec: TaskSpec[Any]) -> None:
+        """注册单个已展开的 spec（无 loop）。"""
         if spec.name in self.specs:
             raise DuplicateTaskError(spec.name)
         self.specs[spec.name] = spec
@@ -236,6 +307,52 @@ class Graph:
         self._resolved_cache.clear()
         self._layers_cache = None
 
+    def _expand_loop_to_specs(self, spec: TaskSpec[Any]) -> list[TaskSpec[Any]]:
+        """展开 ``spec.loop`` 为多实例列表，并记录到 ``_loop_groups``。
+
+        返回展开后的 spec 列表（``loop`` 字段已清空，``args`` 已追加 item）。
+        不写入 ``specs``/``deps``，由调用方注册。
+        """
+        loop = spec.loop
+        assert loop is not None  # 由 _register 保证
+        expanded: list[TaskSpec[Any]] = []
+        expanded_names: list[str] = []
+        for i, item in enumerate(loop.items):
+            key = loop.key_fn(i, item) if loop.key_fn is not None else f"{spec.name}_{i}"
+            new_spec = replace(spec, name=key, args=(*spec.args, item), loop=None)
+            new_spec = self._rewrite_deps(new_spec)
+            expanded.append(new_spec)
+            expanded_names.append(key)
+        self._loop_groups[spec.name] = expanded_names
+        return expanded
+
+    def _rewrite_deps(self, spec: TaskSpec[Any]) -> TaskSpec[Any]:
+        """重写 spec 的依赖：引用 loop 组名或 group 组名时展开为所有实例。"""
+        if not self._loop_groups and not self._groups:
+            return spec
+        # 合并 loop 组与普通组：组名 → 展开后的任务名列表
+        combined: dict[str, Sequence[str]] = {}
+        combined.update(self._loop_groups)
+        combined.update(self._groups)
+        new_hard: list[str] = []
+        new_soft: list[str] = []
+        changed = False
+        for dep in spec.depends_on:
+            if dep in combined:
+                new_hard.extend(combined[dep])
+                changed = True
+            else:
+                new_hard.append(dep)
+        for dep in spec.soft_depends_on:
+            if dep in combined:
+                new_soft.extend(combined[dep])
+                changed = True
+            else:
+                new_soft.append(dep)
+        if not changed:
+            return spec
+        return replace(spec, depends_on=tuple(new_hard), soft_depends_on=tuple(new_soft))
+
     @classmethod
     def from_specs(
         cls,
@@ -260,19 +377,31 @@ class Graph:
         """
         graph = cls(defaults=defaults or GraphDefaults(), namespace=namespace)
         pending_refs: list[str] = []
-        # 批量注册：直接写入 specs/deps，跳过每次 _register 的 cache 清空（N 次清空降为 0 次）。
-        # 重复名检测在循环内做；cache 失效由后续 validate()/layers() 首次调用时自然触发。
+        collected: list[TaskSpec[Any]] = []
         for spec in specs:
             if isinstance(spec, str):
                 pending_refs.append(spec)
             elif isinstance(spec, TaskSpec):
-                if spec.name in graph.specs:
-                    raise DuplicateTaskError(spec.name)
-                graph.specs[spec.name] = spec
-                graph.deps[spec.name] = spec.depends_on
+                collected.append(spec)
             else:
                 raise TypeError(f"from_specs 只接受 TaskSpec 或 str，收到: {type(spec)}")
 
+        # 两阶段：先展开所有 loop（记录 _loop_groups），再统一重写依赖并批量注册。
+        # 批量注册跳过每次 _register 的 cache 清空（N 次清空降为 0 次）；
+        # cache 失效由后续 validate()/layers() 首次调用自然触发。
+        expanded: list[TaskSpec[Any]] = []
+        for spec in collected:
+            if spec.loop is not None:
+                expanded.extend(graph._expand_loop_to_specs(spec))
+            else:
+                expanded.append(graph._rewrite_deps(spec))
+
+        for spec in expanded:
+            if spec.name in graph.specs:
+                raise DuplicateTaskError(spec.name)
+            graph.specs[spec.name] = spec
+            graph.deps[spec.name] = spec.depends_on
+
         if pending_refs:
             graph._pending_refs = pending_refs
 

src/pyflowx/history.py

@@ -0,0 +1,87 @@
+"""运行历史管理：存储、查询与比较多次运行的 :class:`RunReport`。
+
+适用于：
+* 断点续跑：从最近一次失败运行恢复，跳过已成功任务。
+* 趋势分析：对比多次运行的任务耗时、状态变化。
+* 审计追溯：按 run_id 查询历史报告。
+
+用法
+----
+    history = RunHistory(".pyflowx/runs")
+    history.save(report)
+    latest = history.latest()
+    if latest is not None and not latest.success:
+        # 从最近失败处恢复
+        px.run(graph, resume_from=latest)
+"""
+
+from __future__ import annotations
+
+__all__ = ["RunHistory"]
+
+from pathlib import Path
+
+from .report import RunReport
+
+
+class RunHistory:
+    """运行历史记录管理器。
+
+    将 :class:`RunReport` 以 JSON 文件存储到指定目录，文件名为
+    ``{run_id}.json``。支持按 run_id 加载、列出全部、查询最近一次、删除。
+
+    Parameters
+    ----------
+    dir:
+        存储目录。不存在时自动创建。
+    """
+
+    def __init__(self, dir: str | Path) -> None:
+        self.dir = Path(dir)
+        self.dir.mkdir(parents=True, exist_ok=True)
+
+    def save(self, report: RunReport) -> Path:
+        """保存运行报告，返回文件路径。"""
+        path = self.dir / f"{report.run_id}.json"
+        path.write_text(report.to_json(), encoding="utf-8")
+        return path
+
+    def load(self, run_id: str) -> RunReport:
+        """按 run_id 加载报告。
+
+        Raises
+        ------
+        FileNotFoundError
+            指定 run_id 的文件不存在。
+        """
+        path = self.dir / f"{run_id}.json"
+        if not path.exists():
+            raise FileNotFoundError(f"运行报告 {run_id!r} 不存在: {path}")
+        return RunReport.from_json(path.read_text(encoding="utf-8"))
+
+    def list_runs(self) -> list[str]:
+        """列出全部 run_id（按文件名排序）。"""
+        return sorted(p.stem for p in self.dir.glob("*.json"))
+
+    def latest(self) -> RunReport | None:
+        """返回最近修改的报告；目录为空时返回 ``None``。"""
+        runs = sorted(self.dir.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
+        if not runs:
+            return None
+        return RunReport.from_json(runs[0].read_text(encoding="utf-8"))
+
+    def delete(self, run_id: str) -> bool:
+        """删除指定 run_id 的报告。返回是否实际删除。"""
+        path = self.dir / f"{run_id}.json"
+        if path.exists():
+            path.unlink()
+            return True
+        return False
+
+    def __len__(self) -> int:
+        return sum(1 for _ in self.dir.glob("*.json"))
+
+    def __contains__(self, run_id: object) -> bool:
+        if not isinstance(run_id, str):
+            return False
+        return (self.dir / f"{run_id}.json").exists()

src/pyflowx/monitoring.py

@@ -0,0 +1,217 @@
+"""监控导出：Prometheus 指标收集与健康检查。
+
+提供零依赖的轻量监控方案：
+
+* :class:`MetricsCollector` —— 通过 ``on_event`` 回调收集任务指标，
+  导出 Prometheus 文本格式。可配合 ``http.server`` 或 ``prometheus_client``
+  暴露指标端点。
+* :func:`health_check` —— 分析 :class:`RunReport` 返回结构化健康状态。
+
+用法
+----
+    collector = px.MetricsCollector()
+    report = px.run(graph, on_event=collector.on_event)
+    print(collector.metrics_text())
+    print(px.health_check(report))
+"""
+
+from __future__ import annotations
+
+__all__ = ["MetricsCollector", "health_check", "start_metrics_server"]
+
+from collections.abc import Callable
+from typing import Any
+
+from .report import RunReport
+from .task import TaskEvent, TaskStatus
+
+
+class MetricsCollector:
+    """收集任务执行指标，导出 Prometheus 文本格式。
+
+    作为 :func:`run` 的 ``on_event`` 回调使用，收集每个任务的生命周期
+    事件并聚合为 Prometheus 指标。线程安全（回调可能从多个线程调用）。
+
+    指标列表
+    --------
+    - ``pyflowx_task_total{task, status}`` — 任务计数器
+    - ``pyflowx_task_duration_seconds{task}`` — 任务耗时（gauge，最近值）
+    - ``pyflowx_task_duration_seconds_sum{task}`` — 任务累计耗时
+    - ``pyflowx_task_retries_total{task}`` — 重试次数
+    - ``pyflowx_run_total{status}`` — 运行计数器
+    """
+
+    def __init__(self) -> None:
+        self._task_count: dict[str, dict[str, int]] = {}
+        self._task_duration_last: dict[str, float] = {}
+        self._task_duration_sum: dict[str, float] = {}
+        self._task_retries: dict[str, int] = {}
+        self._lock_counts: dict[str, int] = {}
+
+    def on_event(self, event: TaskEvent) -> None:
+        """``on_event`` 回调：记录任务生命周期事件。"""
+        task = event.task
+        status = event.status.value
+
+        # 任务计数
+        self._task_count.setdefault(task, {})
+        self._task_count[task][status] = self._task_count[task].get(status, 0) + 1
+
+        # 耗时（仅在终态时记录）
+        if event.duration is not None and event.status in (TaskStatus.SUCCESS, TaskStatus.FAILED):
+            self._task_duration_last[task] = event.duration
+            self._task_duration_sum[task] = self._task_duration_sum.get(task, 0.0) + event.duration
+
+        # 重试次数
+        if event.attempts > 1 and event.status in (TaskStatus.SUCCESS, TaskStatus.FAILED):
+            self._task_retries[task] = self._task_retries.get(task, 0) + event.attempts - 1
+
+    def record_run(self, report: RunReport) -> None:
+        """记录一次完整运行的结果。"""
+        status = "success" if report.success else "failed"
+        self._lock_counts[status] = self._lock_counts.get(status, 0) + 1
+
+    def metrics_text(self) -> str:
+        """导出 Prometheus 文本格式指标字符串。"""
+        lines: list[str] = []
+
+        # pyflowx_task_total
+        if self._task_count:
+            lines.append("# HELP pyflowx_task_total Total tasks by status.")
+            lines.append("# TYPE pyflowx_task_total counter")
+            for task in sorted(self._task_count):
+                for status in sorted(self._task_count[task]):
+                    count = self._task_count[task][status]
+                    lines.append(f'pyflowx_task_total{{task="{task}",status="{status}"}} {count}')
+
+        # pyflowx_task_duration_seconds (last)
+        if self._task_duration_last:
+            lines.append("# HELP pyflowx_task_duration_seconds Last task duration in seconds.")
+            lines.append("# TYPE pyflowx_task_duration_seconds gauge")
+            for task in sorted(self._task_duration_last):
+                lines.append(f'pyflowx_task_duration_seconds{{task="{task}"}} {self._task_duration_last[task]}')
+
+        # pyflowx_task_duration_seconds_sum
+        if self._task_duration_sum:
+            lines.append("# HELP pyflowx_task_duration_seconds_sum Total task duration in seconds.")
+            lines.append("# TYPE pyflowx_task_duration_seconds_sum counter")
+            for task in sorted(self._task_duration_sum):
+                lines.append(f'pyflowx_task_duration_seconds_sum{{task="{task}"}} {self._task_duration_sum[task]}')
+
+        # pyflowx_task_retries_total
+        if self._task_retries:
+            lines.append("# HELP pyflowx_task_retries_total Total task retries.")
+            lines.append("# TYPE pyflowx_task_retries_total counter")
+            for task in sorted(self._task_retries):
+                lines.append(f'pyflowx_task_retries_total{{task="{task}"}} {self._task_retries[task]}')
+
+        # pyflowx_run_total
+        if self._lock_counts:
+            lines.append("# HELP pyflowx_run_total Total runs by status.")
+            lines.append("# TYPE pyflowx_run_total counter")
+            for status in sorted(self._lock_counts):
+                lines.append(f'pyflowx_run_total{{status="{status}"}} {self._lock_counts[status]}')
+
+        return "\n".join(lines) + "\n"
+
+    def reset(self) -> None:
+        """清空所有收集的指标。"""
+        self._task_count.clear()
+        self._task_duration_last.clear()
+        self._task_duration_sum.clear()
+        self._task_retries.clear()
+        self._lock_counts.clear()
+
+
+def health_check(report: RunReport) -> dict[str, Any]:
+    """分析运行报告，返回结构化健康状态。
+
+    状态判定：
+    - ``healthy`` — 全部任务 SUCCESS
+    - ``degraded`` — 部分（非全部）任务 FAILED
+    - ``unhealthy`` — 全部任务 FAILED
+    - ``unknown`` — 无任务
+
+    Returns
+    -------
+    dict
+        包含 ``status``/``total``/``success``/``failed``/``skipped``
+        /``duration`` 字段。
+    """
+    total = len(report.results)
+    if total == 0:
+        return {"status": "unknown", "total": 0, "message": "无任务结果"}
+
+    counts: dict[str, int] = {}
+    total_duration = 0.0
+    for r in report.results.values():
+        status = r.status.value
+        counts[status] = counts.get(status, 0) + 1
+        if r.duration is not None:
+            total_duration += r.duration
+
+    success = counts.get("success", 0)
+    failed = counts.get("failed", 0)
+    skipped = counts.get("skipped", 0)
+
+    if failed == 0:
+        status = "healthy"
+    elif failed < total:
+        status = "degraded"
+    else:
+        status = "unhealthy"
+
+    return {
+        "status": status,
+        "total": total,
+        "success": success,
+        "failed": failed,
+        "skipped": skipped,
+        "duration": round(total_duration, 3),
+    }
+
+
+def start_metrics_server(collector: MetricsCollector, port: int = 9100, host: str = "0.0.0.0") -> Callable[[], None]:
+    """启动简单 HTTP 服务器暴露 Prometheus 指标。
+
+    返回停止函数，调用后关闭服务器。阻塞调用方线程，建议在后台线程运行。
+
+    .. note::
+       使用标准库 ``http.server``，零依赖。生产环境建议配合
+       ``prometheus_client`` 或反向代理使用。
+    """
+    import sys
+    import threading
+    from http.server import BaseHTTPRequestHandler, HTTPServer
+
+    if sys.version_info >= (3, 12):
+        from typing import override
+    else:
+        from typing_extensions import override  # pragma: no cover
+
+    class _Handler(BaseHTTPRequestHandler):
+        def do_GET(self) -> None:
+            if self.path == "/metrics":
+                body = collector.metrics_text().encode("utf-8")
+                self.send_response(200)
+                self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                self.wfile.write(body)
+            else:
+                self.send_response(404)
+                self.end_headers()
+
+        @override
+        def log_message(self, format: str, *args: object) -> None:
+            pass  # 静默访问日志
+
+    server = HTTPServer((host, port), _Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+
+    def stop() -> None:
+        server.shutdown()
+        server.server_close()
+
+    return stop

src/pyflowx/report.py

@@ -28,6 +28,24 @@ def _noop_fn() -> None:
     return None
 
 
+def _resolve_output_path(result: Any, path: str) -> Any:
+    """从任务结果中按点分路径提取命名输出值。
+
+    路径语法：
+    - ``"$"`` → 整个结果
+    - ``"key"`` → ``result["key"]``
+    - ``"a.b.c"`` → ``result["a"]["b"]["c"]``
+
+    适用于 dict 结果；非 dict 结果仅支持 ``"$"``。
+    """
+    if path == "$":
+        return result
+    value = result
+    for key in path.split("."):
+        value = value[key]
+    return value
+
+
 def _serialize_value(
     value: Any,
     value_serializer: Callable[[Any], Any] | None = None,
@@ -79,6 +97,34 @@ class RunReport:
         """返回 ``name`` 的完整 :class:`TaskResult`。"""
         return self.results[name]
 
+    def output_of(self, task_name: str, output_name: str) -> Any:
+        """获取任务的命名输出。
+
+        根据 :attr:`TaskSpec.outputs` 中声明的路径从任务结果中提取。
+        例如 ``outputs={"artifact": "path"}`` 声明后，
+        ``report.output_of("build", "artifact")`` 返回 ``result["path"]``。
+
+        Parameters
+        ----------
+        task_name:
+            任务名。
+        output_name:
+            ``outputs`` 映射中的键名。
+
+        Raises
+        ------
+        KeyError
+            任务不存在、任务未声明 outputs、或 output_name 未声明时。
+        """
+        result = self.results.get(task_name)
+        if result is None:
+            raise KeyError(f"任务 {task_name!r} 不在报告中。")
+        spec = result.spec
+        if spec.outputs is None or output_name not in spec.outputs:
+            raise KeyError(f"任务 {task_name!r} 未声明输出 {output_name!r}。")
+        path = spec.outputs[output_name]
+        return _resolve_output_path(result.value, path)
+
     def __contains__(self, name: Any) -> bool:
         return name in self.results
 
@@ -203,20 +249,22 @@ class RunReport:
         """
         results_list: list[dict[str, Any]] = []
         for name, r in self.results.items():
-            results_list.append({
-                "name": name,
-                "status": r.status.value,
-                "value": _serialize_value(r.value, value_serializer),
-                "error": repr(r.error) if r.error else None,
-                "attempts": r.attempts,
-                "started_at": r.started_at.isoformat() if r.started_at else None,
-                "finished_at": r.finished_at.isoformat() if r.finished_at else None,
-                "duration_seconds": r.duration,
-                "reason": r.reason,
-                "tags": list(r.spec.tags),
-                "depends_on": list(r.spec.depends_on),
-                "outputs": dict(r.spec.outputs) if r.spec.outputs else None,
-            })
+            results_list.append(
+                {
+                    "name": name,
+                    "status": r.status.value,
+                    "value": _serialize_value(r.value, value_serializer),
+                    "error": repr(r.error) if r.error else None,
+                    "attempts": r.attempts,
+                    "started_at": r.started_at.isoformat() if r.started_at else None,
+                    "finished_at": r.finished_at.isoformat() if r.finished_at else None,
+                    "duration_seconds": r.duration,
+                    "reason": r.reason,
+                    "tags": list(r.spec.tags),
+                    "depends_on": list(r.spec.depends_on),
+                    "outputs": dict(r.spec.outputs) if r.spec.outputs else None,
+                }
+            )
         return {
             "run_id": self.run_id,
             "success": self.success,

src/pyflowx/task.py

@@ -22,7 +22,7 @@ import os
 import shutil
 import sys
 import threading
-from collections.abc import Callable, Coroutine, Generator, Mapping
+from collections.abc import Callable, Coroutine, Generator, Iterable, Mapping
 from contextlib import AbstractContextManager, contextmanager
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -64,6 +64,65 @@ Condition = Callable[[Context], bool]
 CacheKeyFn = Callable[[Context], str]
 
 
+# ---------------------------------------------------------------------- #
+# 任务循环
+# ---------------------------------------------------------------------- #
+@dataclass(frozen=True, slots=True)
+class LoopSpec:
+    """任务循环展开规格。
+
+    添加到 :class:`Graph` 时，``loop`` 非 ``None`` 的 :class:`TaskSpec`
+    会被展开为多实例：每个 ``item`` 生成一个新 spec，``item`` 追加到
+    ``args`` 末尾，``loop`` 字段清空。
+
+    展开后的实例名默认为 ``f"{name}_{i}"``（``i`` 从 0 起的索引），
+    也可通过 ``key_fn`` 自定义。其他任务引用 loop 原名时，``Graph``
+    会自动重写为依赖所有展开实例。
+
+    参数
+    ----
+    items:
+        待迭代的项元组。
+    key_fn:
+        接受 ``(index, item)``，返回实例名。``None`` 时用
+        ``f"{name}_{i}"``。
+
+    Examples
+    --------
+    >>> from pyflowx import LoopSpec
+    >>> spec = px.task(process, loop=LoopSpec.range(3))
+    >>> # 等价于手动创建 process_0, process_1, process_2 三个任务
+    """
+
+    items: tuple[Any, ...]
+    key_fn: Callable[[int, Any], str] | None = None
+
+    def __post_init__(self) -> None:
+        if not self.items:
+            raise ValueError("LoopSpec.items 不能为空。")
+
+    @classmethod
+    def range(cls, *args: int) -> LoopSpec:
+        """构造整数序列循环（签名与 :func:`range` 一致）。
+
+        支持三种形式：
+        - ``range(stop)`` → ``[0, 1, ..., stop-1]``
+        - ``range(start, stop)`` → ``[start, ..., stop-1]``
+        - ``range(start, stop, step)`` → 按步长递增/递减
+        """
+        # classmethod 内的 ``range`` 查找内置函数，不引用本类 LoopSpec.range
+        return cls(items=tuple(range(*args)))
+
+    @classmethod
+    def from_iterable(
+        cls,
+        items: Iterable[Any],
+        key_fn: Callable[[int, Any], str] | None = None,
+    ) -> LoopSpec:
+        """从可迭代对象构造循环。"""
+        return cls(items=tuple(items), key_fn=key_fn)
+
+
 def _format_skip_reason(failed_conditions: list[str]) -> str:
     """格式化跳过原因：≤2 个全展示，>2 个仅展示前 2 个并附总数。"""
     if len(failed_conditions) <= 2:
@@ -274,6 +333,8 @@ class TaskSpec(Generic[T]):
     hooks: TaskHooks = field(default_factory=TaskHooks)
     executor: str = "thread"  # "thread" | "process" | "inline"
     outputs: Mapping[str, str] | None = None
+    loop: LoopSpec | None = None
+    dynamic: bool = False
 
     def __post_init__(self) -> None:
         if not self.name:
@@ -474,6 +535,8 @@ def task(
     cache_key: CacheKeyFn | None = None,
     hooks: TaskHooks | None = None,
     outputs: Mapping[str, str] | None = None,
+    loop: LoopSpec | None = None,
+    dynamic: bool = False,
     name: str | None = None,
 ) -> Any:
     """装饰器：将函数转为 :class:`TaskSpec`。
@@ -516,6 +579,8 @@ def task(
             cache_key=cache_key,
             hooks=hooks if hooks is not None else TaskHooks(),
             outputs=dict(outputs) if outputs else None,
+            loop=loop,
+            dynamic=dynamic,
         )
 
     if fn is None and cmd is None:

tests/test_dataflow.py

@@ -0,0 +1,153 @@
+"""数据流增强测试：outputs 引用解析、pipeline 语义化管道。
+
+覆盖：
+* RunReport.output_of() 基本路径解析（"$" / 单键 / 嵌套键）。
+* output_of() 错误场景（任务不存在 / 未声明 outputs / output_name 未声明）。
+* Graph.pipeline() 语义化管道：数据流通过上下文注入传递。
+* pipeline() 与 chain() 行为一致。
+* outputs 在 to_dict 序列化中保留。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+import pyflowx as px
+from pyflowx import Graph, TaskSpec
+
+
+def test_output_of_entire_result() -> None:
+    """outputs={"value": "$"} → output_of 返回整个结果。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 42, outputs={"value": "$"})])
+    report = px.run(graph, strategy="sequential")
+    assert report.output_of("a", "value") == 42
+
+
+def test_output_of_dict_key() -> None:
+    """outputs={"path": "url"} → output_of 返回 result["url"]。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: {"url": "http://x", "code": 200}, outputs={"path": "url"})])
+    report = px.run(graph, strategy="sequential")
+    assert report.output_of("a", "path") == "http://x"
+
+
+def test_output_of_nested_path() -> None:
+    """outputs={"host": "data.server.host"} → 嵌套字典路径解析。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec(
+                "a",
+                fn=lambda: {"data": {"server": {"host": "localhost"}}},
+                outputs={"host": "data.server.host"},
+            )
+        ]
+    )
+    report = px.run(graph, strategy="sequential")
+    assert report.output_of("a", "host") == "localhost"
+
+
+def test_output_of_task_not_found() -> None:
+    """任务不在报告中时抛 KeyError。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+    report = px.run(graph, strategy="sequential")
+    with pytest.raises(KeyError, match="不在报告中"):
+        report.output_of("nonexistent", "x")
+
+
+def test_output_of_no_outputs_declared() -> None:
+    """任务未声明 outputs 时抛 KeyError。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+    report = px.run(graph, strategy="sequential")
+    with pytest.raises(KeyError, match="未声明输出"):
+        report.output_of("a", "x")
+
+
+def test_output_of_output_name_not_declared() -> None:
+    """output_name 不在 outputs 映射中时抛 KeyError。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1, outputs={"x": "$"})])
+    report = px.run(graph, strategy="sequential")
+    with pytest.raises(KeyError, match="未声明输出"):
+        report.output_of("a", "y")
+
+
+def test_pipeline_data_flow() -> None:
+    """pipeline() 中前驱结果注入后继同名参数。"""
+
+    def extract() -> list[int]:
+        return [1, 2, 3]
+
+    def double(extract: list[int]) -> list[int]:
+        return [x * 2 for x in extract]
+
+    def total(double: list[int]) -> int:
+        return sum(double)
+
+    graph = Graph().pipeline(
+        TaskSpec("extract", fn=extract),
+        TaskSpec("double", fn=double),
+        TaskSpec("total", fn=total),
+    )
+    report = px.run(graph, strategy="sequential")
+    assert report["extract"] == [1, 2, 3]
+    assert report["double"] == [2, 4, 6]
+    assert report["total"] == 12
+
+
+def test_pipeline_same_as_chain() -> None:
+    """pipeline() 与 chain() 产生相同的依赖关系。"""
+    specs1 = [
+        TaskSpec("a", fn=lambda: 1),
+        TaskSpec("b", fn=lambda a: a + 1),
+        TaskSpec("c", fn=lambda b: b + 1),
+    ]
+    specs2 = [
+        TaskSpec("a", fn=lambda: 1),
+        TaskSpec("b", fn=lambda a: a + 1),
+        TaskSpec("c", fn=lambda b: b + 1),
+    ]
+    g1 = Graph().chain(*specs1)
+    g2 = Graph().pipeline(*specs2)
+    assert g1.deps == g2.deps
+
+
+def test_pipeline_with_outputs() -> None:
+    """pipeline() 配合 outputs 声明，通过 output_of 查询。"""
+
+    def make_data() -> dict[str, object]:
+        return {"items": [1, 2, 3], "count": 3}
+
+    graph = Graph().pipeline(
+        TaskSpec("make_data", fn=make_data, outputs={"items": "items", "count": "count"}),
+    )
+    report = px.run(graph, strategy="sequential")
+    assert report.output_of("make_data", "items") == [1, 2, 3]
+    assert report.output_of("make_data", "count") == 3
+
+
+def test_pipeline_returns_self() -> None:
+    """pipeline() 返回 self 支持链式调用。"""
+    graph = Graph()
+    result = graph.pipeline(TaskSpec("a", fn=lambda: 1))
+    assert result is graph
+
+
+def test_output_of_with_strategy_dependency() -> None:
+    """output_of 在 dependency 策略下也正常工作。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec(
+                "a",
+                fn=lambda: {"x": 10, "y": 20},
+                outputs={"x": "x", "y": "y"},
+            )
+        ]
+    )
+    report = px.run(graph, strategy="dependency")
+    assert report.output_of("a", "x") == 10
+    assert report.output_of("a", "y") == 20
+
+
+def test_output_of_list_result_with_dollar() -> None:
+    """列表结果用 "$" 路径取整个值。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: [1, 2, 3], outputs={"all": "$"})])
+    report = px.run(graph, strategy="sequential")
+    assert report.output_of("a", "all") == [1, 2, 3]

tests/test_dynamic.py

@@ -0,0 +1,174 @@
+"""动态任务生成（dynamic=True）测试。
+
+覆盖：
+* 基本动态生成：fn 返回单个 spec / list[spec]。
+* 生成任务 result.value 被替换为生成任务名列表。
+* 新 spec 自动依赖生成方。
+* 多级动态生成（生成的任务再生成）。
+* 非 dependency 策略下 dynamic=True 抛 ValueError。
+* 非动态任务不受影响。
+* 空列表 / None 返回值不生成任务。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+import pyflowx as px
+from pyflowx import Graph, TaskSpec
+
+
+def test_dynamic_single_spec() -> None:
+    """fn 返回单个 TaskSpec，动态注册并执行。"""
+
+    def generator() -> TaskSpec:
+        return TaskSpec("spawned", fn=lambda: 42)
+
+    graph = Graph.from_specs([TaskSpec("gen", fn=generator, dynamic=True)])
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    assert "spawned" in report.results
+    assert report["spawned"] == 42
+    # 生成任务的 result.value 被替换为生成任务名列表
+    assert report["gen"] == ["spawned"]
+
+
+def test_dynamic_list_of_specs() -> None:
+    """fn 返回 list[TaskSpec]，全部注册并执行。"""
+
+    def generator() -> list[TaskSpec]:
+        return [TaskSpec(f"task_{i}", fn=lambda i=i: i * 10) for i in range(3)]
+
+    graph = Graph.from_specs([TaskSpec("gen", fn=generator, dynamic=True)])
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    assert report["task_0"] == 0
+    assert report["task_1"] == 10
+    assert report["task_2"] == 20
+    assert report["gen"] == ["task_0", "task_1", "task_2"]
+
+
+def test_dynamic_auto_depends_on_generator() -> None:
+    """生成的 spec 自动依赖生成方，保证执行顺序。"""
+    order: list[str] = []
+
+    def generator() -> list[TaskSpec]:
+        order.append("gen")
+        return [TaskSpec("child", fn=lambda: order.append("child"))]
+
+    graph = Graph.from_specs([TaskSpec("gen", fn=generator, dynamic=True)])
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    assert order.index("gen") < order.index("child")
+
+
+def test_dynamic_multi_level() -> None:
+    """生成的任务本身也是 dynamic，再生成下一级。"""
+
+    def level1() -> list[TaskSpec]:
+        return [TaskSpec("level2", fn=level2, dynamic=True)]
+
+    def level2() -> list[TaskSpec]:
+        return [TaskSpec("leaf", fn=lambda: "done")]
+
+    graph = Graph.from_specs([TaskSpec("level1", fn=level1, dynamic=True)])
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    assert report["leaf"] == "done"
+    assert report["level2"] == ["leaf"]
+    assert report["level1"] == ["level2"]
+
+
+def test_dynamic_with_existing_deps() -> None:
+    """生成的 spec 可引用图中已有任务。"""
+
+    def base_task() -> int:
+        return 100
+
+    def generator(base_task: int) -> list[TaskSpec]:
+        return [TaskSpec("consumer", fn=lambda: base_task + 1)]
+
+    graph = Graph.from_specs(
+        [
+            TaskSpec("base_task", fn=base_task),
+            TaskSpec("gen", fn=generator, depends_on=("base_task",), dynamic=True),
+        ]
+    )
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    assert report["consumer"] == 101
+
+
+def test_dynamic_non_dependency_strategy_raises() -> None:
+    """非 dependency 策略下使用 dynamic=True 抛 ValueError。"""
+    graph = Graph.from_specs([TaskSpec("gen", fn=lambda: None, dynamic=True)])
+    with pytest.raises(ValueError, match="仅支持 strategy='dependency'"):
+        px.run(graph, strategy="sequential")
+
+
+def test_dynamic_none_return() -> None:
+    """dynamic 任务返回 None 时不生成新任务。"""
+
+    def generator() -> None:
+        return None
+
+    graph = Graph.from_specs([TaskSpec("gen", fn=generator, dynamic=True)])
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    # result.value 仍为 None（无生成任务）
+    assert report["gen"] is None
+
+
+def test_dynamic_empty_list_return() -> None:
+    """dynamic 任务返回空列表时不生成新任务。"""
+
+    def generator() -> list[TaskSpec]:
+        return []
+
+    graph = Graph.from_specs([TaskSpec("gen", fn=generator, dynamic=True)])
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    assert report["gen"] == []
+
+
+def test_non_dynamic_task_unaffected() -> None:
+    """非动态任务的行为不受影响。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=lambda: 1),
+            TaskSpec("b", fn=lambda a: a + 1, depends_on=("a",)),
+        ]
+    )
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    assert report["a"] == 1
+    assert report["b"] == 2
+
+
+def test_dynamic_with_task_decorator() -> None:
+    """@task 装饰器支持 dynamic 参数。"""
+
+    @px.task(dynamic=True)
+    def gen() -> list[TaskSpec]:
+        return [TaskSpec(f"s{i}", fn=lambda i=i: i) for i in range(3)]
+
+    graph = Graph.from_specs([gen])
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    assert report["s0"] == 0
+    assert report["s1"] == 1
+    assert report["s2"] == 2
+
+
+def test_dynamic_spec_with_explicit_depends_on() -> None:
+    """生成的 spec 已有 depends_on 时不重复添加生成方。"""
+
+    def generator() -> list[TaskSpec]:
+        return [TaskSpec("child", fn=lambda: 1, depends_on=("gen",))]
+
+    graph = Graph.from_specs([TaskSpec("gen", fn=generator, dynamic=True)])
+    report = px.run(graph, strategy="dependency")
+    assert report.success
+    # depends_on 不重复
+    child_spec = graph.specs["child"]
+    assert child_spec.depends_on.count("gen") == 1

tests/test_group.py

@@ -0,0 +1,190 @@
+"""任务分组（Graph.group）测试。
+
+覆盖：
+* 基本分组创建与依赖重写（硬依赖 / 软依赖）。
+* 组与 loop 组合：组内成员可包含 loop 展开后的实例名。
+* 校验：空组名、与任务名冲突、组名重复、组内任务未注册、空组列表。
+* 链式调用：group() 返回 self。
+* 执行：分组后图能正常跑通。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+import pyflowx as px
+from pyflowx import Graph, LoopSpec, TaskSpec
+
+
+def test_group_basic_creation() -> None:
+    """group() 记录到 _groups 字典并返回 self。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=lambda: 1),
+            TaskSpec("b", fn=lambda: 2),
+        ]
+    )
+    result = graph.group("phase1", ["a", "b"])
+    assert result is graph
+    assert graph._groups["phase1"] == ("a", "b")
+
+
+def test_group_dependency_rewrite_hard() -> None:
+    """引用组名的硬依赖被重写为组内全部任务。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=lambda: 1),
+            TaskSpec("b", fn=lambda: 2),
+        ]
+    )
+    graph.group("phase1", ["a", "b"])
+    graph.add(TaskSpec("c", fn=lambda: 3, depends_on=("phase1",)))
+    assert graph.specs["c"].depends_on == ("a", "b")
+
+
+def test_group_dependency_rewrite_soft() -> None:
+    """引用组名的软依赖被重写为组内全部任务。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=lambda: 1),
+            TaskSpec("b", fn=lambda: 2),
+        ]
+    )
+    graph.group("phase1", ["a", "b"])
+    graph.add(TaskSpec("c", fn=lambda: 3, soft_depends_on=("phase1",)))
+    assert graph.specs["c"].soft_depends_on == ("a", "b")
+
+
+def test_group_before_adding_depender() -> None:
+    """先声明 group 再注册引用方：from_specs 后 add 引用组名。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("extract", fn=lambda: [1, 2]),
+            TaskSpec("validate", fn=lambda: True),
+        ]
+    )
+    graph.group("prep", ["extract", "validate"])
+    graph.add(TaskSpec("process", fn=lambda: None, depends_on=("prep",)))
+    assert set(graph.specs["process"].depends_on) == {"extract", "validate"}
+
+
+def test_group_chained_with_loop() -> None:
+    """组内成员可包含 loop 展开后的实例名。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("proc", fn=lambda x: x * 2, loop=LoopSpec.range(3)),
+        ]
+    )
+    # proc 展开为 proc_0, proc_1, proc_2
+    assert {"proc_0", "proc_1", "proc_2"}.issubset(graph.specs.keys())
+    graph.group("proc_group", ["proc_0", "proc_1", "proc_2"])
+    graph.add(TaskSpec("post", fn=lambda: None, depends_on=("proc_group",)))
+    assert set(graph.specs["post"].depends_on) == {"proc_0", "proc_1", "proc_2"}
+
+
+def test_group_execution() -> None:
+    """分组后的图能正常执行。"""
+    calls: list[str] = []
+
+    def make(name: str):
+        def _f() -> None:
+            calls.append(name)
+
+        return _f
+
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=make("a")),
+            TaskSpec("b", fn=make("b")),
+            TaskSpec("c", fn=make("c")),
+        ]
+    )
+    graph.group("prep", ["a", "b"])
+    graph.add(TaskSpec("done", fn=make("done"), depends_on=("prep", "c")))
+    report = px.run(graph, strategy="sequential")
+    assert report.success
+    assert set(calls) == {"a", "b", "c", "done"}
+    # done 在 a/b/c 之后
+    assert calls.index("done") > calls.index("a")
+    assert calls.index("done") > calls.index("b")
+    assert calls.index("done") > calls.index("c")
+
+
+def test_group_empty_name_raises() -> None:
+    """空组名抛 ValueError。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+    with pytest.raises(ValueError, match="非空字符串"):
+        graph.group("", ["a"])
+
+
+def test_group_name_conflicts_with_task() -> None:
+    """组名与已注册任务名冲突时抛 ValueError。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+    with pytest.raises(ValueError, match="与已注册任务名冲突"):
+        graph.group("a", [])
+
+
+def test_group_duplicate_name_raises() -> None:
+    """重复声明同名组抛 ValueError。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=lambda: 1),
+            TaskSpec("b", fn=lambda: 2),
+        ]
+    )
+    graph.group("g1", ["a"])
+    with pytest.raises(ValueError, match="已存在"):
+        graph.group("g1", ["b"])
+
+
+def test_group_unregistered_task_raises() -> None:
+    """组内引用未注册的任务抛 ValueError。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+    with pytest.raises(ValueError, match="未注册的任务"):
+        graph.group("g1", ["a", "nonexistent"])
+
+
+def test_group_empty_tasks_raises() -> None:
+    """空组列表抛 ValueError。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+    with pytest.raises(ValueError, match="不能为空"):
+        graph.group("g1", [])
+
+
+def test_group_chain_returns_self() -> None:
+    """group() 返回 self 支持链式调用。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=lambda: 1),
+            TaskSpec("b", fn=lambda: 2),
+            TaskSpec("c", fn=lambda: 3),
+        ]
+    )
+    result = graph.group("g1", ["a"]).group("g2", ["b", "c"])
+    assert result is graph
+    assert graph._groups["g1"] == ("a",)
+    assert graph._groups["g2"] == ("b", "c")
+
+
+def test_group_mixed_hard_soft() -> None:
+    """同一任务的硬依赖和软依赖可同时引用不同组。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=lambda: 1),
+            TaskSpec("b", fn=lambda: 2),
+            TaskSpec("c", fn=lambda: 3),
+            TaskSpec("d", fn=lambda: 4),
+        ]
+    )
+    graph.group("g1", ["a", "b"])
+    graph.group("g2", ["c", "d"])
+    graph.add(
+        TaskSpec(
+            "e",
+            fn=lambda: 5,
+            depends_on=("g1",),
+            soft_depends_on=("g2",),
+        )
+    )
+    assert set(graph.specs["e"].depends_on) == {"a", "b"}
+    assert set(graph.specs["e"].soft_depends_on) == {"c", "d"}

tests/test_loop.py

@@ -0,0 +1,227 @@
+"""LoopSpec 任务循环展开测试。"""
+
+from __future__ import annotations
+
+import pyflowx as px
+from pyflowx import Graph, LoopSpec, TaskSpec, task
+
+
+def test_loop_range_expands_to_instances() -> None:
+    """LoopSpec.range 展开为多实例，item 追加到 args。"""
+
+    def process(item: int) -> int:
+        return item * 2
+
+    spec = TaskSpec("process", fn=process, loop=LoopSpec.range(3))
+    graph = Graph.from_specs([spec])
+
+    assert "process" not in graph.specs
+    assert set(graph.specs.keys()) == {"process_0", "process_1", "process_2"}
+    assert graph.specs["process_0"].args == (0,)
+    assert graph.specs["process_2"].args == (2,)
+    assert graph.specs["process_0"].loop is None
+
+
+def test_loop_from_iterable() -> None:
+    """LoopSpec.from_iterable 接受任意可迭代对象。"""
+
+    def echo(item: str) -> str:
+        return item
+
+    spec = TaskSpec("echo", fn=echo, loop=LoopSpec.from_iterable(["a", "b"]))
+    graph = Graph.from_specs([spec])
+
+    assert set(graph.specs.keys()) == {"echo_0", "echo_1"}
+    assert graph.specs["echo_0"].args == ("a",)
+    assert graph.specs["echo_1"].args == ("b",)
+
+
+def test_loop_range_with_start_step() -> None:
+    """LoopSpec.range 支持 start/stop/step 三参数（与 Python range 一致）。"""
+
+    def process(item: int) -> int:
+        return item
+
+    spec = TaskSpec("p", fn=process, loop=LoopSpec.range(0, 10, 2))
+    graph = Graph.from_specs([spec])
+
+    assert set(graph.specs.keys()) == {"p_0", "p_1", "p_2", "p_3", "p_4"}
+    assert graph.specs["p_0"].args == (0,)
+    assert graph.specs["p_4"].args == (8,)
+
+
+def test_loop_range_two_args() -> None:
+    """LoopSpec.range(start, stop) 两参数形式。"""
+
+    def process(item: int) -> int:
+        return item
+
+    spec = TaskSpec("p", fn=process, loop=LoopSpec.range(5, 8))
+    graph = Graph.from_specs([spec])
+
+    assert set(graph.specs.keys()) == {"p_0", "p_1", "p_2"}
+    assert graph.specs["p_0"].args == (5,)
+    assert graph.specs["p_2"].args == (7,)
+
+
+def test_loop_custom_key_fn() -> None:
+    """key_fn 自定义实例名。"""
+
+    def process(item: int) -> int:
+        return item
+
+    spec = TaskSpec(
+        "process",
+        fn=process,
+        loop=LoopSpec.from_iterable([10, 20], key_fn=lambda _i, item: f"process_{item}"),
+    )
+    graph = Graph.from_specs([spec])
+
+    assert set(graph.specs.keys()) == {"process_10", "process_20"}
+    assert graph.specs["process_10"].args == (10,)
+
+
+def test_loop_dependency_rewrite() -> None:
+    """其他任务引用 loop 原名时，自动重写为依赖所有展开实例。"""
+
+    def gen(item: int) -> int:
+        return item
+
+    def aggregate(gen: dict[str, int]) -> int:
+        return sum(gen.values())
+
+    gen_spec = TaskSpec("gen", fn=gen, loop=LoopSpec.range(3))
+    agg_spec = TaskSpec("agg", fn=aggregate, depends_on=("gen",))
+    graph = Graph.from_specs([gen_spec, agg_spec])
+
+    assert set(graph.specs["agg"].depends_on) == {"gen_0", "gen_1", "gen_2"}
+
+
+def test_loop_soft_dependency_rewrite() -> None:
+    """软依赖引用 loop 原名时也重写。"""
+
+    def gen(item: int) -> int:
+        return item
+
+    def consumer(gen: dict[str, int] | None = None) -> int:
+        return sum(gen.values()) if gen else 0
+
+    gen_spec = TaskSpec("gen", fn=gen, loop=LoopSpec.range(2))
+    cons_spec = TaskSpec("cons", fn=consumer, soft_depends_on=("gen",))
+    graph = Graph.from_specs([gen_spec, cons_spec])
+
+    assert set(graph.specs["cons"].soft_depends_on) == {"gen_0", "gen_1"}
+
+
+def test_loop_chained_dependencies() -> None:
+    """两个 loop 任务链式依赖，下游 loop 引用上游 loop 原名。"""
+
+    def stage1(item: int) -> int:
+        return item + 1
+
+    def stage2(stage1: int) -> int:
+        return stage1 * 2
+
+    s1 = TaskSpec("s1", fn=stage1, loop=LoopSpec.range(2))
+    s2 = TaskSpec("s2", fn=stage2, loop=LoopSpec.range(2), depends_on=("s1",))
+    graph = Graph.from_specs([s1, s2])
+
+    # s2 的每个实例应依赖 s1 的所有实例（因为 s2 引用 "s1" 原名）
+    assert set(graph.specs["s2_0"].depends_on) == {"s1_0", "s1_1"}
+    assert set(graph.specs["s2_1"].depends_on) == {"s1_0", "s1_1"}
+
+
+def test_loop_add_method() -> None:
+    """Graph.add 也支持 loop 展开。"""
+
+    def process(item: int) -> int:
+        return item
+
+    spec = TaskSpec("process", fn=process, loop=LoopSpec.range(2))
+    graph = Graph().add(spec)
+
+    assert set(graph.specs.keys()) == {"process_0", "process_1"}
+
+
+def test_loop_execution() -> None:
+    """loop 展开后能正常执行，结果注入下游。"""
+
+    def gen(item: int) -> int:
+        return item * 10
+
+    def aggregate(gen_0: int, gen_1: int, gen_2: int) -> int:
+        return gen_0 + gen_1 + gen_2
+
+    gen_spec = TaskSpec("gen", fn=gen, loop=LoopSpec.range(3))
+    agg_spec = TaskSpec("agg", fn=aggregate, depends_on=("gen",))
+    graph = Graph.from_specs([gen_spec, agg_spec])
+
+    report = px.run(graph, strategy="sequential")
+    assert report.success
+    assert report["agg"] == 0 + 10 + 20
+
+
+def test_loop_with_task_decorator() -> None:
+    """@task 装饰器支持 loop 参数。"""
+
+    @task
+    def base() -> int:
+        return 0
+
+    @task(loop=LoopSpec.range(2), depends_on=("base",))
+    def worker(base: int, item: int) -> int:
+        return base + item
+
+    graph = Graph.from_specs([base, worker])
+    assert set(graph.specs.keys()) == {"base", "worker_0", "worker_1"}
+    assert set(graph.specs["worker_0"].depends_on) == {"base"}
+
+
+def test_loop_groups_recorded() -> None:
+    """_loop_groups 记录原名到展开名的映射。"""
+
+    def process(item: int) -> int:
+        return item
+
+    spec = TaskSpec("p", fn=process, loop=LoopSpec.range(3))
+    graph = Graph.from_specs([spec])
+
+    assert graph._loop_groups == {"p": ["p_0", "p_1", "p_2"]}
+
+
+def test_loop_empty_items_raises() -> None:
+    """空 items 在 LoopSpec 构造时报错。"""
+    import pytest
+
+    with pytest.raises(ValueError, match="不能为空"):
+        LoopSpec.from_iterable([])
+
+    with pytest.raises(ValueError, match="不能为空"):
+        LoopSpec.range(0)
+
+    with pytest.raises(ValueError, match="不能为空"):
+        LoopSpec.range(5, 5)  # start == stop, 空序列
+
+
+def test_loop_preserves_other_fields() -> None:
+    """loop 展开后保留原 spec 的其他字段（retry/tags 等）。"""
+
+    def process(item: int) -> int:
+        return item
+
+    retry = px.RetryPolicy(max_attempts=3)
+    spec = TaskSpec(
+        "p",
+        fn=process,
+        loop=LoopSpec.range(2),
+        retry=retry,
+        tags=("batch",),
+        priority=5,
+    )
+    graph = Graph.from_specs([spec])
+
+    for name in ("p_0", "p_1"):
+        s = graph.specs[name]
+        assert s.retry == retry
+        assert s.tags == ("batch",)
+        assert s.priority == 5

tests/test_monitoring.py

@@ -0,0 +1,230 @@
+"""监控导出测试：MetricsCollector + health_check + HTTP 服务器。
+
+覆盖：
+* MetricsCollector.on_event 收集任务计数/耗时/重试。
+* metrics_text 输出 Prometheus 文本格式。
+* record_run 记录运行级指标。
+* reset 清空指标。
+* health_check: healthy/degraded/unhealthy/unknown 状态判定。
+* start_metrics_server HTTP 端点返回指标文本。
+"""
+
+from __future__ import annotations
+
+import contextlib
+import socket
+import time
+import urllib.error
+import urllib.request
+
+import pytest
+
+import pyflowx as px
+from pyflowx import Graph, MetricsCollector, TaskSpec, health_check, start_metrics_server
+from pyflowx.task import TaskEvent, TaskStatus
+
+
+class TestMetricsCollector:
+    """MetricsCollector 测试。"""
+
+    def test_collects_task_count(self) -> None:
+        """on_event 正确收集任务计数。"""
+        collector = MetricsCollector()
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.RUNNING))
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=0.5))
+        collector.on_event(TaskEvent(task="b", status=TaskStatus.RUNNING))
+        collector.on_event(TaskEvent(task="b", status=TaskStatus.FAILED, duration=1.0, error="err"))
+        text = collector.metrics_text()
+        assert 'pyflowx_task_total{task="a",status="success"} 1' in text
+        assert 'pyflowx_task_total{task="b",status="failed"} 1' in text
+        assert 'pyflowx_task_total{task="a",status="running"} 1' in text
+
+    def test_collects_duration(self) -> None:
+        """终态事件记录耗时。"""
+        collector = MetricsCollector()
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=1.5))
+        text = collector.metrics_text()
+        assert 'pyflowx_task_duration_seconds{task="a"} 1.5' in text
+        assert 'pyflowx_task_duration_seconds_sum{task="a"} 1.5' in text
+
+    def test_duration_sum_accumulates(self) -> None:
+        """多次执行累加耗时。"""
+        collector = MetricsCollector()
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=1.0))
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=2.0))
+        text = collector.metrics_text()
+        assert 'pyflowx_task_duration_seconds_sum{task="a"} 3.0' in text
+
+    def test_collects_retries(self) -> None:
+        """attempts > 1 时记录重试。"""
+        collector = MetricsCollector()
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=0.1, attempts=3))
+        text = collector.metrics_text()
+        assert 'pyflowx_task_retries_total{task="a"} 2' in text
+
+    def test_no_retries_not_recorded(self) -> None:
+        """attempts <= 1 时不记录重试指标。"""
+        collector = MetricsCollector()
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=0.1, attempts=1))
+        text = collector.metrics_text()
+        assert "pyflowx_task_retries_total" not in text
+
+    def test_record_run(self) -> None:
+        """record_run 记录运行级指标。"""
+        collector = MetricsCollector()
+        graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+        report = px.run(graph, strategy="sequential")
+        collector.record_run(report)
+        text = collector.metrics_text()
+        assert 'pyflowx_run_total{status="success"} 1' in text
+
+    def test_reset(self) -> None:
+        """reset 清空所有指标。"""
+        collector = MetricsCollector()
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=1.0))
+        assert "pyflowx_task_total" in collector.metrics_text()
+        collector.reset()
+        text = collector.metrics_text()
+        assert "pyflowx_task_total" not in text
+
+    def test_with_run_integration(self) -> None:
+        """与 px.run 集成：作为 on_event 回调收集指标。"""
+        collector = MetricsCollector()
+        graph = Graph.from_specs(
+            [
+                TaskSpec("a", fn=lambda: 1),
+                TaskSpec("b", fn=lambda a: a + 1, depends_on=("a",)),
+            ]
+        )
+        px.run(graph, strategy="sequential", on_event=collector.on_event)
+        text = collector.metrics_text()
+        assert 'task="a"' in text
+        assert 'task="b"' in text
+        assert 'status="success"' in text
+
+    def test_metrics_text_format(self) -> None:
+        """metrics_text 输出符合 Prometheus 文本格式。"""
+        collector = MetricsCollector()
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=0.5))
+        text = collector.metrics_text()
+        # HELP 和 TYPE 行存在
+        assert "# HELP pyflowx_task_total" in text
+        assert "# TYPE pyflowx_task_total counter" in text
+        # 以换行结尾
+        assert text.endswith("\n")
+
+
+class TestHealthCheck:
+    """health_check 测试。"""
+
+    def test_healthy(self) -> None:
+        """全部成功 → healthy。"""
+        graph = Graph.from_specs(
+            [
+                TaskSpec("a", fn=lambda: 1),
+                TaskSpec("b", fn=lambda: 2),
+            ]
+        )
+        report = px.run(graph, strategy="sequential")
+        result = health_check(report)
+        assert result["status"] == "healthy"
+        assert result["total"] == 2
+        assert result["success"] == 2
+        assert result["failed"] == 0
+
+    def test_degraded(self) -> None:
+        """部分失败 → degraded。"""
+        graph = Graph.from_specs(
+            [
+                TaskSpec("a", fn=lambda: 1),
+                TaskSpec("b", fn=lambda: (_ for _ in ()).throw(RuntimeError("fail"))),
+            ]
+        )
+        with contextlib.suppress(px.TaskFailedError):
+            px.run(graph, strategy="sequential", on_event=lambda _e: None)
+        # 构造部分失败报告
+        from pyflowx.report import RunReport
+        from pyflowx.task import TaskResult
+
+        report = RunReport(success=False)
+        report.results["a"] = TaskResult(spec=graph.specs["a"], status=TaskStatus.SUCCESS, value=1)
+        report.results["b"] = TaskResult(spec=graph.specs["b"], status=TaskStatus.FAILED, error=RuntimeError("fail"))
+        result = health_check(report)
+        assert result["status"] == "degraded"
+        assert result["failed"] == 1
+        assert result["success"] == 1
+
+    def test_unhealthy(self) -> None:
+        """全部失败 → unhealthy。"""
+        from pyflowx.report import RunReport
+        from pyflowx.task import TaskResult
+
+        report = RunReport(success=False)
+        report.results["a"] = TaskResult(spec=TaskSpec("a", fn=lambda: 1), status=TaskStatus.FAILED)
+        result = health_check(report)
+        assert result["status"] == "unhealthy"
+
+    def test_unknown_empty(self) -> None:
+        """无任务 → unknown。"""
+        from pyflowx.report import RunReport
+
+        report = RunReport()
+        result = health_check(report)
+        assert result["status"] == "unknown"
+
+    def test_includes_duration(self) -> None:
+        """健康检查包含总耗时。"""
+        from datetime import datetime
+
+        from pyflowx.report import RunReport
+        from pyflowx.task import TaskResult
+
+        report = RunReport()
+        spec = TaskSpec("a", fn=lambda: 1)
+        report.results["a"] = TaskResult(
+            spec=spec,
+            status=TaskStatus.SUCCESS,
+            value=1,
+            started_at=datetime(2026, 1, 1, 0, 0, 0),
+            finished_at=datetime(2026, 1, 1, 0, 0, 1),
+        )
+        result = health_check(report)
+        assert result["duration"] == 1.0
+
+
+class TestMetricsServer:
+    """start_metrics_server HTTP 测试。"""
+
+    @staticmethod
+    def _free_port() -> int:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.bind(("127.0.0.1", 0))
+            return s.getsockname()[1]
+
+    def test_serves_metrics(self) -> None:
+        """HTTP 端点返回 Prometheus 格式指标。"""
+        collector = MetricsCollector()
+        collector.on_event(TaskEvent(task="a", status=TaskStatus.SUCCESS, duration=0.1))
+        port = self._free_port()
+        stop = start_metrics_server(collector, port=port, host="127.0.0.1")
+        try:
+            time.sleep(0.1)  # 等待服务器启动
+            with urllib.request.urlopen(f"http://127.0.0.1:{port}/metrics") as resp:
+                body = resp.read().decode("utf-8")
+                assert resp.status == 200
+                assert "pyflowx_task_total" in body
+        finally:
+            stop()
+
+    def test_404_for_unknown_path(self) -> None:
+        """非 /metrics 路径返回 404。"""
+        collector = MetricsCollector()
+        port = self._free_port()
+        stop = start_metrics_server(collector, port=port, host="127.0.0.1")
+        try:
+            time.sleep(0.1)
+            with pytest.raises(urllib.error.HTTPError) as exc_info:
+                urllib.request.urlopen(f"http://127.0.0.1:{port}/unknown")
+            assert exc_info.value.code == 404
+        finally:
+            stop()

tests/test_persistence.py

@@ -0,0 +1,290 @@
+"""持久化与恢复测试：检查点恢复 + 运行历史管理。
+
+覆盖：
+* run(resume_from=RunReport) 跳过已成功任务。
+* run(resume_from=Path) 从 JSON 文件恢复。
+* 恢复后 FAILED/SKIPPED 任务被重新执行。
+* 恢复后下游任务能获取上游恢复结果。
+* RunHistory: save/load/list/latest/delete。
+* RunHistory: __len__ / __contains__。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import pyflowx as px
+from pyflowx import Graph, RunHistory, TaskSpec
+
+
+def test_resume_skips_successful_tasks() -> None:
+    """resume_from 中的 SUCCESS 任务被跳过，不重新执行。"""
+    call_count: dict[str, int] = {"a": 0, "b": 0}
+
+    def make_a():
+        def _a() -> int:
+            call_count["a"] += 1
+            return 1
+
+        return _a
+
+    def make_b():
+        def _b(a: int) -> int:
+            call_count["b"] += 1
+            return a + 1
+
+        return _b
+
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=make_a()),
+            TaskSpec("b", fn=make_b(), depends_on=("a",)),
+        ]
+    )
+    # 第一次运行
+    report1 = px.run(graph, strategy="sequential")
+    assert call_count == {"a": 1, "b": 1}
+
+    # 第二次运行，从 report1 恢复
+    report2 = px.run(graph, strategy="sequential", resume_from=report1)
+    assert call_count == {"a": 1, "b": 1}  # 没有重新执行
+    assert report2["a"] == 1
+    assert report2["b"] == 2
+
+
+def test_resume_from_json_path(tmp_path: Path) -> None:
+    """resume_from 接受 JSON 文件路径。"""
+    graph = Graph.from_specs([TaskSpec("a", fn=lambda: 42)])
+    report1 = px.run(graph, strategy="sequential")
+
+    json_path = tmp_path / "report.json"
+    json_path.write_text(report1.to_json(), encoding="utf-8")
+
+    call_count = 0
+
+    def _a() -> int:
+        nonlocal call_count
+        call_count += 1
+        return 42
+
+    graph2 = Graph.from_specs([TaskSpec("a", fn=_a)])
+    report2 = px.run(graph2, strategy="sequential", resume_from=json_path)
+    assert call_count == 0  # 从检查点恢复，未重新执行
+    assert report2["a"] == 42
+
+
+def test_resume_reruns_failed_tasks() -> None:
+    """resume_from 中 FAILED 的任务被重新执行。"""
+    state: dict[str, bool] = {"should_fail": True}
+
+    def _a() -> str:
+        if state["should_fail"]:
+            raise RuntimeError("第一次失败")
+        return "success"
+
+    graph = Graph.from_specs([TaskSpec("a", fn=_a)])
+    # 第一次运行失败
+    with pytest.raises(px.TaskFailedError):
+        px.run(graph, strategy="sequential")
+
+    # 修复问题
+    state["should_fail"] = False
+
+    # 第二次运行：从头开始（无 resume_from）
+    report2 = px.run(graph, strategy="sequential")
+    assert report2["a"] == "success"
+
+
+def test_resume_downstream_gets_restored_result() -> None:
+    """恢复后下游任务能获取上游恢复结果。
+
+    a 第一次成功，b 第一次失败；第二次 a 被恢复，b 重新执行获取 a 的恢复值。
+    """
+    state: dict[str, bool] = {"b_fail": True}
+    b_call_count = 0
+
+    def _a() -> int:
+        return 10
+
+    def _b(a: int) -> int:
+        nonlocal b_call_count
+        b_call_count += 1
+        if state["b_fail"]:
+            raise RuntimeError("b 第一次失败")
+        return a * 2
+
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=_a),
+            TaskSpec("b", fn=_b, depends_on=("a",)),
+        ]
+    )
+    # 第一次运行：a 成功，b 失败
+    with pytest.raises(px.TaskFailedError):
+        px.run(graph, strategy="sequential")
+
+    # 修复 b
+    state["b_fail"] = False
+    b_call_count = 0  # 重置计数
+
+    # 第二次运行：a 被恢复（值=10），b 重新执行获取 a=10
+    # 需要构造一个只含 a 成功结果的 prev_report
+    prev_report = px.run(
+        Graph.from_specs([TaskSpec("a", fn=_a)]),
+        strategy="sequential",
+    )
+    report2 = px.run(graph, strategy="sequential", resume_from=prev_report)
+    assert report2["a"] == 10  # a 被恢复
+    assert b_call_count == 1  # b 被重新执行一次
+    assert report2["b"] == 20  # b 获取 a=10
+
+
+def test_resume_with_dependency_strategy() -> None:
+    """dependency 策略下 resume_from 也正常工作。"""
+    graph = Graph.from_specs(
+        [
+            TaskSpec("a", fn=lambda: 1),
+            TaskSpec("b", fn=lambda a: a + 1, depends_on=("a",)),
+        ]
+    )
+    report1 = px.run(graph, strategy="dependency")
+    assert report1["a"] == 1
+    assert report1["b"] == 2
+
+    call_count = 0
+
+    def _a() -> int:
+        nonlocal call_count
+        call_count += 1
+        return 1
+
+    graph2 = Graph.from_specs(
+        [
+            TaskSpec("a", fn=_a),
+            TaskSpec("b", fn=lambda a: a + 1, depends_on=("a",)),
+        ]
+    )
+    report2 = px.run(graph2, strategy="dependency", resume_from=report1)
+    assert call_count == 0  # a 被恢复
+    assert report2["a"] == 1
+    assert report2["b"] == 2
+
+
+def test_resume_ignores_tasks_not_in_graph() -> None:
+    """resume_from 中不在当前图里的任务被忽略。"""
+    graph1 = Graph.from_specs([TaskSpec("a", fn=lambda: 1), TaskSpec("b", fn=lambda: 2)])
+    report1 = px.run(graph1, strategy="sequential")
+
+    # 新图只有 a，没有 b
+    graph2 = Graph.from_specs([TaskSpec("a", fn=lambda: 999)])
+    report2 = px.run(graph2, strategy="sequential", resume_from=report1)
+    assert "b" not in report2.results
+    assert report2["a"] == 1  # a 被恢复
+
+
+class TestRunHistory:
+    """RunHistory 测试。"""
+
+    def test_save_and_load(self, tmp_path: Path) -> None:
+        """保存后能正确加载。"""
+        history = RunHistory(tmp_path / "runs")
+        graph = Graph.from_specs([TaskSpec("a", fn=lambda: 42)])
+        report = px.run(graph, strategy="sequential")
+        path = history.save(report)
+        assert path.exists()
+
+        loaded = history.load(report.run_id)
+        assert loaded["a"] == 42
+        assert loaded.run_id == report.run_id
+
+    def test_list_runs(self, tmp_path: Path) -> None:
+        """list_runs 返回全部 run_id。"""
+        history = RunHistory(tmp_path / "runs")
+        for i in range(3):
+            graph = Graph.from_specs([TaskSpec(f"t{i}", fn=lambda i=i: i)])
+            report = px.run(graph, strategy="sequential")
+            history.save(report)
+        runs = history.list_runs()
+        assert len(runs) == 3
+
+    def test_latest(self, tmp_path: Path) -> None:
+        """latest 返回最近保存的报告。"""
+        history = RunHistory(tmp_path / "runs")
+        assert history.latest() is None
+
+        graph1 = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+        r1 = px.run(graph1, strategy="sequential")
+        history.save(r1)
+
+        graph2 = Graph.from_specs([TaskSpec("b", fn=lambda: 2)])
+        r2 = px.run(graph2, strategy="sequential")
+        history.save(r2)
+
+        latest = history.latest()
+        assert latest is not None
+        assert latest.run_id == r2.run_id
+
+    def test_delete(self, tmp_path: Path) -> None:
+        """delete 删除指定报告。"""
+        history = RunHistory(tmp_path / "runs")
+        graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+        report = px.run(graph, strategy="sequential")
+        history.save(report)
+        assert report.run_id in history
+        assert history.delete(report.run_id)
+        assert report.run_id not in history
+        assert not history.delete(report.run_id)  # 再删返回 False
+
+    def test_load_not_found(self, tmp_path: Path) -> None:
+        """加载不存在的 run_id 抛 FileNotFoundError。"""
+        history = RunHistory(tmp_path / "runs")
+        with pytest.raises(FileNotFoundError, match="不存在"):
+            history.load("nonexistent")
+
+    def test_len(self, tmp_path: Path) -> None:
+        """__len__ 返回存储数量。"""
+        history = RunHistory(tmp_path / "runs")
+        assert len(history) == 0
+        graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+        report = px.run(graph, strategy="sequential")
+        history.save(report)
+        assert len(history) == 1
+
+    def test_contains(self, tmp_path: Path) -> None:
+        """__contains__ 检查 run_id 是否存在。"""
+        history = RunHistory(tmp_path / "runs")
+        assert "abc" not in history
+        graph = Graph.from_specs([TaskSpec("a", fn=lambda: 1)])
+        report = px.run(graph, strategy="sequential")
+        history.save(report)
+        assert report.run_id in history
+
+    def test_creates_dir(self, tmp_path: Path) -> None:
+        """目录不存在时自动创建。"""
+        dir = tmp_path / "nested" / "runs"
+        assert not dir.exists()
+        RunHistory(dir)
+        assert dir.exists()
+
+    def test_resume_from_history(self, tmp_path: Path) -> None:
+        """从 RunHistory 加载报告用于 resume_from。"""
+        history = RunHistory(tmp_path / "runs")
+        graph = Graph.from_specs([TaskSpec("a", fn=lambda: 42)])
+        report1 = px.run(graph, strategy="sequential")
+        history.save(report1)
+
+        call_count = 0
+
+        def _a() -> int:
+            nonlocal call_count
+            call_count += 1
+            return 999
+
+        graph2 = Graph.from_specs([TaskSpec("a", fn=_a)])
+        prev = history.latest()
+        assert prev is not None
+        report2 = px.run(graph2, strategy="sequential", resume_from=prev)
+        assert call_count == 0
+        assert report2["a"] == 42

tests/test_report.py

@@ -652,10 +652,12 @@ 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")
 
         # 序列化 → 反序列化
@@ -751,10 +753,12 @@ 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

uv.lock

@@ -1343,7 +1343,7 @@ office = [
 
 [package.dev-dependencies]
 dev = [
-    { name = "pyflowx", extra = ["dev", "docs", "office"] },
+    { name = "pyflowx", extra = ["dev", "docs", "fast", "office"] },
 ]
 
 [package.metadata]
@@ -1377,7 +1377,7 @@ requires-dist = [
 provides-extras = ["dev", "docs", "fast", "office"]
 
 [package.metadata.requires-dev]
-dev = [{ name = "pyflowx", extras = ["dev", "docs", "office"], editable = "." }]
+dev = [{ name = "pyflowx", extras = ["dev", "docs", "fast", "office"], editable = "." }]
 
 [[package]]
 name = "pygments"