feat: RunReport 结果序列化支持 to_json/to_dict/to_csv/from_json

新增 RunReport 序列化方法，支持 JSON/dict/CSV 多格式导出与反序列化重建；
非 JSON 可序列化值默认回退到 repr，可通过 value_serializer 自定义。

.trae/docs/iter-11-result-serialization.md

@@ -0,0 +1,85 @@
+# 迭代 11：结果序列化
+
+## 本轮目标
+
+1. `RunReport.to_dict(value_serializer=)` —— 转为 JSON 兼容字典
+2. `RunReport.to_json(indent=, value_serializer=)` —— 序列化为 JSON 字符串
+3. `RunReport.to_csv(include_value=)` —— 表格 CSV 导出
+4. `RunReport.from_json(text)` —— 反序列化（仅供查询，不重新执行）
+
+## 改动文件清单
+
+- `src/pyflowx/report.py` —— 新增 `to_dict`/`to_json`/`to_csv`/`from_json` 方法
+- `tests/test_report.py` —— 新增序列化测试
+- `README.md` —— 更新文档
+
+## 关键设计
+
+### 1. 值序列化策略
+
+任务返回值可能是任意 Python 对象（不可 JSON 序列化）。默认策略：
+- `None` 直接返回 `None`
+- 可 JSON 序列化的原样返回
+- 不可序列化的回退到 `repr(value)`
+- 调用方可通过 `value_serializer` 自定义
+
+### 2. to_dict 结构
+
+```python
+{
+    "success": bool,
+    "summary": {...},
+    "results": [
+        {
+            "name": str,
+            "status": str,  # TaskStatus.value
+            "value": Any,
+            "error": str | None,  # repr(error)
+            "attempts": int,
+            "started_at": str | None,  # ISO 8601
+            "finished_at": str | None,
+            "duration_seconds": float | None,
+            "reason": str | None,
+            "tags": list[str],
+            "depends_on": list[str],
+        },
+        ...,
+    ],
+}
+```
+
+`results` 为列表（保序），便于 CSV/JSON 数组导出。
+
+### 3. to_csv
+
+CSV 列：`name, status, attempts, duration_seconds, started_at, finished_at, error, reason[, value]`
+
+`include_value=False` 时省略 value 列（值可能含逗号/换行符，干扰解析）。
+
+### 4. from_json
+
+- 反序列化只重建可序列化字段
+- `TaskSpec` 的 `fn`/`cmd` 等不可恢复，使用 noop 占位
+- 重建后的 report **仅供查询/分析**，不能用于重新执行
+- `datetime` 通过 `fromisoformat` 还原
+
+## 验收标准
+
+- to_dict/to_json/to_csv/from_json 全部实现
+- 非 JSON 序列化值有合理回退（repr）
+- 自定义 value_serializer 可用
+- from_json 可 round-trip 已序列化的 report
+- 覆盖率 ≥ 95%
+- ruff / pyrefly / pytest 全部通过
+
+## 验证结果
+
+- **ruff**：All checks passed（含 format）
+- **pyrefly**：0 errors
+- **pytest**：1257 passed（含 27 个新增序列化测试）
+- **覆盖率**：97.29%（report.py 99%，仅 `_noop_fn` 占位函数体未覆盖，与 `task.py:_task_noop` 同模式）
+- **修复**：扩展 `_make_result` 测试辅助以支持 tags/depends_on/reason 参数
+
+## 遗留事项
+
+- 无

README.md

@@ -37,6 +37,7 @@ PyFlowX 把"任务依赖"这件事做到极致简单：**参数名就是依赖
 - **流式获取** —— `run_iter()` 生成器逐个 yield 任务结果，适用于大量任务逐个处理
 - **任务取消** —— `CancelToken` 线程安全取消令牌，支持外部取消与 `KeyboardInterrupt` 优雅停止
 - **子图执行** —— `run(only=[...], tags=[...])` 按名称或标签运行图的子集，自动包含传递依赖
