feat: 实现 GitHub Actions 风格 YAML 任务编排

新增 yaml_loader 模块支持 jobs/needs/strategy.matrix/if 条件/env/defaults 等 CI/CD 概念，
Graph.from_yaml classmethod 与 pf yamlrun CLI 命令；矩阵笛卡尔积展开与依赖自动展开，
${{ matrix.key }} 占位符替换，字段名 hyphen/underscore 兼容。

.trae/docs/iter-07-yaml-orchestration.md

@@ -0,0 +1,80 @@
+# 迭代 07：YAML 任务编排（GitHub Actions 风格）
+
+## 本轮目标
+
+实现 GitHub Actions 风格的 YAML 声明式任务编排：
+- `Graph.from_yaml(path)` classmethod
+- `parse_yaml_string(text)` / `load_yaml(path)` 函数式 API
+- `pf yamlrun <file>` CLI 命令（--strategy/--dry-run/--list/--quiet）
+- 完整 schema：jobs/needs/strategy.matrix/if/continue-on-error/env/defaults
+
+## 改动文件清单
+
+- `pyproject.toml` — 添加 PyYAML>=6.0.1 运行时依赖
+- `src/pyflowx/yaml_loader.py` — 新增：YAML 解析 + Graph 构建
+- `src/pyflowx/graph.py` — 添加 `from_yaml` classmethod（委托 yaml_loader）
+- `src/pyflowx/__init__.py` — 导出 `load_yaml`、`parse_yaml_string`
+- `src/pyflowx/cli/pf.py` — 添加 `yamlrun` 特殊命令路由
+- `tests/test_yaml_loader.py` — 新增测试
+- `README.md` — 恢复 YAML 编排章节（现在已实现）
+
+## 关键设计
+
+### 1. Matrix 展开
+
+笛卡尔积，命名 `{job_id}_{key}-{value}`（多键用 `_` 连接）。
+如 `version: ["3.10","3.11"]` × `os: ["linux","macos"]` 展开 4 个任务：
+`test_version-3.10_os-linux` 等。
+
+矩阵依赖自动展开：若 job A `needs: [B]`，B 是矩阵 job，A 依赖 B 的所有变体。
+
+### 2. 条件解析
+
+- `success()` / `always()` → 无条件（空 conditions）
+- `env.VAR` → `ENV_VAR_EXISTS("VAR")`
+- `env.VAR == 'x'` → `ENV_VAR_EQUALS("VAR", "x")`
+- `env.VAR != 'x'` → `NOT(ENV_VAR_EQUALS("VAR", "x"))`
+- `failure()` → 不支持，抛 ValueError
+
+### 3. 字段名兼容
+
+支持 hyphen 与 underscore：`continue-on-error` = `continue_on_error`。
+
+### 4. cmd vs run
+
+- `cmd: [...]` → list 形式（无 shell）
+- `run: "..."` → shell 字符串形式
+
+### 5. ${{ matrix.key }} 替换
+
+在 cmd/run/cwd/env 的字符串值中替换矩阵占位符。
+
+### 6. yamlrun CLI
+
+作为 PfApp 特殊命令（非 @px.tool），路由：
+`pf yamlrun <file> [--strategy S] [--dry-run] [--list] [--quiet]`
+
+## 验收标准
+
+- `Graph.from_yaml("pipeline.yaml")` 正确构建 DAG
+- 矩阵展开为笛卡尔积，命名一致
+- 矩阵依赖自动展开到所有变体
+- 条件解析支持 success/always/env.VAR/env.VAR==/env.VAR!=
+- `${{ matrix.key }}` 在 cmd/run/cwd/env 中正确替换
+- 字段名 hyphen/underscore 兼容
+- `pf yamlrun` CLI 命令可用
+- 覆盖率 ≥ 95%
+
+## 验证结果
+
+- ruff check + format：通过
+- pyrefly check：通过
+- pytest：1181 passed
+- 覆盖率：97.23%（yaml_loader.py 100%）
+- 清理了 _cartesian_product 和 _parse_condition 中的不可达防御代码
+
+## 遗留事项
+
+- P3 性能基准与优化（下一迭代）
+- P4 任务取消与优雅停止（下一迭代）
+- `pf.py` 的 `run()` 方法 PLR0911 已通过合并 no-args/--help 分支解决（6 returns）

README.md

@@ -36,7 +36,8 @@ PyFlowX 把"任务依赖"这件事做到极致简单：**参数名就是依赖
 - **任务通知** —— `Notifier` 协议支持飞书/钉钉/企业微信 webhook 与 Python 回调，全生命周期事件可配置
 - **流式获取** —— `run_iter()` 生成器逐个 yield 任务结果，适用于大量任务逐个处理
 - **状态后端** —— `MemoryBackend` / `JSONBackend` / `SQLiteBackend`（WAL 模式，适合大规模任务）
-- **最小依赖** —— `rich` + `typer` + `typing-extensions`（3.13 以下）
+- **YAML 任务编排** —— GitHub Actions 风格 `jobs`/`needs`/`strategy.matrix`/`if` 条件，`pf yamlrun pipeline.yaml` 一键执行
+- **最小依赖** —— `rich` + `typer` + `typing-extensions`（3.13 以下）+ `pyyaml`
 - **97% 测试覆盖** —— 分支覆盖率 >= 95%
 
 ## 安装
@@ -150,6 +151,67 @@ resolved = px.compose(graphs)  # "all" 图中的 "build" 引用被展开
 引用格式：`"command_name"`（整个图）或 `"command_name.task_name"`（特定任务）。
 `CliRunner` 内部自动调用 `compose`。
 