+- **结果序列化** —— `RunReport.to_json()` / `to_dict()` / `to_csv()` 多格式导出运行结果，`from_json()` 支持反序列化重建
 - **状态后端** —— `MemoryBackend` / `JSONBackend` / `SQLiteBackend`（WAL 模式，适合大规模任务）
 - **YAML 任务编排** —— GitHub Actions 风格 `jobs`/`needs`/`strategy.matrix`/`if` 条件，`pf yamlrun pipeline.yaml` 一键执行
 - **最小依赖** —— `rich` + `typer` + `typing-extensions`（3.13 以下）+ `pyyaml`
@@ -537,6 +538,41 @@ pf yamlrun pipeline.yaml --only report,deploy
 pf yamlrun pipeline.yaml --tags ingest,report
 ```
 
+### 结果序列化
+
+`RunReport` 支持将运行结果导出为 JSON / dict / CSV，便于持久化、跨进程传输或对接外部工具：
+
+```python
+report = px.run(graph, strategy="sequential")
+
+# JSON 字符串（含 success/summary/results）
+text = report.to_json(indent=2)
+
+# JSON 兼容字典
+data = report.to_dict()
+
+# CSV 表格导出（不含 value 列以避免特殊字符干扰）
+csv_text = report.to_csv(include_value=False)
+```
+
+任务返回值可能是任意 Python 对象（不一定 JSON 可序列化）。默认策略：
+可序列化原样保留，不可序列化回退到 `repr(value)`；调用方可通过
+`value_serializer` 自定义：
+
+```python
+# 自定义值序列化（如把 Path 转为字符串）
+text = report.to_json(value_serializer=str)
+```
+
+`from_json()` 可从 JSON 字符串重建 `RunReport`，**仅供查询/分析**，不能重新
+执行（`TaskSpec` 的 `fn`/`cmd` 等不可序列化字段无法恢复）：
+
+```python
+restored = px.RunReport.from_json(text)
+print(restored["double"])  # 任务返回值
+print(restored.result_of("a").status)  # TaskStatus
+```
+
 ## 开发
 
 ```bash

src/pyflowx/report.py

@@ -6,11 +6,40 @@
 
 from __future__ import annotations
 
-from collections.abc import Iterator
+import csv
+import io
+import json
+from collections.abc import Callable, Iterator
 from dataclasses import dataclass, field
+from datetime import datetime
 from typing import Any
 
-from .task import TaskResult, TaskStatus
+from .task import TaskResult, TaskSpec, TaskStatus
+
+
+def _noop_fn() -> None:
+    """反序列化重建报告时的占位 ``fn``。"""
+    return None
+
+
+def _serialize_value(
+    value: Any,
+    value_serializer: Callable[[Any], Any] | None = None,
+) -> Any:
+    """序列化任务返回值为 JSON 兼容形式。
+
+    ``None`` 直接返回；提供 ``value_serializer`` 时由其转换；否则尝试
+    JSON 序列化，失败则回退到 ``repr(value)``，保证任意值都能落盘。
+    """
+    if value is None:
+        return None
+    if value_serializer is not None:
+        return value_serializer(value)
+    try:
+        _ = json.dumps(value)
+    except (TypeError, ValueError):
+        return repr(value)
+    return value
 
 
 @dataclass
@@ -94,3 +123,140 @@ class RunReport:
             err = f" error={r.error!r}" if r.error else ""
             lines.append(f"  {name}: {r.status.value} ({dur} attempts={r.attempts}){err}")
         return "\n".join(lines)
+
+    # ---- 序列化 ------------------------------------------------------- #
+    def to_dict(
+        self,
+        value_serializer: Callable[[Any], Any] | None = None,
+    ) -> dict[str, Any]:
+        """转换为 JSON 兼容的字典。
+
+        参数
+        ----
+        value_serializer:
+            自定义任务值序列化函数。``None`` 时尝试 JSON 兼容转换，
+            失败回退到 ``repr()``。
+
+        返回结构包含顶层 ``success``、``summary`` 与 ``results`` 列表
+        （按完成顺序保序），每个结果项含 name/status/value/error/
+        attempts/started_at/finished_at/duration_seconds/reason/tags/
+        depends_on 字段。
+        """
+        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),
+                }
+            )
+        return {
+            "success": self.success,
+            "summary": self.summary(),
+            "results": results_list,
+        }
+
+    def to_json(
+        self,
+        indent: int = 2,
+        value_serializer: Callable[[Any], Any] | None = None,
+    ) -> str:
+        """序列化为 JSON 字符串。
+
+        参数
+        ----
+        indent:
+            JSON 缩进空格数，``0`` 表示紧凑单行。
+        value_serializer:
+            自定义任务值序列化函数，透传给 :meth:`to_dict`。
+        """
+        return json.dumps(
+            self.to_dict(value_serializer),
+            ensure_ascii=False,
+            indent=indent if indent > 0 else None,
+        )
+
+    def to_csv(self, include_value: bool = True) -> str:
+        """导出为 CSV 字符串。
+
+        列顺序：``name, status, attempts, duration_seconds, started_at,
+        finished_at, error, reason`` + 可选 ``value``。
+
+        参数
+        ----
+        include_value:
+            是否包含 value 列。任务返回值可能含逗号/换行符干扰解析，
+            不需要值时设为 ``False`` 得到更干净输出。
+        """
+        buf = io.StringIO()
+        writer = csv.writer(buf)
+        header = [
+            "name",
+            "status",
+            "attempts",
+            "duration_seconds",
+            "started_at",
+            "finished_at",
+            "error",
+            "reason",
+        ]
+        if include_value:
+            header.append("value")
+        writer.writerow(header)
+        for name, r in self.results.items():
+            row = [
+                name,
+                r.status.value,
+                r.attempts,
+                f"{r.duration:.6f}" if r.duration is not None else "",
+                r.started_at.isoformat() if r.started_at else "",
+                r.finished_at.isoformat() if r.finished_at else "",
+                repr(r.error) if r.error else "",
+                r.reason or "",
+            ]
+            if include_value:
+                row.append(_serialize_value(r.value))
+            writer.writerow(row)
+        return buf.getvalue()
+
+    @classmethod
+    def from_json(cls, text: str) -> RunReport:
+        """从 JSON 字符串重建 :class:`RunReport`。
+
+        注意
+        ----
+        只能重建可序列化字段；:class:`TaskSpec` 的 ``fn``/``cmd`` 等
+        不可序列化字段无法恢复，重建的 spec 仅含 ``name``/``tags``/
+        ``depends_on``。重建后的 report 仅供查询/分析，**不能用于重新执行**。
+        """
+        data = json.loads(text)
+        report = cls(success=data.get("success", True))
+        for entry in data.get("results", []):
+            spec: TaskSpec[Any] = TaskSpec(
+                name=entry["name"],
+                fn=_noop_fn,
+                tags=tuple(entry.get("tags", ())),
+                depends_on=tuple(entry.get("depends_on", ())),
+            )
+            started = entry.get("started_at")
+            finished = entry.get("finished_at")
+            result: TaskResult[Any] = TaskResult(
+                spec=spec,
+                status=TaskStatus(entry["status"]),
+                value=entry.get("value"),
+                attempts=entry.get("attempts", 0),
+                started_at=datetime.fromisoformat(started) if started else None,
+                finished_at=datetime.fromisoformat(finished) if finished else None,
+                reason=entry.get("reason"),
+            )
+            report.results[spec.name] = result
+        return report

tests/test_report.py

@@ -2,9 +2,15 @@
 
 from __future__ import annotations
 
+import csv
+import io
+import json
 from datetime import datetime, timedelta
+from pathlib import Path
 from typing import Any
 
+import pytest
+
 import pyflowx as px
 from pyflowx.task import TaskResult, TaskSpec, TaskStatus
 
@@ -20,9 +26,12 @@ def _make_result(
     error: BaseException | None = None,
     duration: float = 0.5,
     attempts: int = 1,
+    tags: tuple[str, ...] = (),
+    depends_on: tuple[str, ...] = (),
+    reason: str | None = None,
 ) -> TaskResult[Any]:
     """构造测试用 TaskResult 实例."""