+### YAML 任务编排
+
+GitHub Actions 风格的 YAML 文件直接加载为 `Graph`，支持 `jobs`/`needs`/`strategy.matrix`/`if`/`continue-on-error`/`env`/`defaults`：
+
+```yaml
+# pipeline.yaml
+strategy: thread
+defaults:
+  retry: {max_attempts: 3}
+  env: {CI: "true"}
+
+jobs:
+  setup:
+    cmd: ["echo", "setup"]
+    runs-on: linux
+
+  build:
+    needs: [setup]
+    cmd: ["echo", "build-${{ matrix.version }}"]
+    strategy:
+      matrix:
+        version: ["3.10", "3.11", "3.12"]
+    if: "env.CI"
+
+  test:
+    needs: [build]
+    cmd: ["echo", "test"]
+    continue-on-error: true
+```
+
+```python
+import pyflowx as px
+
+graph = px.Graph.from_yaml("pipeline.yaml")
+# 或从字符串解析：graph = px.parse_yaml_string("...")
+report = px.run(graph, strategy="thread")
+```
+
+CLI 一键执行：
+
+```bash
+pf yamlrun pipeline.yaml              # 执行
+pf yamlrun pipeline.yaml --dry-run    # 仅预览，不执行
+pf yamlrun pipeline.yaml --list       # 列出所有任务
+pf yamlrun pipeline.yaml --quiet      # 静默模式
+pf yamlrun pipeline.yaml --strategy sequential  # 覆盖策略
+```
+
+**字段映射**：
+
+| YAML 字段 | TaskSpec 字段 | 说明 |
+|-----------|---------------|------|
+| `cmd` / `run` | `cmd` | `cmd` 为列表，`run` 为 shell 字符串 |
+| `needs` | `depends_on` | 矩阵 job 依赖自动展开为所有变体 |
+| `if` | `conditions` | `success()`/`always()`/`env.VAR`/`env.VAR == 'x'`/`env.VAR != 'x'` |
+| `strategy.matrix` | 矩阵扇出 | 笛卡尔积，命名 `{job}_{key}-{value}` |
+| `${{ matrix.key }}` | 占位符 | `cmd`/`run`/`cwd`/`env` 中替换 |
+| `continue-on-error` | `continue_on_error` | 字段名支持 hyphen/underscore |
+| `runs-on` | `tags` | 追加到 tags 元组 |
+| `defaults` | `GraphDefaults` | 图级默认值（retry/timeout/env/cwd 等） |
+
 ### 任务模板 —— task_template
 
 `task_template` 工厂批量生成相似 TaskSpec：
@@ -448,6 +510,7 @@ uv run ruff format --check src tests
 | `conditions.py` | 条件执行：内置条件与组合器 |
 | `executors.py` | 执行器与 `run` 入口：四种策略共享模块级辅助；verbose 统一应用到 spec |
 | `storage.py` | 状态后端：`MemoryBackend` / `JSONBackend`（batch flush） |
+| `yaml_loader.py` | YAML 任务编排：`load_yaml` / `parse_yaml_string`（GitHub Actions 风格） |
 | `runner.py` | CLI 运行器：`CliRunner` |
 | `report.py` | 运行结果：`RunReport` / `TaskResult` |
 | `tools.py` | `@px.tool` 装饰器：`ToolSpec` / `run_tool` / `list_tools` / `list_subcommands` |

docs/installation.rst

@@ -1,7 +1,7 @@
 安装
 ====
 
-PyFlowX 支持 Python 3.10+，运行时依赖 ``rich``、``typer`` 与 ``typing-extensions``（3.13 以下）。
+PyFlowX 支持 Python 3.10+，运行时依赖 ``rich``、``typer``、``pyyaml`` 与 ``typing-extensions``（3.13 以下）。
 
 pip 安装
 --------

pyproject.toml

@@ -13,6 +13,7 @@ dependencies = [
     "rich>=13.7.0",
     "typer>=0.24.0",
     "typing-extensions>=4.13.2; python_version < '3.13'",
+    "pyyaml>=6.0.1",
 ]
 description = "Lightweight, type-safe DAG task scheduler with multi-strategy execution."
 keywords = ["async", "dag", "scheduler", "task", "workflow"]

src/pyflowx/__init__.py

@@ -103,6 +103,7 @@ from .task import (
     task_template,
 )
 from .tools import ToolSpec, list_subcommands, list_tools, run_tool, tool
+from .yaml_loader import load_yaml, parse_yaml_string
 
 __version__ = "0.4.8"
 