-    spec: TaskSpec[Any] = TaskSpec[Any](name, _fn)
+    spec: TaskSpec[Any] = TaskSpec[Any](name, _fn, tags=tags, depends_on=depends_on)
     start = datetime(2024, 1, 1, 0, 0, 0)
     # 用 timedelta 精确表达秒数，避免 int() 截断小数
     end = start + timedelta(seconds=duration) if duration else None
@@ -34,6 +43,7 @@ def _make_result(
         attempts=attempts,
         started_at=start,
         finished_at=end,
+        reason=reason,
     )
 
 
@@ -172,3 +182,333 @@ class TestRunReportQueries:
         report.results["a"] = TaskResult[Any](spec=spec, status=TaskStatus.PENDING)
         durs = report.durations()
         assert durs["a"] == 0.0
+
+
+# ---------------------------------------------------------------------- #
+# 序列化：to_dict / to_json / to_csv / from_json
+# ---------------------------------------------------------------------- #
+class TestRunReportToDict:
+    """测试 RunReport.to_dict."""
+
+    def test_to_dict_basic(self) -> None:
+        """应输出包含 success/summary/results 的字典。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=42, duration=0.5, tags=("ingest",), depends_on=("x",))
+        d = report.to_dict()
+        assert d["success"] is True
+        assert d["summary"]["total_tasks"] == 1
+        assert len(d["results"]) == 1
+        entry = d["results"][0]
+        assert entry["name"] == "a"
+        assert entry["status"] == "success"
+        assert entry["value"] == 42
+        assert entry["error"] is None
+        assert entry["attempts"] == 1
+        assert entry["started_at"] == "2024-01-01T00:00:00"
+        assert entry["finished_at"] == "2024-01-01T00:00:00.500000"
+        assert entry["duration_seconds"] == 0.5
+        assert entry["reason"] is None
+        assert entry["tags"] == ["ingest"]
+        assert entry["depends_on"] == ["x"]
+
+    def test_to_dict_preserves_order(self) -> None:
+        """results 列表应按插入顺序保留。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a")
+        report.results["b"] = _make_result("b")
+        report.results["c"] = _make_result("c")
+        d = report.to_dict()
+        assert [r["name"] for r in d["results"]] == ["a", "b", "c"]
+
+    def test_to_dict_with_error(self) -> None:
+        """error 应转为 repr 字符串。"""
+        report = px.RunReport(success=False)
+        report.results["a"] = _make_result("a", status=TaskStatus.FAILED, error=ValueError("boom"))
+        d = report.to_dict()
+        entry = d["results"][0]
+        assert entry["error"] == "ValueError('boom')"
+        assert entry["status"] == "failed"
+
+    def test_to_dict_none_value(self) -> None:
+        """None 值应保持为 None。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=None)
+        d = report.to_dict()
+        assert d["results"][0]["value"] is None
+
+    def test_to_dict_non_serializable_value_falls_back_to_repr(self) -> None:
+        """非 JSON 可序列化的值应回退到 repr。"""
+        report = px.RunReport()
+        # Path 对象不可 JSON 序列化
+        report.results["a"] = _make_result("a", value=Path("/tmp/x"))
+        d = report.to_dict()
+        assert d["results"][0]["value"] == "PosixPath('/tmp/x')"
+
+    def test_to_dict_with_custom_serializer(self) -> None:
+        """自定义 value_serializer 应被调用。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=Path("/tmp/x"))
+        d = report.to_dict(value_serializer=str)
+        assert d["results"][0]["value"] == "/tmp/x"
+
+    def test_to_dict_no_timestamps(self) -> None:
+        """未开始/未结束的任务 started_at/finished_at/duration 应为 None。"""
+        report = px.RunReport()
+        spec: TaskSpec[Any] = TaskSpec[Any]("a", _fn)  # type: ignore[arg-type]
+        report.results["a"] = TaskResult[Any](spec=spec, status=TaskStatus.PENDING)
+        d = report.to_dict()
+        entry = d["results"][0]
+        assert entry["started_at"] is None
+        assert entry["finished_at"] is None
+        assert entry["duration_seconds"] is None
+
+    def test_to_dict_with_reason(self) -> None:
+        """reason 字段应被保留。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", status=TaskStatus.SKIPPED, duration=0.0, reason="条件不满足")
+        d = report.to_dict()
+        assert d["results"][0]["reason"] == "条件不满足"
+
+    def test_to_dict_empty_report(self) -> None:
+        """空报告也应可序列化。"""
+        report = px.RunReport()
+        d = report.to_dict()
+        assert d["success"] is True
+        assert d["results"] == []
+        assert d["summary"]["total_tasks"] == 0
+
+
+class TestRunReportToJson:
+    """测试 RunReport.to_json."""
+
+    def test_to_json_basic(self) -> None:
+        """应输出合法 JSON 字符串。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=42)
+        s = report.to_json()
+        data = json.loads(s)
+        assert data["success"] is True
+        assert data["results"][0]["value"] == 42
+
+    def test_to_json_indent_zero_compact(self) -> None:
+        """indent=0 应输出紧凑单行（无换行）。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=1)
+        s = report.to_json(indent=0)
+        assert "\n" not in s
+
+    def test_to_json_indent_default(self) -> None:
+        """默认 indent=2 应输出多行格式。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=1)
+        s = report.to_json()
+        assert "\n" in s
+
+    def test_to_json_ensure_chinese(self) -> None:
+        """中文字符不应被转义（ensure_ascii=False）。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", status=TaskStatus.SKIPPED, duration=0.0, reason="条件不满足")
+        s = report.to_json()
+        assert "条件不满足" in s
+        assert "\\u" not in s
+
+    def test_to_json_with_value_serializer(self) -> None:
+        """to_json 应透传 value_serializer。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=Path("/tmp/x"))
+        s = report.to_json(value_serializer=str)
+        data = json.loads(s)
+        assert data["results"][0]["value"] == "/tmp/x"
+
+
+class TestRunReportToCsv:
+    """测试 RunReport.to_csv."""
+
+    def test_to_csv_basic(self) -> None:
+        """应输出带表头与数据行的 CSV。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=42, duration=0.5)
+        s = report.to_csv()
+        reader = csv.reader(io.StringIO(s))
+        rows = list(reader)
+        assert rows[0] == [
+            "name",
+            "status",
+            "attempts",
+            "duration_seconds",
+            "started_at",
+            "finished_at",
+            "error",
+            "reason",
+            "value",
+        ]
+        assert rows[1][0] == "a"
+        assert rows[1][1] == "success"
+        assert rows[1][2] == "1"
+        assert rows[1][3] == "0.500000"
+        assert rows[1][8] == "42"
+
+    def test_to_csv_without_value(self) -> None:
+        """include_value=False 应省略 value 列。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=42)
+        s = report.to_csv(include_value=False)
+        reader = csv.reader(io.StringIO(s))
+        rows = list(reader)
+        assert rows[0] == [
+            "name",
+            "status",
+            "attempts",
+            "duration_seconds",
+            "started_at",
+            "finished_at",
+            "error",
+            "reason",
+        ]
+        assert len(rows[1]) == 8
+
+    def test_to_csv_empty_value_quoted(self) -> None:
+        """含逗号的值应被引号包裹（csv 模块自动处理）。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value="a,b,c")
+        s = report.to_csv()
+        # csv 模块会自动给含特殊字符的值加引号
+        assert '"a,b,c"' in s
+
+    def test_to_csv_empty_report(self) -> None:
+        """空报告应只输出表头。"""
+        report = px.RunReport()
+        s = report.to_csv()
+        reader = csv.reader(io.StringIO(s))
+        rows = list(reader)
+        assert len(rows) == 1
+        assert rows[0][0] == "name"
+
+    def test_to_csv_non_serializable_value(self) -> None:
+        """非 JSON 可序列化值应回退到 repr。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a", value=Path("/tmp/x"))
+        s = report.to_csv()
+        assert "PosixPath('/tmp/x')" in s
+
+
+class TestRunReportFromJson:
+    """测试 RunReport.from_json."""
+
+    def test_from_json_round_trip(self) -> None:
+        """to_json → from_json 应能恢复所有可序列化字段。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result(
+            "a",
+            value=42,
+            duration=0.5,
+            tags=("ingest",),
+            depends_on=("x",),
+        )
+        report.results["b"] = _make_result(
+            "b",
+            status=TaskStatus.SKIPPED,
+            duration=0.0,
+            reason="条件不满足",
+        )
+        text = report.to_json()
+        restored = px.RunReport.from_json(text)
+        assert restored.success == report.success
+        assert len(restored) == 2
+        assert restored["a"] == 42
+        assert restored.result_of("a").status == TaskStatus.SUCCESS
+        assert restored.result_of("a").attempts == 1
+        assert restored.result_of("a").duration == 0.5
+        assert restored.result_of("a").started_at == datetime(2024, 1, 1, 0, 0, 0)
+        assert restored.result_of("a").finished_at == datetime(2024, 1, 1, 0, 0, 0, 500000)
+        assert restored.result_of("a").spec.tags == ("ingest",)
+        assert restored.result_of("a").spec.depends_on == ("x",)
+        assert restored.result_of("b").status == TaskStatus.SKIPPED
+        assert restored.result_of("b").reason == "条件不满足"
+
+    def test_from_json_preserves_order(self) -> None:
+        """from_json 应保持 results 顺序。"""
+        report = px.RunReport()
+        report.results["a"] = _make_result("a")
+        report.results["b"] = _make_result("b")
+        report.results["c"] = _make_result("c")
+        restored = px.RunReport.from_json(report.to_json())
+        assert list(restored) == ["a", "b", "c"]
+
+    def test_from_json_empty_report(self) -> None:
+        """空报告应可正确反序列化。"""
+        report = px.RunReport()
+        restored = px.RunReport.from_json(report.to_json())
+        assert len(restored) == 0
+        assert restored.success is True
+
+    def test_from_json_invalid_status_raises(self) -> None:
+        """非法 status 值应抛 ValueError（TaskStatus 枚举校验）。"""
+        data = {
+            "success": True,
+            "summary": {},
+            "results": [{"name": "a", "status": "unknown_status"}],
+        }
+        with pytest.raises(ValueError):
+            _ = px.RunReport.from_json(json.dumps(data))
+
+    def test_from_json_missing_optional_fields(self) -> None:
+        """缺少可选字段（tags/depends_on/attempts）应使用默认值。"""
+        data = {
+            "success": False,
+            "results": [{"name": "a", "status": "success"}],
+        }
+        restored = px.RunReport.from_json(json.dumps(data))
+        assert restored.success is False
+        assert restored.result_of("a").attempts == 0
+        assert restored.result_of("a").spec.tags == ()
+        assert restored.result_of("a").spec.depends_on == ()
+
+    def test_from_json_invalid_json_raises(self) -> None:
+        """非法 JSON 字符串应抛 JSONDecodeError。"""
+        with pytest.raises(json.JSONDecodeError):
+            _ = px.RunReport.from_json("not json{")
+
+
+class TestRunReportSerializationIntegration:
+    """序列化与实际执行流程的集成测试。"""
+
+    def test_real_run_serialization_round_trip(self) -> None:
+        """实际 run() 产出的报告应可序列化/反序列化。"""
+
+        def extract() -> list[int]:
+            return [1, 2, 3]
+
+        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",)),
+            ]
+        )
+        report = px.run(graph, strategy="sequential")
+
+        # 序列化 → 反序列化
+        text = report.to_json()
+        restored = px.RunReport.from_json(text)
+
+        assert restored.success is True
+        assert restored["double"] == [2, 4, 6]
+        assert restored.result_of("extract").spec.tags == ("ingest",)
+        assert restored.result_of("double").spec.depends_on == ("extract",)
+
+    def test_csv_export_from_real_run(self) -> None:
+        """实际 run() 报告应可导出 CSV。"""
+
+        def extract() -> list[int]:
+            return [1, 2, 3]
+
+        graph = px.Graph.from_specs([px.TaskSpec("extract", extract)])
+        report = px.run(graph, strategy="sequential")
+        csv_text = report.to_csv()
+        assert "extract" in csv_text
+        assert "success" in csv_text
+        # 值应被 repr 化（列表用 repr）
+        assert "[1, 2, 3]" in csv_text