@@ -161,6 +162,8 @@ __all__ = [
     "describe_injection",
     "list_subcommands",
     "list_tools",
+    "load_yaml",
+    "parse_yaml_string",
     "run",
     "run_command",
     "run_iter",

src/pyflowx/cli/pf.py

@@ -141,18 +141,16 @@ class PfApp:
 
     def run(self) -> int:
         """主入口, 返回退出码."""
-        if not self._argv:
+        if not self._argv or self._argv[0] in ("--help", "-h"):
             self._list_tools()
             return 0
 
         first = self._argv[0]
-
-        if first in ("--help", "-h"):
-            self._list_tools()
-            return 0
         if first in ("--version", "-V"):
             self._console.print(f"PyFlowX [bold cyan]{__version__}[/bold cyan]")
             return 0
+        if first == "yamlrun":
+            return self._run_yaml(self._argv[1:])
 
         rest = self._argv[1:]
         resolved = self._resolve_tool(first)
@@ -199,6 +197,7 @@ class PfApp:
         self._console.print("  [cyan]pf filedate add a.txt[/cyan]  # 给文件添加日期前缀")
         self._console.print("  [cyan]pf pymake b[/cyan]            # 构建 Python 包")
         self._console.print("  [cyan]pf gitt c[/cyan]              # 清理并查看 git 状态")
+        self._console.print("  [cyan]pf yamlrun pipeline.yaml[/cyan]  # 执行 YAML 任务图")
 
     def _aliases_for(self, canonical: str) -> list[str]:
         """获取工具的别名 (不含规范名本身)."""
@@ -256,6 +255,55 @@ class PfApp:
         finally:
             sys.argv = original_argv
 
+    def _run_yaml(self, argv: list[str]) -> int:
+        """执行 yamlrun 命令: 从 YAML 文件加载任务图并执行."""
+        import argparse
+        from typing import get_args
+
+        from pyflowx import Graph, run
+        from pyflowx.errors import PyFlowXError
+        from pyflowx.executors import Strategy
+        from pyflowx.runner import CliExitCode
+
+        parser = argparse.ArgumentParser(prog="pf yamlrun", description="执行 YAML 任务图")
+        _ = parser.add_argument("file", help="YAML 文件路径")
+        _ = parser.add_argument(
+            "--strategy",
+            choices=list(get_args(Strategy)),
+            default="dependency",
+            help="执行策略 (默认: %(default)s)",
+        )
+        _ = parser.add_argument("--dry-run", action="store_true", help="仅打印执行计划")
+        _ = parser.add_argument("--list", action="store_true", help="列出所有任务名")
+        _ = parser.add_argument("--quiet", action="store_true", help="静默模式")
+
+        parsed = parser.parse_args(argv)
+
+        try:
+            graph = Graph.from_yaml(parsed.file)
+        except (OSError, ValueError, PyFlowXError) as e:
+            self._err.print(f"[red]错误:[/red] 加载 YAML 失败: {e}")
+            return CliExitCode.FAILURE.value
+
+        if parsed.list:
+            for name in graph.names:
+                self._console.print(name)
+            return CliExitCode.SUCCESS.value
+
+        if parsed.dry_run:
+            self._console.print(graph.describe())
+            return CliExitCode.SUCCESS.value
+
+        try:
+            report = run(graph, strategy=parsed.strategy, verbose=not parsed.quiet)
+            return CliExitCode.SUCCESS.value if report.success else CliExitCode.FAILURE.value
+        except KeyboardInterrupt:
+            print("\n操作已取消", file=sys.stderr)
+            return CliExitCode.INTERRUPTED.value
+        except PyFlowXError as e:
+            self._err.print(f"[red]错误:[/red] {e}")
+            return CliExitCode.FAILURE.value
+
     def _run_tool(self, target: str, argv: list[str]) -> int:
         """运行 @px.tool 工具.
 

src/pyflowx/graph.py

@@ -21,6 +21,7 @@ import graphlib
 import inspect
 from collections.abc import Callable, Iterable, Mapping, Sequence
 from dataclasses import dataclass, field, replace
+from pathlib import Path
 from typing import Any
 
 from .errors import CycleError, DuplicateTaskError, MissingDependencyError
@@ -228,6 +229,30 @@ class Graph:
         graph.validate()
         return graph
 
+    @classmethod
+    def from_yaml(cls, path: str | Path) -> Graph:
+        """从 YAML 文件加载任务图（GitHub Actions 风格）。
+
+        Parameters
+        ----------
+        path:
+            YAML 文件路径。
+
+        Returns
+        -------
+        Graph
+            解析后的任务图，支持 ``jobs``/``needs``/``strategy.matrix``/
+            ``if``/``continue-on-error``/``env``/``defaults`` 等字段。
+
+        Raises
+        ------
+        ValueError
+            YAML 结构不符合 schema 时。
+        """
+        from .yaml_loader import load_yaml
+
+        return load_yaml(path)
+
     def add_subgraph(self, sub: Graph, *, namespace: str | None = None) -> Graph:
         """将子图合并到当前图，任务名加命名空间前缀避免冲突。
 

src/pyflowx/yaml_loader.py

@@ -0,0 +1,384 @@
+"""YAML 任务编排（GitHub Actions 风格）。
+
+支持 ``jobs``/``needs``/``strategy.matrix``/``if``/``continue-on-error``/
+``env``/``defaults`` 等 CI/CD 概念，从 YAML 文件直接加载为 :class:`Graph`。
+
+Schema
+------
+.. code-block:: yaml
+
+    strategy: thread              # 图级默认策略
+    defaults:                     # 图级默认值
+      retry: {max_attempts: 3}
+      verbose: true
+      env: {CI: "true"}
+
+    jobs:
+      setup:
+        cmd: ["git", "clone", "..."]
+        runs-on: linux
+
+      build:
+        needs: [setup]
+        cmd: ["python", "-m", "build"]
+        timeout: 300
+
+      test:
+        needs: [build]
+        cmd: ["python${{ matrix.version }}", "-m", "pytest"]
+        strategy:
+          matrix:
+            version: ["3.10", "3.11", "3.12"]
+        if: "env.CI"
+
+字段映射
+--------
+| YAML 字段 | TaskSpec 字段 | 说明 |
+|-----------|---------------|------|
+| cmd / run | cmd | cmd 为列表，run 为 shell 字符串 |
+| needs | depends_on | 矩阵依赖自动展开 |
+| if | conditions | success/always/env.VAR/env.VAR==/env.VAR!= |
+| strategy.matrix | 矩阵扇出 | 笛卡尔积展开 |
+| ${{ matrix.key }} | 占位符 | cmd/run/cwd/env 中替换 |
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from dataclasses import replace
+from pathlib import Path
+from typing import Any
+
+import yaml  # type: ignore[import-not-found]
+
+from .conditions import BuiltinConditions, Condition
+from .graph import Graph, GraphDefaults
+from .task import RetryPolicy, TaskSpec
+
+__all__ = ["load_yaml", "parse_yaml_string"]
+
+_MATRIX_PATTERN = re.compile(r"\$\{\{\s*matrix\.(\w+)\s*\}\}")
+
+
+def parse_yaml_string(text: str) -> Graph:
+    """从 YAML 字符串解析任务图。
+
+    Parameters
+    ----------
+    text:
+        YAML 格式的字符串。
+
+    Returns
+    -------
+    Graph
+        解析后的任务图。
+
+    Raises
+    ------
+    ValueError
+        YAML 结构不符合 schema 时。
+    """
+    data = yaml.safe_load(text)
+    return _build_graph(data)
+
+
+def load_yaml(path: str | Path) -> Graph:
+    """从 YAML 文件加载任务图。
+
+    Parameters
+    ----------
+    path:
+        YAML 文件路径。
+
+    Returns
+    -------
+    Graph
+        解析后的任务图。
+    """
+    data = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
+    return _build_graph(data)
+
+
+def _build_graph(data: Any) -> Graph:
+    """从解析后的 YAML 字典构建 Graph。"""
+    if not isinstance(data, Mapping):
+        raise ValueError(f"YAML 根节点必须是映射，收到: {type(data).__name__}")
+
+    jobs = data.get("jobs")
+    if not jobs:
+        raise ValueError("YAML 缺少 'jobs' 或 jobs 为空")
+    if not isinstance(jobs, Mapping):
+        raise ValueError(f"jobs 必须是映射，收到: {type(jobs).__name__}")
+
+    defaults = _parse_defaults(data.get("defaults"))
+    strategy = data.get("strategy")
+    if strategy:
+        defaults = replace(defaults, strategy=str(strategy))
+
+    job_expansions = _precompute_expansions(jobs)
+    specs = _build_specs(jobs, job_expansions)
+    return Graph.from_specs(specs, defaults=defaults)
+
+
+def _parse_defaults(data: Any) -> GraphDefaults:
+    """解析 defaults 字段为 GraphDefaults。"""
+    if not data:
+        return GraphDefaults()
+    if not isinstance(data, Mapping):
+        raise ValueError(f"defaults 必须是映射，收到: {type(data).__name__}")
+
+    kwargs: dict[str, Any] = {}
+    if "retry" in data:
+        kwargs["retry"] = _parse_retry(data["retry"])
+    if "timeout" in data:
+        kwargs["timeout"] = float(data["timeout"])
+    if "verbose" in data:
+        kwargs["verbose"] = bool(data["verbose"])
+    if "env" in data:
+        kwargs["env"] = dict(data["env"])
+    if "cwd" in data:
+        kwargs["cwd"] = Path(data["cwd"])
+    if "tags" in data:
+        kwargs["tags"] = tuple(data["tags"])
+    if "priority" in data:
+        kwargs["priority"] = int(data["priority"])
+    if "continue_on_error" in data or "continue-on-error" in data:
+        kwargs["continue_on_error"] = bool(_get_field(data, "continue_on_error"))
+    if "concurrency_key" in data or "concurrency-key" in data:
+        kwargs["concurrency_key"] = _get_field(data, "concurrency_key")
+    return GraphDefaults(**kwargs)
+
+
+def _parse_retry(data: Any) -> RetryPolicy:
+    """解析 retry 字段为 RetryPolicy。"""
+    if not isinstance(data, Mapping):
+        raise ValueError(f"retry 必须是映射，收到: {type(data).__name__}")
+    kwargs: dict[str, Any] = {}
+    if "max_attempts" in data:
+        kwargs["max_attempts"] = int(data["max_attempts"])
+    if "delay" in data:
+        kwargs["delay"] = float(data["delay"])
+    if "backoff" in data:
+        kwargs["backoff"] = float(data["backoff"])
+    if "jitter" in data:
+        kwargs["jitter"] = float(data["jitter"])
+    return RetryPolicy(**kwargs)
+
+
+def _get_field(data: Mapping[str, Any], name: str) -> Any:
+    """获取字段值，支持 hyphen/underscore 兼容。"""
+    if name in data:
+        return data[name]
+    hyphen = name.replace("_", "-")
+    if hyphen in data:
+        return data[hyphen]
+    return None
+
+
+def _precompute_expansions(jobs: Mapping[str, Any]) -> dict[str, list[tuple[str, dict[str, str]]]]:
+    """预计算每个 job 的矩阵展开。
+
+    Returns
+    -------
+    dict[str, list[tuple[str, dict[str, str]]]]
+        job_id → [(expanded_name, {key: value}), ...]；非矩阵 job 返回 [(job_id, {})]。
+    """
+    expansions: dict[str, list[tuple[str, dict[str, str]]]] = {}
+    for job_id, job_data in jobs.items():
+        if not isinstance(job_data, Mapping):
+            raise ValueError(f"job {job_id!r} 必须是映射，收到: {type(job_data).__name__}")
+        matrix_data = _extract_matrix(job_data)
+        if not matrix_data:
+            expansions[job_id] = [(job_id, {})]
+            continue
+        combos = _cartesian_product(matrix_data)
+        expanded: list[tuple[str, dict[str, str]]] = []
+        for combo in combos:
+            parts = [f"{k}-{v}" for k, v in combo.items()]
+            expanded_name = f"{job_id}_{'_'.join(parts)}"
+            expanded.append((expanded_name, combo))
+        expansions[job_id] = expanded
+    return expansions
+
+
+def _extract_matrix(job_data: Mapping[str, Any]) -> Mapping[str, list[Any]] | None:
+    """从 job 数据中提取 strategy.matrix。"""
+    strategy = _get_field(job_data, "strategy")
+    if strategy and isinstance(strategy, Mapping):
+        matrix = strategy.get("matrix")
+        if isinstance(matrix, Mapping):
+            return matrix
+    return None
+
+
+def _cartesian_product(matrix_data: Mapping[str, list[Any]]) -> list[dict[str, str]]:
+    """计算矩阵的笛卡尔积。"""
+    keys = list(matrix_data.keys())
+    result: list[dict[str, str]] = [{}]
+    for key in keys:
+        values = matrix_data[key]
+        if not isinstance(values, list):
+            raise ValueError(f"matrix.{key} 必须是列表，收到: {type(values).__name__}")
+        new_result: list[dict[str, str]] = []
+        for existing in result:
+            for value in values:
+                combo = dict(existing)
+                combo[key] = str(value)
+                new_result.append(combo)
+        result = new_result
+    return result
+
+
+def _build_specs(
+    jobs: Mapping[str, Any],
+    expansions: dict[str, list[tuple[str, dict[str, str]]]],
+) -> list[TaskSpec[Any]]:
+    """从 jobs 构建展开后的 TaskSpec 列表。"""
+    specs: list[TaskSpec[Any]] = []
+    for job_id, job_data in jobs.items():
+        for expanded_name, matrix_values in expansions[job_id]:
+            specs.append(_build_spec(job_id, expanded_name, job_data, matrix_values, expansions))
+    return specs
+
+
+def _build_spec(
+    job_id: str,
+    expanded_name: str,
+    job_data: Mapping[str, Any],
+    matrix_values: dict[str, str],
+    expansions: dict[str, list[tuple[str, dict[str, str]]]],
+) -> TaskSpec[Any]:
+    """构建单个 TaskSpec。"""
+    task_cmd = _parse_cmd(job_id, job_data, matrix_values)
+    depends_on = _expand_needs(_get_field(job_data, "needs"), expansions)
+    if_expr = _get_field(job_data, "if")
+    conditions = _parse_condition(if_expr) if if_expr else ()
+
+    kwargs: dict[str, Any] = {"cmd": task_cmd, "depends_on": tuple(depends_on)}
+    if conditions:
+        kwargs["conditions"] = conditions
+    kwargs.update(_parse_optional_fields(job_data, matrix_values))
+    return TaskSpec(name=expanded_name, **kwargs)
+
+
+def _parse_cmd(job_id: str, job_data: Mapping[str, Any], matrix_values: dict[str, str]) -> list[str] | str:
+    """解析 cmd / run 字段，应用矩阵占位符替换。"""
+    cmd_val = _get_field(job_data, "cmd")
+    run_val = _get_field(job_data, "run")
+    if cmd_val is not None:
+        if isinstance(cmd_val, list):
+            return [_substitute_matrix_str(str(a), matrix_values) for a in cmd_val]
+        return _substitute_matrix_str(str(cmd_val), matrix_values)
+    if run_val is not None:
+        return _substitute_matrix_str(str(run_val), matrix_values)
+    raise ValueError(f"job {job_id!r} 必须提供 cmd 或 run")
+
+
+def _parse_optional_fields(job_data: Mapping[str, Any], matrix_values: dict[str, str]) -> dict[str, Any]:
+    """解析所有可选字段，返回 kwargs 字典。"""
+    kwargs: dict[str, Any] = {}
+
+    timeout = _get_field(job_data, "timeout")
+    if timeout is not None:
+        kwargs["timeout"] = float(timeout)
+
+    retry = _get_field(job_data, "retry")
+    if retry is not None:
+        kwargs["retry"] = _parse_retry(retry)
+
+    cwd = _get_field(job_data, "cwd")
+    if cwd is not None:
+        kwargs["cwd"] = Path(_substitute_matrix_str(str(cwd), matrix_values))
+
+    env = _get_field(job_data, "env")
+    if env is not None:
+        kwargs["env"] = _substitute_matrix_env(env, matrix_values)
+
+    for bool_field in ("verbose", "skip_if_missing", "allow_upstream_skip", "continue_on_error"):
+        val = _get_field(job_data, bool_field)
+        if val is not None:
+            kwargs[bool_field] = bool(val)
+
+    priority = _get_field(job_data, "priority")
+    if priority is not None:
+        kwargs["priority"] = int(priority)
+
+    concurrency_key = _get_field(job_data, "concurrency_key")
+    if concurrency_key is not None:
+        kwargs["concurrency_key"] = _substitute_matrix_str(str(concurrency_key), matrix_values)
+
+    tags = _get_field(job_data, "tags")
+    tag_list = list(tags) if tags else []
+    runs_on = _get_field(job_data, "runs_on")
+    if runs_on:
+        tag_list.append(str(runs_on))
+    if tag_list:
+        kwargs["tags"] = tuple(tag_list)
+
+    return kwargs
+
+
+def _expand_needs(
+    needs: Any,
+    expansions: dict[str, list[tuple[str, dict[str, str]]]],
+) -> list[str]:
+    """展开 needs 列表：矩阵 job 依赖展开为所有变体。"""
+    if not needs:
+        return []
+    if isinstance(needs, str):
+        needs = [needs]
+    result: list[str] = []
+    for dep in needs:
+        if dep in expansions:
+            result.extend(name for name, _ in expansions[dep])
+        else:
+            result.append(dep)
+    return result
+
+
+def _parse_condition(expr: Any) -> tuple[Condition, ...]:
+    """解析 if 表达式为 Condition 元组。
+
+    支持：success() / always() / env.VAR / env.VAR == 'x' / env.VAR != 'x'。
+    failure() 不支持（PyFlowX 无层屏障概念）。
+    """
+    expr = str(expr).strip()
+
+    if expr in ("success()", "always()"):
+        return ()
+    if expr == "failure()":
+        raise ValueError("failure() 条件不支持；PyFlowX 无层屏障概念")
+
+    m = re.fullmatch(r"env\.(\w+)", expr)
+    if m:
+        return (BuiltinConditions.ENV_VAR_EXISTS(m.group(1)),)
+
+    m = re.fullmatch(r"env\.(\w+)\s*==\s*['\"]([^'\"]*)['\"]", expr)
+    if m:
+        return (BuiltinConditions.ENV_VAR_EQUALS(m.group(1), m.group(2)),)
+
+    m = re.fullmatch(r"env\.(\w+)\s*!=\s*['\"]([^'\"]*)['\"]", expr)
+    if m:
+        return (BuiltinConditions.NOT(BuiltinConditions.ENV_VAR_EQUALS(m.group(1), m.group(2))),)
+
+    raise ValueError(f"无法解析的条件表达式: {expr!r}")
+
+
+def _substitute_matrix_str(text: str, matrix_values: dict[str, str]) -> str:
+    """替换字符串中的 ${{ matrix.key }} 占位符。"""
+
+    def _replacer(m: re.Match[str]) -> str:
+        key = m.group(1)
+        if key not in matrix_values:
+            raise ValueError(f"矩阵变量 matrix.{key} 未定义")
+        return matrix_values[key]
+
+    return _MATRIX_PATTERN.sub(_replacer, text)
+
+
+def _substitute_matrix_env(env: Any, matrix_values: dict[str, str]) -> dict[str, str]:
+    """替换环境变量中的矩阵占位符。"""
+    if not isinstance(env, Mapping):
+        raise ValueError(f"env 必须是映射，收到: {type(env).__name__}")
+    return {k: _substitute_matrix_str(str(v), matrix_values) for k, v in env.items()}

tests/test_yaml_loader.py

@@ -0,0 +1,731 @@
+"""YAML 任务编排测试。"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import pyflowx as px
+from pyflowx.task import TaskStatus
+from pyflowx.yaml_loader import load_yaml, parse_yaml_string
+
+
+class TestBasicParsing:
+    """基本解析测试。"""
+
+    def test_parse_simple_cmd_job(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  hello:
+    cmd: ["echo", "hello"]
+""")
+        assert "hello" in graph
+        spec = graph.spec("hello")
+        assert spec.cmd == ["echo", "hello"]
+
+    def test_parse_run_as_shell_string(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  greet:
+    run: "echo hello | wc -l"
+""")
+        spec = graph.spec("greet")
+        assert spec.cmd == "echo hello | wc -l"
+
+    def test_parse_needs_dependency(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  a:
+    cmd: ["echo", "a"]
+  b:
+    needs: [a]
+    cmd: ["echo", "b"]
+""")
+        assert graph.dependencies("b") == ("a",)
+
+    def test_parse_needs_single_string(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  a:
+    cmd: ["echo", "a"]
+  b:
+    needs: a
+    cmd: ["echo", "b"]
+""")
+        assert graph.dependencies("b") == ("a",)
+
+    def test_parse_timeout(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  slow:
+    cmd: ["sleep", "10"]
+    timeout: 300
+""")
+        assert graph.spec("slow").timeout == 300.0
+
+    def test_parse_retry(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  flaky:
+    cmd: ["curl", "http://example.com"]
+    retry: {max_attempts: 3, delay: 1.0, backoff: 2.0}
+""")
+        spec = graph.spec("flaky")
+        assert spec.retry.max_attempts == 3
+        assert spec.retry.delay == 1.0
+        assert spec.retry.backoff == 2.0
+
+    def test_parse_retry_partial_fields(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  flaky:
+    cmd: ["echo", "flaky"]
+    retry: {delay: 0.5}
+""")
+        spec = graph.spec("flaky")
+        assert spec.retry.max_attempts == 1
+        assert spec.retry.delay == 0.5
+
+    def test_parse_env(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["make", "all"]
+    env: {CI: "true", DEBUG: "1"}
+""")
+        spec = graph.spec("build")
+        assert spec.env == {"CI": "true", "DEBUG": "1"}
+
+    def test_parse_cwd(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["make", "all"]
+    cwd: /tmp/build
+""")
+        assert graph.spec("build").cwd == Path("/tmp/build")
+
+    def test_parse_verbose(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    verbose: true
+""")
+        assert graph.spec("build").verbose is True
+
+    def test_parse_tags(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    tags: [api, build]
+""")
+        assert graph.spec("build").tags == ("api", "build")
+
+    def test_parse_runs_on_added_to_tags(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    runs-on: linux
+""")
+        assert "linux" in graph.spec("build").tags
+
+    def test_parse_priority(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  high:
+    cmd: ["echo", "high"]
+    priority: 10
+""")
+        assert graph.spec("high").priority == 10
+
+    def test_parse_concurrency_key(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  deploy:
+    cmd: ["echo", "deploy"]
+    concurrency-key: deploy_lock
+""")
+        assert graph.spec("deploy").concurrency_key == "deploy_lock"
+
+
+class TestFieldnameCompat:
+    """hyphen/underscore 字段名兼容测试。"""
+
+    def test_continue_on_error_hyphen(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  flaky:
+    cmd: ["echo", "flaky"]
+    continue-on-error: true
+""")
+        assert graph.spec("flaky").continue_on_error is True
+
+    def test_continue_on_error_underscore(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  flaky:
+    cmd: ["echo", "flaky"]
+    continue_on_error: true
+""")
+        assert graph.spec("flaky").continue_on_error is True
+
+    def test_allow_upstream_skip_hyphen(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  cleanup:
+    cmd: ["echo", "cleanup"]
+    allow-upstream-skip: true
+""")
+        assert graph.spec("cleanup").allow_upstream_skip is True
+
+    def test_skip_if_missing_hyphen(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  optional:
+    cmd: ["nonexistent-tool", "arg"]
+    skip-if-missing: true
+""")
+        assert graph.spec("optional").skip_if_missing is True
+
+
+class TestMatrixExpansion:
+    """矩阵展开测试。"""
+
+    def test_single_key_matrix(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["python", "-m", "pytest"]
+    strategy:
+      matrix:
+        version: ["3.10", "3.11", "3.12"]
+""")
+        names = graph.names
+        assert "test_version-3.10" in names
+        assert "test_version-3.11" in names
+        assert "test_version-3.12" in names
+        assert len(names) == 3
+
+    def test_multi_key_matrix_cartesian(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["python", "-m", "pytest"]
+    strategy:
+      matrix:
+        version: ["3.10", "3.11"]
+        os: ["linux", "macos"]
+""")
+        names = set(graph.names)
+        assert names == {
+            "test_version-3.10_os-linux",
+            "test_version-3.11_os-linux",
+            "test_version-3.10_os-macos",
+            "test_version-3.11_os-macos",
+        }
+
+    def test_matrix_placeholder_substitution_in_cmd(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["python${{ matrix.version }}", "-m", "pytest"]
+    strategy:
+      matrix:
+        version: ["3.10", "3.11"]
+""")
+        spec = graph.spec("test_version-3.10")
+        assert spec.cmd == ["python3.10", "-m", "pytest"]
+
+    def test_matrix_placeholder_in_env(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["echo", "test"]
+    env: {VERSION: "${{ matrix.version }}"}
+    strategy:
+      matrix:
+        version: ["3.10", "3.11"]
+""")
+        spec = graph.spec("test_version-3.11")
+        assert spec.env == {"VERSION": "3.11"}
+
+    def test_matrix_placeholder_in_cwd(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["echo", "test"]
+    cwd: "/tmp/${{ matrix.os }}"
+    strategy:
+      matrix:
+        os: ["linux", "macos"]
+""")
+        spec = graph.spec("test_os-linux")
+        assert spec.cwd == Path("/tmp/linux")
+
+    def test_matrix_dependency_auto_expand(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    strategy:
+      matrix:
+        version: ["3.10", "3.11"]
+  test:
+    needs: [build]
+    cmd: ["echo", "test"]
+""")
+        test_spec = graph.spec("test")
+        assert set(test_spec.depends_on) == {"build_version-3.10", "build_version-3.11"}
+
+    def test_matrix_dependency_with_non_matrix_job(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  setup:
+    cmd: ["echo", "setup"]
+  build:
+    needs: [setup]
+    cmd: ["echo", "build"]
+    strategy:
+      matrix:
+        version: ["3.10", "3.11"]
+  deploy:
+    needs: [build, setup]
+    cmd: ["echo", "deploy"]
+""")
+        deploy_spec = graph.spec("deploy")
+        assert "setup" in deploy_spec.depends_on
+        assert "build_version-3.10" in deploy_spec.depends_on
+        assert "build_version-3.11" in deploy_spec.depends_on
+
+
+class TestConditions:
+    """条件解析测试。"""
+
+    def test_success_condition(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    if: "success()"
+""")
+        assert graph.spec("build").conditions == ()
+
+    def test_always_condition(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    if: "always()"
+""")
+        assert graph.spec("build").conditions == ()
+
+    def test_env_var_exists_condition(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    if: "env.CI"
+""")
+        spec = graph.spec("build")
+        assert len(spec.conditions) == 1
+        monkeypatch.delenv("CI", raising=False)
+        assert spec.conditions[0]({}) is False
+        monkeypatch.setenv("CI", "true")
+        assert spec.conditions[0]({}) is True
+
+    def test_env_var_equals_condition(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  deploy:
+    cmd: ["echo", "deploy"]
+    if: "env.CI == 'true'"
+""")
+        spec = graph.spec("deploy")
+        assert len(spec.conditions) == 1
+
+    def test_env_var_not_equals_condition(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  skip_deploy:
+    cmd: ["echo", "skip"]
+    if: "env.CI != 'true'"
+""")
+        spec = graph.spec("skip_deploy")
+        assert len(spec.conditions) == 1
+
+    def test_failure_condition_not_supported(self) -> None:
+        with pytest.raises(ValueError, match="failure"):
+            parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    if: "failure()"
+""")
+
+    def test_invalid_condition_raises(self) -> None:
+        with pytest.raises(ValueError, match="无法解析"):
+            parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    if: "invalid_expr"
+""")
+
+
+class TestGraphDefaults:
+    """图级默认值测试。"""
+
+    def test_defaults_retry(self) -> None:
+        graph = parse_yaml_string("""
+defaults:
+  retry: {max_attempts: 3, jitter: 0.5}
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        spec = graph.resolved_spec("build")
+        assert spec.retry.max_attempts == 3
+        assert spec.retry.jitter == 0.5
+
+    def test_defaults_verbose(self) -> None:
+        graph = parse_yaml_string("""
+defaults:
+  verbose: true
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        spec = graph.resolved_spec("build")
+        assert spec.verbose is True
+
+    def test_defaults_env(self) -> None:
+        graph = parse_yaml_string("""
+defaults:
+  env: {CI: "true"}
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        spec = graph.resolved_spec("build")
+        assert spec.env == {"CI": "true"}
+
+    def test_defaults_timeout(self) -> None:
+        graph = parse_yaml_string("""
+defaults:
+  timeout: 120
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        assert graph.resolved_spec("build").timeout == 120.0
+
+    def test_defaults_cwd(self, tmp_path: Path) -> None:
+        graph = parse_yaml_string(f"""
+defaults:
+  cwd: {tmp_path}
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        assert graph.resolved_spec("build").cwd == tmp_path
+
+    def test_defaults_tags(self) -> None:
+        graph = parse_yaml_string("""
+defaults:
+  tags: [ci]
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        assert graph.resolved_spec("build").tags == ("ci",)
+
+    def test_defaults_priority(self) -> None:
+        graph = parse_yaml_string("""
+defaults:
+  priority: 5
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        assert graph.resolved_spec("build").priority == 5
+
+    def test_defaults_continue_on_error(self) -> None:
+        graph = parse_yaml_string("""
+defaults:
+  continue-on-error: true
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        assert graph.resolved_spec("build").continue_on_error is True
+
+    def test_defaults_concurrency_key(self) -> None:
+        graph = parse_yaml_string("""
+defaults:
+  concurrency-key: lock
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        assert graph.resolved_spec("build").concurrency_key == "lock"
+
+    def test_strategy_at_top_level(self) -> None:
+        graph = parse_yaml_string("""
+strategy: thread
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+        assert graph.defaults.strategy == "thread"
+
+
+class TestErrorHandling:
+    """错误处理测试。"""
+
+    def test_missing_jobs_raises(self) -> None:
+        with pytest.raises(ValueError, match="jobs"):
+            parse_yaml_string("strategy: thread")
+
+    def test_empty_jobs_raises(self) -> None:
+        with pytest.raises(ValueError, match="jobs"):
+            parse_yaml_string("jobs: {}")
+
+    def test_job_missing_cmd_and_run_raises(self) -> None:
+        with pytest.raises(ValueError, match="cmd 或 run"):
+            parse_yaml_string("""
+jobs:
+  bad:
+    timeout: 10
+""")
+
+    def test_non_mapping_root_raises(self) -> None:
+        with pytest.raises(ValueError, match="根节点"):
+            parse_yaml_string("- item")
+
+    def test_non_mapping_job_raises(self) -> None:
+        with pytest.raises(ValueError, match="映射"):
+            parse_yaml_string("""
+jobs:
+  bad: "just a string"
+""")
+
+    def test_undefined_matrix_var_raises(self) -> None:
+        with pytest.raises(ValueError, match=r"matrix\.undefined"):
+            parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["echo", "${{ matrix.undefined }}"]
+    strategy:
+      matrix:
+        version: ["3.10"]
+""")
+
+    def test_invalid_retry_raises(self) -> None:
+        with pytest.raises(ValueError, match="retry"):
+            parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    retry: "not a mapping"
+""")
+
+    def test_jobs_not_mapping_raises(self) -> None:
+        with pytest.raises(ValueError, match="jobs 必须是映射"):
+            parse_yaml_string("jobs: [1, 2, 3]")
+
+    def test_defaults_not_mapping_raises(self) -> None:
+        with pytest.raises(ValueError, match="defaults 必须是映射"):
+            parse_yaml_string("""
+defaults: "invalid"
+jobs:
+  build:
+    cmd: ["echo", "build"]
+""")
+
+    def test_matrix_value_not_list_raises(self) -> None:
+        with pytest.raises(ValueError, match="必须是列表"):
+            parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["echo", "test"]
+    strategy:
+      matrix:
+        version: "3.10"
+""")
+
+    def test_env_not_mapping_raises(self) -> None:
+        with pytest.raises(ValueError, match="env 必须是映射"):
+            parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    env: "invalid"
+""")
+
+    def test_need_referencing_unknown_job_raises(self) -> None:
+        with pytest.raises(px.MissingDependencyError, match="unknown"):
+            parse_yaml_string("""
+jobs:
+  build:
+    needs: [unknown]
+    cmd: ["echo", "build"]
+""")
+
+    def test_empty_if_expr(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  build:
+    cmd: ["echo", "build"]
+    if: ""
+""")
+        assert graph.spec("build").conditions == ()
+
+    def test_cmd_as_string_with_matrix(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: "echo ${{ matrix.version }}"
+    strategy:
+      matrix:
+        version: ["3.10"]
+""")
+        assert graph.spec("test_version-3.10").cmd == "echo 3.10"
+
+    def test_matrix_not_mapping_returns_none(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["echo", "test"]
+    strategy:
+      matrix: "invalid"
+""")
+        assert "test" in graph
+
+    def test_empty_matrix_keys(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["echo", "test"]
+    strategy:
+      matrix: {}
+""")
+        assert "test" in graph
+
+
+class TestLoadFromFile:
+    """从文件加载测试。"""
+
+    def test_load_yaml_file(self, tmp_path: Path) -> None:
+        yaml_file = tmp_path / "pipeline.yaml"
+        yaml_file.write_text(
+            """
+jobs:
+  hello:
+    cmd: ["echo", "hello"]
+""",
+            encoding="utf-8",
+        )
+        graph = load_yaml(yaml_file)
+        assert "hello" in graph
+
+    def test_from_yaml_classmethod(self, tmp_path: Path) -> None:
+        yaml_file = tmp_path / "pipeline.yaml"
+        yaml_file.write_text(
+            """
+jobs:
+  hello:
+    cmd: ["echo", "hello"]
+""",
+            encoding="utf-8",
+        )
+        graph = px.Graph.from_yaml(yaml_file)
+        assert "hello" in graph
+
+    def test_load_yaml_file_not_found(self) -> None:
+        with pytest.raises(OSError):
+            load_yaml("/nonexistent/pipeline.yaml")
+
+
+class TestExecution:
+    """端到端执行测试。"""
+
+    def test_run_simple_yaml_graph(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  hello:
+    cmd: ["echo", "hello"]
+""")
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        assert report.result_of("hello").status == TaskStatus.SUCCESS
+
+    def test_run_yaml_with_dependency(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  a:
+    cmd: ["echo", "from-a"]
+  b:
+    needs: [a]
+    cmd: ["echo", "from-b"]
+""")
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        assert "a" in report.results
+        assert "b" in report.results
+
+    def test_run_matrix_graph(self) -> None:
+        graph = parse_yaml_string("""
+jobs:
+  test:
+    cmd: ["echo", "test-${{ matrix.version }}"]
+    strategy:
+      matrix:
+        version: ["3.10", "3.11"]
+""")
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        assert report.result_of("test_version-3.10").status == TaskStatus.SUCCESS
+        assert report.result_of("test_version-3.11").status == TaskStatus.SUCCESS
+
+    def test_condition_skips_task(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("SKIP_ME", raising=False)
+        graph = parse_yaml_string("""
+jobs:
+  conditional:
+    cmd: ["echo", "should-skip"]
+    if: "env.SKIP_ME"
+""")
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        assert report.result_of("conditional").status == TaskStatus.SKIPPED
+
+    def test_condition_executes_when_env_set(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("RUN_ME", "yes")
+        graph = parse_yaml_string("""
+jobs:
+  conditional:
+    cmd: ["echo", "should-run"]
+    if: "env.RUN_ME"
+""")
+        report = px.run(graph, strategy="sequential")
+        assert report.success
+        assert report.result_of("conditional").status == TaskStatus.SUCCESS
+
+
+class TestPublicAPI:
+    """公共 API 导出测试。"""
+
+    def test_load_yaml_exported(self) -> None:
+        assert hasattr(px, "load_yaml")
+        assert px.load_yaml is load_yaml
+
+    def test_parse_yaml_string_exported(self) -> None:
+        assert hasattr(px, "parse_yaml_string")
+        assert px.parse_yaml_string is parse_yaml_string
+
+    def test_graph_from_yaml_classmethod(self) -> None:
+        assert hasattr(px.Graph, "from_yaml")

uv.lock

@@ -571,7 +571,7 @@ name = "exceptiongroup"
 version = "1.3.1"
 source = { registry = "https://mirrors.aliyun.com/pypi/simple/" }
 dependencies = [
-    { name = "typing-extensions", marker = "python_full_version < '3.13'" },
+    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
 ]
 sdist = { url = "https://mirrors.aliyun.com/pypi/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219" }
 wheels = [
@@ -1220,6 +1220,7 @@ name = "pyflowx"
 version = "0.4.8"
 source = { editable = "." }
 dependencies = [
+    { name = "pyyaml" },
     { name = "rich" },
     { name = "typer" },
     { name = "typing-extensions", marker = "python_full_version < '3.13'" },
@@ -1278,6 +1279,7 @@ requires-dist = [
     { name = "pytest-html", marker = "extra == 'dev'", specifier = ">=4.1.1" },
     { name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.14.0" },
     { name = "pytest-xdist", marker = "extra == 'dev'", specifier = ">=3.6.1" },
+    { name = "pyyaml", specifier = ">=6.0.1" },
     { name = "rich", specifier = ">=13.7.0" },
     { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.8.0" },
     { name = "sphinx", marker = "extra == 'docs'", specifier = ">=7.0" },