diff --git a/skills/product-data/README.md b/skills/product-data/README.md
index 4ec997b..2ee960d 100644
--- a/skills/product-data/README.md
+++ b/skills/product-data/README.md
@@ -43,7 +43,8 @@ cat input.json | python skills/product-data/scripts/<tool>.py
 | --- | --- |
 | `scripts/normalize_dataset_draft.py` | 把 dataset draft text v1 转成 canonical records |
 | `scripts/validate_dataset_records.py` | 校验 canonical records |
-| `scripts/export_dataset_records.py` | 导出紧凑 JSONL/JSON |
+| `scripts/export_dataset_records.py` | 导出紧凑 JSONL/JSON，并默认生成同目录 `records.csv` 表格 |
+| `scripts/export_dataset_table.py` | 已有 canonical records 时，只补生成同事流转表格 |
 
 ## 平台内优先级
 
diff --git a/skills/product-data/SKILL.md b/skills/product-data/SKILL.md
index dce614d..0f48579 100644
--- a/skills/product-data/SKILL.md
+++ b/skills/product-data/SKILL.md
@@ -28,6 +28,7 @@ skills/product-data/
     normalize_dataset_draft.py
     validate_dataset_records.py
     export_dataset_records.py
+    export_dataset_table.py
 ```
 
 在 ZK Data Agent 平台里，优先使用已注册的 `data_agent_*` 工具，因为它们带有平台级 review 状态、确认门禁和会话 output 路由。
@@ -164,9 +165,16 @@ review 展示必须简短清晰，不要重复解释工具和流程。每次 rev
 11. 生成 dataset draft text v1。
 12. 调用 `data_agent_normalize_dataset_draft`，必须传入 `confirmed_plan_id`。
 13. 调用 `data_agent_validate_dataset_records`。
-14. 如果用户要求落盘 canonical records，调用 `data_agent_export_dataset_records`，`output_path` 固定传 `output/records.jsonl`，默认导出紧凑 JSONL，不要用 `write_file` 手写 JSON。
+14. 如果用户要求落盘 canonical records，调用 `data_agent_export_dataset_records`，`output_path` 固定传 `output/records.jsonl`，默认导出紧凑 JSONL，并同时生成同目录 `records.csv` 表格，不要用 `write_file` 手写 JSON 或 CSV。
 15. 本阶段默认只推进到 canonical metadata records；除非用户另行要求，不做最终训练/评测格式导出。
 
+生成阶段要避免一次性把大量数据塞进工具参数：
+
+- 单次 `data_agent_normalize_dataset_draft` 最多处理 8 条 case；计划数量更多时，分批生成、分批 normalize，再汇总校验和导出。
+- dataset draft 中的 Agent target 推荐写成单引号形式，例如 `target: Agent(tag='地图导航')`。工具会规范化为 `Agent(tag="地图导航")`，这样可以降低 tool call JSON 里双引号转义失败的概率。
+- 如果 draft 已经保存在文件中，优先传 `draft_path`，不要再把大段 `draft_text` 作为工具参数传入。
+- 校验和导出 records 时，如果 records 已经保存在 JSON/JSONL 文件中，优先传 `records_path`。
+
 ## 数据生成输出格式
 
 当需要生成数据样本时，默认使用 dataset draft text v1，不要直接输出 JSON、JSONL、CSV 或最终表格格式。除非用户明确要求机器可读格式，否则优先输出便于人工 review 的文本格式。
@@ -196,6 +204,7 @@ notes: 可选，说明覆盖的问题或边界
 - `小爱:` 表示小爱回复 tts。
 - 最后一个 `用户:` 是本轮 query。
 - `target:` 是本轮 query 对应的监督标签，必须存在。
+- 为避免工具参数 JSON 转义失败，dataset draft 里 Agent 标签优先写 `target: Agent(tag='xxx')`；转换工具会统一规范为 `Agent(tag="xxx")`。
 - 单轮数据只需要写一行 `用户:`，然后写 `target:`。
 - 多轮数据需要按照时间顺序写多组 `用户:` / `小爱:`。
 - 多轮数据的最后一轮只写 `用户:` 和 `target:`，不要写最后一轮 `小爱:`，因为本轮 query 不包含 tts。
@@ -207,7 +216,7 @@ notes: 可选，说明覆盖的问题或边界
 
 `data_agent_normalize_dataset_draft` 会把 dataset draft text 转成 canonical records，并统一补齐 `record_id`、`source`、`timestamp`、`context`、`target_type` 等机械字段。
 
-canonical records 落盘必须使用 `data_agent_export_dataset_records`，默认格式是紧凑 JSONL：一行一个 canonical record，不带外层数组，不手写缩进 JSON。默认 `output_path` 固定传 `output/records.jsonl`；不要传 `output/<数据集名>/records.jsonl`，不要传 `tasks/...`，不要自定义文件名。只有用户明确要求兼容旧文件时，才使用 `output_format="json"` 导出紧凑 JSON 数组，此时工具会归一为 `output/records.json`。
+canonical records 落盘必须使用 `data_agent_export_dataset_records`，默认格式是紧凑 JSONL：一行一个 canonical record，不带外层数组，不手写缩进 JSON。默认 `output_path` 固定传 `output/records.jsonl`；不要传 `output/<数据集名>/records.jsonl`，不要传 `tasks/...`，不要自定义文件名。工具默认同时在同目录生成 `records.csv`，用于同事之间流转和表格查看。只有用户明确要求兼容旧文件时，才使用 `output_format="json"` 导出紧凑 JSON 数组，此时工具会归一为 `output/records.json`，但表格仍然是同目录 `records.csv`。
 
 当前 canonical record v1 工作格式：
 
@@ -337,12 +346,18 @@ skills/product-data/scripts/normalize_dataset_draft.py
 
 ```json
 {
-  "draft_text": "# dataset_label: 地图餐饮边界\n\n### case: 找附近美食\n用户: 附近有什么好吃的\ntarget: Agent(tag=\"餐饮服务\")",
+  "draft_text": "# dataset_label: 地图餐饮边界\n\n### case: 找附近美食\n用户: 附近有什么好吃的\ntarget: Agent(tag='餐饮服务')",
   "batch_id": "aabbccdd",
   "source_type": "generated"
 }
 ```
 
+如果草稿已经落盘，也可以传：
+
+```json
+{"draft_path": "output/dataset_draft.txt", "batch_id": "aabbccdd", "source_type": "generated"}
+```
+
 用途：把 dataset draft text v1 转成 canonical records。
 
 ### `product_data_validate_dataset_records`
@@ -387,6 +402,31 @@ skills/product-data/scripts/export_dataset_records.py
 
 用途：导出紧凑 JSONL/JSON。默认文件名是 `records.jsonl` 或 `records.json`。
 
+默认还会生成同目录 `records.csv` 表格，字段固定为：
+
+```text
+request_id,timestamp,query,prev_session,context,label,是否迁移Function,function
+```
+
+### `product_data_export_dataset_table`
+
+脚本：
+
+```text
+skills/product-data/scripts/export_dataset_table.py
+```
+
+输入支持：
+
+```json
+{
+  "records_path": "output/records.jsonl",
+  "output_dir": "output"
+}
+```
+
+用途：已有 canonical records 时，只补生成同事流转表格 `records.csv`，不改写元数据文件。主流程仍优先用 `product_data_export_dataset_records`，因为它会一次性导出元数据和表格。
+
 ## 当前可用工具
 
 - `read_file`：读取用户提供的产品定义、标签定义、样例 query 文件。
@@ -405,16 +445,16 @@ skills/product-data/scripts/export_dataset_records.py
 - `data_agent_show_generation_plan`：用户要求查看当前计划，或继续上下文时需要恢复计划详情时使用。
 - `data_agent_update_generation_plan`：用户对计划提出修改意见后使用，更新计划并重新展示。
 - `data_agent_confirm_generation_plan`：用户明确确认当前计划版本后使用，获取 `confirmed_plan_id`。
-- `data_agent_normalize_dataset_draft`：用户确认计划后，把 dataset draft text v1 转成 canonical records；必须传入 `confirmed_plan_id`。
-- `data_agent_validate_dataset_records`：对 canonical records 做结构、标签、时间戳和多轮上下文校验。
-- `data_agent_export_dataset_records`：校验 canonical records 并落盘；默认写紧凑 JSONL，一行一条，固定传 `output/records.jsonl`，不要再用 `write_file` 手写 records 文件。
+- `data_agent_normalize_dataset_draft`：用户确认计划后，把 dataset draft text v1 转成 canonical records；必须传入 `confirmed_plan_id`；支持 `draft_text` 或 `draft_path`，大草稿优先 `draft_path`。
+- `data_agent_validate_dataset_records`：对 canonical records 做结构、标签、时间戳和多轮上下文校验；支持 `records` 或 `records_path`。
+- `data_agent_export_dataset_records`：校验 canonical records 并落盘；支持 `records` 或 `records_path`；默认写紧凑 JSONL，一行一条，固定传 `output/records.jsonl`，并同时生成 `output/records.csv` 表格，不要再用 `write_file` 手写 records 或表格文件。
 
 ## 约束
 
 - 不要静默解决产品或标签歧义。
 - 如果 `ask_user_question` 不可用，使用普通回复向用户提问并停止，不要自己替用户确认。
 - canonical records 通过校验前，不要生成最终导出格式。
-- canonical records 需要落盘时，必须用 `data_agent_export_dataset_records`；不要自己拼接 JSON/JSONL；不要创建数据集子目录或自定义 records 文件名。
+- canonical records 需要落盘时，必须用 `data_agent_export_dataset_records`；不要自己拼接 JSON/JSONL/CSV；不要创建数据集子目录或自定义 records 文件名。
 - 如果使用 portable scripts，它们只负责格式转换、校验和导出，不替代 `generation_goal` / `generation_plan` 的用户 review。
 - 除非用户明确要求，否则不要把“修改标签定义”和“生成数据”混在一起做。
 - 不要因为用户说“生成一些数据”就跳过边界总结和 generation plan review。
diff --git a/skills/product-data/knowledge/canonical_record_v1.md b/skills/product-data/knowledge/canonical_record_v1.md
index b43f6b1..041f9f7 100644
--- a/skills/product-data/knowledge/canonical_record_v1.md
+++ b/skills/product-data/knowledge/canonical_record_v1.md
@@ -44,3 +44,14 @@ canonical record 是产品数据生成链路的中间元数据格式。它不是
 - 相邻轮时间间隔超过 5 分钟时给 warning。
 - `label.target` 必须有值。
 - `target_type` 只能是 `agent`、`function` 或 `unknown`。
+
+## 默认导出
+
+调用 `data_agent_export_dataset_records` 或 portable `export_dataset_records.py` 时，默认产物包括：
+
+- `records.jsonl`：canonical records，一行一条紧凑 JSON。
+- `records.csv`：同事流转表格，字段为 `request_id,timestamp,query,prev_session,context,label,是否迁移Function,function`。
+
+`records.csv` 中的 `prev_session` 和 `context` 是紧凑 JSON 字符串；`prev_session` 中的 `timestamp` 按历史表格习惯输出为字符串。
+
+如果已有 `records.jsonl` 或 `records.json`，只需要补表格，可以执行 portable `export_dataset_table.py` 或等价工具；默认仍输出到同目录的 `records.csv`。
diff --git a/skills/product-data/knowledge/dataset_draft_v1.md b/skills/product-data/knowledge/dataset_draft_v1.md
index b23ea57..a81ca7e 100644
--- a/skills/product-data/knowledge/dataset_draft_v1.md
+++ b/skills/product-data/knowledge/dataset_draft_v1.md
@@ -9,14 +9,14 @@
 
 ### case: case名称
 用户: 本轮 query
-target: Agent(tag="xxx")
+target: Agent(tag='xxx')
 notes: 可选，说明覆盖的问题或边界
 
 ### case: 多轮 case 名称
 用户: 前一轮 query
 小爱: 前一轮 tts
 用户: 本轮 query
-target: Agent(tag="xxx")
+target: Agent(tag='xxx')
 notes: 可选，说明覆盖的问题或边界
 ```
 
@@ -27,6 +27,7 @@ notes: 可选，说明覆盖的问题或边界
 - `小爱:` 表示小爱回复 tts。
 - 最后一个 `用户:` 是本轮 query。
 - `target:` 是本轮 query 对应的监督标签，必须存在。
+- Agent 标签推荐写单引号形式 `Agent(tag='xxx')`，转换工具会规范成 `Agent(tag="xxx")`。
 - 单轮数据只需要写一行 `用户:`，然后写 `target:`。
 - 多轮数据需要按照时间顺序写多组 `用户:` / `小爱:`。
 - 多轮数据的最后一轮只写 `用户:` 和 `target:`，不要写最后一轮 `小爱:`。
diff --git a/skills/product-data/schemas/export_dataset_records.input.schema.json b/skills/product-data/schemas/export_dataset_records.input.schema.json
index 8b7fa06..315eda7 100644
--- a/skills/product-data/schemas/export_dataset_records.input.schema.json
+++ b/skills/product-data/schemas/export_dataset_records.input.schema.json
@@ -11,7 +11,8 @@
       "default": "jsonl"
     },
     "require_validation_ok": {"type": "boolean", "default": true},
-    "overwrite": {"type": "boolean", "default": true}
+    "overwrite": {"type": "boolean", "default": true},
+    "export_table": {"type": "boolean", "default": true}
   },
   "anyOf": [
     {"required": ["records"]},
diff --git a/skills/product-data/schemas/export_dataset_records.output.schema.json b/skills/product-data/schemas/export_dataset_records.output.schema.json
index 7ddb6f3..0ca9a0d 100644
--- a/skills/product-data/schemas/export_dataset_records.output.schema.json
+++ b/skills/product-data/schemas/export_dataset_records.output.schema.json
@@ -4,8 +4,11 @@
     "ok": {"type": "boolean"},
     "output_path": {"type": "string"},
     "output_format": {"type": "string"},
+    "table_output_path": {"type": "string"},
+    "table_output_format": {"type": "string"},
     "record_count": {"type": "integer"},
     "bytes_written": {"type": "integer"},
+    "table_bytes_written": {"type": "integer"},
     "validation": {"type": "object"},
     "error": {"type": "string"}
   },
diff --git a/skills/product-data/schemas/export_dataset_table.input.schema.json b/skills/product-data/schemas/export_dataset_table.input.schema.json
new file mode 100644
index 0000000..1d0543c
--- /dev/null
+++ b/skills/product-data/schemas/export_dataset_table.input.schema.json
@@ -0,0 +1,15 @@
+{
+  "type": "object",
+  "properties": {
+    "records": {"type": "array"},
+    "records_path": {"type": "string"},
+    "output_path": {"type": "string"},
+    "output_dir": {"type": "string"},
+    "require_validation_ok": {"type": "boolean", "default": true},
+    "overwrite": {"type": "boolean", "default": true}
+  },
+  "anyOf": [
+    {"required": ["records"]},
+    {"required": ["records_path"]}
+  ]
+}
diff --git a/skills/product-data/schemas/export_dataset_table.output.schema.json b/skills/product-data/schemas/export_dataset_table.output.schema.json
new file mode 100644
index 0000000..7ddb6f3
--- /dev/null
+++ b/skills/product-data/schemas/export_dataset_table.output.schema.json
@@ -0,0 +1,13 @@
+{
+  "type": "object",
+  "properties": {
+    "ok": {"type": "boolean"},
+    "output_path": {"type": "string"},
+    "output_format": {"type": "string"},
+    "record_count": {"type": "integer"},
+    "bytes_written": {"type": "integer"},
+    "validation": {"type": "object"},
+    "error": {"type": "string"}
+  },
+  "required": ["ok"]
+}
diff --git a/skills/product-data/scripts/export_dataset_records.py b/skills/product-data/scripts/export_dataset_records.py
index 0d970b4..e525207 100644
--- a/skills/product-data/scripts/export_dataset_records.py
+++ b/skills/product-data/scripts/export_dataset_records.py
@@ -21,6 +21,7 @@ def main() -> int:
             output_format=output_format,
             require_validation_ok=bool(payload.get("require_validation_ok", True)),
             overwrite=bool(payload.get("overwrite", True)),
+            export_table=bool(payload.get("export_table", True)),
         )
         emit_success(result)
         return 0
diff --git a/skills/product-data/scripts/export_dataset_table.py b/skills/product-data/scripts/export_dataset_table.py
new file mode 100644
index 0000000..1e0f783
--- /dev/null
+++ b/skills/product-data/scripts/export_dataset_table.py
@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from product_data_portable import (
+    default_table_output_path,
+    emit_error,
+    emit_success,
+    export_dataset_table,
+    load_json_payload,
+    load_records,
+)
+
+
+def main() -> int:
+    try:
+        payload = load_json_payload()
+        records = load_records(payload)
+        result = export_dataset_table(
+            records,
+            output_path=default_table_output_path(payload),
+            require_validation_ok=bool(payload.get("require_validation_ok", True)),
+            overwrite=bool(payload.get("overwrite", True)),
+        )
+        emit_success(result)
+        return 0
+    except Exception as exc:  # noqa: BLE001 - CLI 需要把错误稳定转成 JSON
+        emit_error(exc)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/skills/product-data/scripts/product_data_portable.py b/skills/product-data/scripts/product_data_portable.py
index 3311ba5..b5a1cf7 100644
--- a/skills/product-data/scripts/product_data_portable.py
+++ b/skills/product-data/scripts/product_data_portable.py
@@ -7,10 +7,12 @@ from __future__ import annotations
 """
 
 import argparse
+import csv
 import json
 import re
 import sys
 import time
+from io import StringIO
 from pathlib import Path
 from typing import Any
 
@@ -88,6 +90,8 @@ def read_records_file(path: str) -> list[dict[str, Any]]:
             records.append(item)
         return records
     decoded = json.loads(text)
+    if isinstance(decoded, dict) and isinstance(decoded.get("records"), list):
+        decoded = decoded["records"]
     if not isinstance(decoded, list):
         raise ProductDataError(f"{path} must contain a JSON array")
     return decoded
@@ -374,6 +378,7 @@ def export_dataset_records(
     output_format: str = "jsonl",
     require_validation_ok: bool = True,
     overwrite: bool = True,
+    export_table: bool = True,
 ) -> dict[str, Any]:
     if output_format not in {"jsonl", "json"}:
         raise ProductDataError("output_format must be jsonl or json")
@@ -382,8 +387,11 @@ def export_dataset_records(
         raise ProductDataError(f"records failed validation with {validation['error_count']} errors")
 
     path = Path(output_path).expanduser()
+    table_path = path.with_name("records.csv")
     if path.exists() and not overwrite:
         raise ProductDataError(f"output_path already exists: {output_path}")
+    if export_table and table_path.exists() and not overwrite:
+        raise ProductDataError("table output_path already exists: records.csv")
     path.parent.mkdir(parents=True, exist_ok=True)
 
     if output_format == "jsonl":
@@ -394,13 +402,106 @@ def export_dataset_records(
     else:
         content = json.dumps(records, ensure_ascii=False, separators=(",", ":")) + "\n"
     path.write_text(content, encoding="utf-8")
-    return {
+    payload = {
         "output_path": str(path),
         "output_format": output_format,
         "record_count": len(records),
         "bytes_written": len(content.encode("utf-8")),
         "validation": validation,
     }
+    if export_table:
+        table_content = render_dataset_records_table_csv(records)
+        table_path.write_text(table_content, encoding="utf-8-sig")
+        payload.update(
+            {
+                "table_output_path": str(table_path),
+                "table_output_format": "csv",
+                "table_bytes_written": len(table_content.encode("utf-8-sig")),
+            }
+        )
+    return payload
+
+
+def export_dataset_table(
+    records: list[dict[str, Any]],
+    *,
+    output_path: str,
+    require_validation_ok: bool = True,
+    overwrite: bool = True,
+) -> dict[str, Any]:
+    validation = validate_dataset_records(records)
+    if require_validation_ok and not validation["ok"]:
+        raise ProductDataError(f"records failed validation with {validation['error_count']} errors")
+    path = Path(output_path).expanduser()
+    if path.exists() and not overwrite:
+        raise ProductDataError(f"output_path already exists: {output_path}")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    content = render_dataset_records_table_csv(records)
+    path.write_text(content, encoding="utf-8-sig")
+    return {
+        "output_path": str(path),
+        "output_format": "csv",
+        "record_count": len(records),
+        "bytes_written": len(content.encode("utf-8-sig")),
+        "validation": validation,
+    }
+
+
+def render_dataset_records_table_csv(records: list[dict[str, Any]]) -> str:
+    output = StringIO()
+    writer = csv.DictWriter(output, fieldnames=dataset_table_columns(), lineterminator="\n")
+    writer.writeheader()
+    for record in records:
+        writer.writerow(dataset_record_table_row(record))
+    return output.getvalue()
+
+
+def dataset_table_columns() -> list[str]:
+    return [
+        "request_id",
+        "timestamp",
+        "query",
+        "prev_session",
+        "context",
+        "label",
+        "是否迁移Function",
+        "function",
+    ]
+
+
+def dataset_record_table_row(record: dict[str, Any]) -> dict[str, str]:
+    source = record.get("source") if isinstance(record.get("source"), dict) else {}
+    turn = record.get("turn") if isinstance(record.get("turn"), dict) else {}
+    label = record.get("label") if isinstance(record.get("label"), dict) else {}
+    context = record.get("context") if isinstance(record.get("context"), dict) else {}
+    prev_session = record.get("prev_session")
+    if not isinstance(prev_session, list):
+        prev_session = []
+    return {
+        "request_id": str(source.get("request_id") or ""),
+        "timestamp": str(source.get("timestamp") or turn.get("timestamp") or ""),
+        "query": str(turn.get("query") or ""),
+        "prev_session": json.dumps(table_prev_session(prev_session), ensure_ascii=False, separators=(",", ":")),
+        "context": json.dumps(context, ensure_ascii=False, separators=(",", ":")),
+        "label": str(label.get("dataset_label") or ""),
+        "是否迁移Function": "",
+        "function": str(label.get("target") or ""),
+    }
+
+
+def table_prev_session(prev_session: list[Any]) -> list[dict[str, str]]:
+    rows: list[dict[str, str]] = []
+    for item in prev_session[-10:]:
+        if not isinstance(item, dict):
+            continue
+        rows.append(
+            {
+                "query": str(item.get("query") or ""),
+                "tts": str(item.get("tts") or ""),
+                "timestamp": str(item.get("timestamp") or ""),
+            }
+        )
+    return rows
 
 
 def default_output_path(payload: dict[str, Any]) -> str:
@@ -413,6 +514,16 @@ def default_output_path(payload: dict[str, Any]) -> str:
     return str(Path("output") / filename)
 
 
+def default_table_output_path(payload: dict[str, Any]) -> str:
+    if isinstance(payload.get("output_path"), str) and payload["output_path"].strip():
+        return payload["output_path"].strip()
+    if isinstance(payload.get("output_dir"), str) and payload["output_dir"].strip():
+        return str(Path(payload["output_dir"]).expanduser() / "records.csv")
+    if isinstance(payload.get("records_path"), str) and payload["records_path"].strip():
+        return str(Path(payload["records_path"]).expanduser().with_name("records.csv"))
+    return str(Path("output") / "records.csv")
+
+
 def target_type(target: str) -> str:
     if re.match(r"^Agent\s*\(\s*tag\s*=", target):
         return "agent"
diff --git a/skills/product-data/tools.yaml b/skills/product-data/tools.yaml
index 8e17632..ff458f5 100644
--- a/skills/product-data/tools.yaml
+++ b/skills/product-data/tools.yaml
@@ -23,10 +23,18 @@ tools:
       command: python skills/product-data/scripts/validate_dataset_records.py --input <input.json>
       stdin: true
   - name: product_data_export_dataset_records
-    description: Validate canonical records and write compact JSONL or JSON files without asking the model to hand-write JSON.
+    description: Validate canonical records and write compact JSONL or JSON files plus a sibling records.csv sharing table without asking the model to hand-write files.
     script: scripts/export_dataset_records.py
     input_schema: schemas/export_dataset_records.input.schema.json
     output_schema: schemas/export_dataset_records.output.schema.json
     fallback:
       command: python skills/product-data/scripts/export_dataset_records.py --input <input.json>
       stdin: true
+  - name: product_data_export_dataset_table
+    description: Convert existing canonical records into the shared records.csv table format without rewriting the metadata file.
+    script: scripts/export_dataset_table.py
+    input_schema: schemas/export_dataset_table.input.schema.json
+    output_schema: schemas/export_dataset_table.output.schema.json
+    fallback:
+      command: python skills/product-data/scripts/export_dataset_table.py --input <input.json>
+      stdin: true
diff --git a/src/agent_runtime.py b/src/agent_runtime.py
index e5bcca3..350ccd8 100644
--- a/src/agent_runtime.py
+++ b/src/agent_runtime.py
@@ -832,8 +832,15 @@ class LocalCodingAgent:
                             return result
                         # fall through to the normal tool-call branch below
                 # normal error path if not recovered
+                final_output = str(exc)
+                session.append_assistant(
+                    final_output,
+                    message_id=f'assistant_{len(session.messages)}',
+                    stop_reason='backend_error',
+                )
+                _append_final_text_stream_events(stream_events, final_output)
                 result = AgentRunResult(
-                    final_output=str(exc),
+                    final_output=final_output,
                     turns=max(turn_index - 1, 0),
                     tool_calls=tool_calls,
                     transcript=session.transcript(),
@@ -1535,15 +1542,42 @@ class LocalCodingAgent:
             usage=usage,
         )
         assistant_message = session.messages[assistant_index]
+        try:
+            parsed_tool_calls = self._tool_calls_from_message(assistant_message.tool_calls)
+        except OpenAICompatError:
+            self._scrub_malformed_assistant_tool_calls(session, assistant_index)
+            raise
         turn = AssistantTurn(
             content=assistant_message.content,
-            tool_calls=self._tool_calls_from_message(assistant_message.tool_calls),
+            tool_calls=parsed_tool_calls,
             finish_reason=finish_reason,
             raw_message=assistant_message.to_openai_message(),
             usage=usage,
         )
         return turn, tuple(events)
 
+    def _scrub_malformed_assistant_tool_calls(
+        self,
+        session: AgentSessionState,
+        assistant_index: int,
+    ) -> None:
+        message = session.messages[assistant_index]
+        filtered_blocks = tuple(
+            block
+            for block in message.blocks
+            if block.get('type') not in {'tool_use', 'tool_call'}
+        )
+        session.messages[assistant_index] = replace(
+            message,
+            tool_calls=(),
+            blocks=filtered_blocks,
+            stop_reason='malformed_tool_arguments',
+            metadata={
+                **message.metadata,
+                'malformed_tool_arguments': True,
+            },
+        )
+
     def _tool_calls_from_message(
         self,
         tool_calls: tuple[dict[str, object], ...],
@@ -1558,7 +1592,18 @@ class LocalCodingAgent:
                 continue
             raw_arguments = function_block.get('arguments', '')
             if isinstance(raw_arguments, str) and raw_arguments.strip():
-                arguments = json.loads(raw_arguments)
+                try:
+                    arguments = json.loads(raw_arguments)
+                except json.JSONDecodeError as exc:
+                    preview = raw_arguments[:240].replace('\n', '\\n')
+                    raise OpenAICompatError(
+                        '模型返回的工具参数不是合法 JSON，已停止本轮以避免执行错误工具。'
+                        f'工具：{name}；位置：line {exc.lineno} column {exc.colno}；'
+                        f'原因：{exc.msg}；参数片段：{preview!r}。'
+                        '请回复“继续”，我会要求模型重新生成合法 JSON 参数；'
+                        '对于大段 dataset draft，应分批或使用 draft_path/records_path，'
+                        '不要把大量带引号的文本一次性塞进工具参数。'
+                    ) from exc
                 if not isinstance(arguments, dict):
                     raise OpenAICompatError(
                         f'Tool arguments must decode to an object, got {type(arguments).__name__}'
diff --git a/src/agent_tool_specs/data_agent.py b/src/agent_tool_specs/data_agent.py
index d31c594..590150d 100644
--- a/src/agent_tool_specs/data_agent.py
+++ b/src/agent_tool_specs/data_agent.py
@@ -429,7 +429,8 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo
             description=(
                 'Parse dataset draft text written with 用户/小爱/target blocks and build canonical '
                 'data-agent records with generated record IDs, timestamps, source metadata, context, and labels. '
-                'Generated data requires a confirmed_plan_id from data_agent_confirm_generation_plan.'
+                'Generated data requires a confirmed_plan_id from data_agent_confirm_generation_plan. '
+                'Use draft_path instead of draft_text when the draft is large or contains many quoted targets.'
             ),
             parameters={
                 'type': 'object',
@@ -438,6 +439,10 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo
                         'type': 'string',
                         'description': 'Dataset draft text in dataset draft text v1 format.',
                     },
+                    'draft_path': {
+                        'type': 'string',
+                        'description': 'Path to a UTF-8 dataset draft text file. Prefer this for large drafts.',
+                    },
                     'batch_id': {
                         'type': 'string',
                         'description': 'Batch identifier used in generated record_id values.',
@@ -464,7 +469,10 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo
                         'description': 'Required for generated data; returned by data_agent_confirm_generation_plan.',
                     },
                 },
-                'required': ['draft_text'],
+                'anyOf': [
+                    {'required': ['draft_text']},
+                    {'required': ['draft_path']},
+                ],
             },
             handler=resolve_handler(handlers, 'data_agent_normalize_dataset_draft', 'data-agent'),
         ),
@@ -484,8 +492,15 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo
                         ],
                         'description': 'Canonical records as an array, or a JSON string containing the array.',
                     },
+                    'records_path': {
+                        'type': 'string',
+                        'description': 'Path to records JSON/JSONL. Prefer this when records are large.',
+                    },
                 },
-                'required': ['records'],
+                'anyOf': [
+                    {'required': ['records']},
+                    {'required': ['records_path']},
+                ],
             },
             handler=resolve_handler(handlers, 'data_agent_validate_dataset_records', 'data-agent'),
         ),
@@ -493,7 +508,8 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo
             name='data_agent_export_dataset_records',
             description=(
                 'Validate canonical data-agent records and write them to a workspace file. '
-                'Defaults to compact JSONL, one canonical record per line, so the model does not hand-write JSON output.'
+                'Defaults to compact JSONL, one canonical record per line, and also writes records.csv '
+                'in the same output directory for human sharing.'
             ),
             parameters={
                 'type': 'object',
@@ -505,6 +521,10 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo
                         ],
                         'description': 'Canonical records as an array, or a JSON string containing the array.',
                     },
+                    'records_path': {
+                        'type': 'string',
+                        'description': 'Path to records JSON/JSONL. Prefer this when records are large.',
+                    },
                     'output_path': {
                         'type': 'string',
                         'description': (
@@ -526,8 +546,16 @@ def build_data_agent_tools(handlers: Mapping[str, ToolHandler]) -> list[AgentToo
                         'type': 'boolean',
                         'description': 'When false, refuse to overwrite an existing file.',
                     },
+                    'export_table': {
+                        'type': 'boolean',
+                        'description': 'Defaults to true. Also write records.csv with the shared spreadsheet columns.',
+                    },
                 },
-                'required': ['records', 'output_path'],
+                'required': ['output_path'],
+                'anyOf': [
+                    {'required': ['records']},
+                    {'required': ['records_path']},
+                ],
             },
             handler=resolve_handler(handlers, 'data_agent_export_dataset_records', 'data-agent'),
         ),
diff --git a/src/agent_tools.py b/src/agent_tools.py
index 1f65992..450f187 100644
--- a/src/agent_tools.py
+++ b/src/agent_tools.py
@@ -1474,7 +1474,7 @@ def _data_agent_confirm_generation_plan_tool(arguments: dict[str, Any], context:
 
 
 def _normalize_dataset_draft_tool(arguments: dict[str, Any], _context: ToolExecutionContext) -> str:
-    draft_text = _require_string(arguments, 'draft_text')
+    draft_text = _draft_text_from_arguments(arguments, _context)
     batch_id = arguments.get('batch_id', 'aabbccdd')
     if not isinstance(batch_id, str) or not batch_id.strip():
         raise ToolExecutionError('batch_id must be a non-empty string')
@@ -1508,10 +1508,8 @@ def _normalize_dataset_draft_tool(arguments: dict[str, Any], _context: ToolExecu
 
 
 def _validate_dataset_records_tool(arguments: dict[str, Any], _context: ToolExecutionContext) -> str:
-    if 'records' not in arguments:
-        raise ToolExecutionError('records is required')
     try:
-        records = records_from_tool_argument(arguments['records'])
+        records = _records_from_arguments(arguments, _context)
         payload = validate_dataset_records(records)
     except (DataRecordError, json.JSONDecodeError) as exc:
         raise ToolExecutionError(str(exc)) from exc
@@ -1519,8 +1517,6 @@ def _validate_dataset_records_tool(arguments: dict[str, Any], _context: ToolExec
 
 
 def _export_dataset_records_tool(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
-    if 'records' not in arguments:
-        raise ToolExecutionError('records is required')
     output_format = arguments.get('output_format', 'jsonl')
     if not isinstance(output_format, str):
         raise ToolExecutionError('output_format must be a string')
@@ -1530,8 +1526,11 @@ def _export_dataset_records_tool(arguments: dict[str, Any], context: ToolExecuti
     overwrite = arguments.get('overwrite', True)
     if not isinstance(overwrite, bool):
         raise ToolExecutionError('overwrite must be a boolean')
+    export_table = arguments.get('export_table', True)
+    if not isinstance(export_table, bool):
+        raise ToolExecutionError('export_table must be a boolean')
     try:
-        records = records_from_tool_argument(arguments['records'])
+        records = _records_from_arguments(arguments, context)
         canonical_filename = 'records.json' if output_format == 'json' else 'records.jsonl'
         payload = export_dataset_records(
             records,
@@ -1544,12 +1543,67 @@ def _export_dataset_records_tool(arguments: dict[str, Any], context: ToolExecuti
             output_format=output_format,
             require_validation_ok=require_validation_ok,
             overwrite=overwrite,
+            export_table=export_table,
         )
     except (DataRecordError, json.JSONDecodeError) as exc:
         raise ToolExecutionError(str(exc)) from exc
     return json.dumps(payload, ensure_ascii=False, indent=2)
 
 
+def _draft_text_from_arguments(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
+    draft_text = arguments.get('draft_text')
+    draft_path = arguments.get('draft_path')
+    has_text = isinstance(draft_text, str) and bool(draft_text.strip())
+    has_path = isinstance(draft_path, str) and bool(draft_path.strip())
+    if has_text and has_path:
+        raise ToolExecutionError('provide only one of draft_text or draft_path')
+    if has_text:
+        return draft_text
+    if not has_path:
+        raise ToolExecutionError('draft_text or draft_path is required')
+    path = _resolve_path(draft_path, context, allow_missing=False, allow_outside_root=True)
+    if not path.is_file():
+        raise ToolExecutionError(f'draft_path is not a file: {path}')
+    return path.read_text(encoding='utf-8', errors='replace')
+
+
+def _records_from_arguments(arguments: dict[str, Any], context: ToolExecutionContext) -> list[dict[str, Any]]:
+    has_records = 'records' in arguments and arguments.get('records') is not None
+    records_path = arguments.get('records_path')
+    has_path = isinstance(records_path, str) and bool(records_path.strip())
+    if has_records and has_path:
+        raise ToolExecutionError('provide only one of records or records_path')
+    if has_records:
+        return records_from_tool_argument(arguments['records'])
+    if not has_path:
+        raise ToolExecutionError('records or records_path is required')
+    path = _resolve_path(records_path, context, allow_missing=False, allow_outside_root=True)
+    if not path.is_file():
+        raise ToolExecutionError(f'records_path is not a file: {path}')
+    return _records_from_file(path)
+
+
+def _records_from_file(path: Path) -> list[dict[str, Any]]:
+    text = path.read_text(encoding='utf-8', errors='replace').strip()
+    if not text:
+        raise DataRecordError('records_path is empty')
+    if path.suffix == '.jsonl':
+        records: list[dict[str, Any]] = []
+        for line_number, line in enumerate(text.splitlines(), start=1):
+            stripped = line.strip()
+            if not stripped:
+                continue
+            item = json.loads(stripped)
+            if not isinstance(item, dict):
+                raise DataRecordError(f'records_path line {line_number} must decode to an object')
+            records.append(item)
+        return records_from_tool_argument(records)
+    payload = json.loads(text)
+    if isinstance(payload, dict) and isinstance(payload.get('records'), list):
+        payload = payload['records']
+    return records_from_tool_argument(payload)
+
+
 def _data_agent_output_path(
     output_path: str,
     context: ToolExecutionContext,
diff --git a/src/data_agent_records.py b/src/data_agent_records.py
index 86d6fa8..48c8369 100644
--- a/src/data_agent_records.py
+++ b/src/data_agent_records.py
@@ -7,9 +7,11 @@ Skill 应该先让模型生成紧凑、便于人工 review 的 dataset draft tex
 再由这里的逻辑解析并补齐为 canonical records，供后续校验和导出使用。
 """
 
+import csv
 import json
 import re
 import time
+from io import StringIO
 from pathlib import Path
 from typing import Any
 
@@ -461,6 +463,7 @@ def export_dataset_records(
     output_format: str = 'jsonl',
     require_validation_ok: bool = True,
     overwrite: bool = True,
+    export_table: bool = True,
 ) -> dict[str, Any]:
     """校验并导出 canonical records。
 
@@ -480,8 +483,11 @@ def export_dataset_records(
         )
 
     path = _resolve_output_path(root, output_path)
+    table_path = path.with_name('records.csv')
     if path.exists() and not overwrite:
         raise DataRecordError(f'output_path already exists: {output_path}')
+    if export_table and table_path.exists() and not overwrite:
+        raise DataRecordError('table output_path already exists: records.csv')
     path.parent.mkdir(parents=True, exist_ok=True)
 
     if output_format == 'jsonl':
@@ -494,13 +500,83 @@ def export_dataset_records(
 
     path.write_text(content, encoding='utf-8')
     rel_path = str(path.relative_to(Path(root).resolve()))
-    return {
+    payload = {
         'output_path': rel_path,
         'output_format': output_format,
         'record_count': len(records),
         'bytes_written': len(content.encode('utf-8')),
         'validation': validation,
     }
+    if export_table:
+        table_content = render_dataset_records_table_csv(records)
+        table_path.write_text(table_content, encoding='utf-8-sig')
+        payload.update(
+            {
+                'table_output_path': str(table_path.relative_to(Path(root).resolve())),
+                'table_output_format': 'csv',
+                'table_bytes_written': len(table_content.encode('utf-8-sig')),
+            }
+        )
+    return payload
+
+
+def render_dataset_records_table_csv(records: list[dict[str, Any]]) -> str:
+    """把 canonical records 转成同事流转用的表格 CSV。"""
+
+    output = StringIO()
+    writer = csv.DictWriter(output, fieldnames=_dataset_table_columns(), lineterminator='\n')
+    writer.writeheader()
+    for record in records:
+        writer.writerow(_dataset_record_table_row(record))
+    return output.getvalue()
+
+
+def _dataset_table_columns() -> list[str]:
+    return [
+        'request_id',
+        'timestamp',
+        'query',
+        'prev_session',
+        'context',
+        'label',
+        '是否迁移Function',
+        'function',
+    ]
+
+
+def _dataset_record_table_row(record: dict[str, Any]) -> dict[str, str]:
+    source = record.get('source') if isinstance(record.get('source'), dict) else {}
+    turn = record.get('turn') if isinstance(record.get('turn'), dict) else {}
+    label = record.get('label') if isinstance(record.get('label'), dict) else {}
+    context = record.get('context') if isinstance(record.get('context'), dict) else {}
+    prev_session = record.get('prev_session')
+    if not isinstance(prev_session, list):
+        prev_session = []
+    return {
+        'request_id': str(source.get('request_id') or ''),
+        'timestamp': str(source.get('timestamp') or turn.get('timestamp') or ''),
+        'query': str(turn.get('query') or ''),
+        'prev_session': json.dumps(_table_prev_session(prev_session), ensure_ascii=False, separators=(',', ':')),
+        'context': json.dumps(context, ensure_ascii=False, separators=(',', ':')),
+        'label': str(label.get('dataset_label') or ''),
+        '是否迁移Function': '',
+        'function': str(label.get('target') or ''),
+    }
+
+
+def _table_prev_session(prev_session: list[Any]) -> list[dict[str, str]]:
+    rows: list[dict[str, str]] = []
+    for item in prev_session[-10:]:
+        if not isinstance(item, dict):
+            continue
+        rows.append(
+            {
+                'query': str(item.get('query') or ''),
+                'tts': str(item.get('tts') or ''),
+                'timestamp': str(item.get('timestamp') or ''),
+            }
+        )
+    return rows
 
 
 def _parse_draft_text(draft_text: str) -> dict[str, Any]:
@@ -797,7 +873,7 @@ def _case_to_record(
     if not current_query:
         raise DataRecordError(f'case {index} current 用户 line must be non-empty')
 
-    target = str(case.get('target') or '').strip()
+    target = _canonical_target(str(case.get('target') or '').strip())
     if not target:
         raise DataRecordError(f'case {index} target is required')
 
@@ -866,6 +942,17 @@ def _target_type(target: str) -> str:
     return 'unknown'
 
 
+def _canonical_target(target: str) -> str:
+    """规范常见 Agent 标签写法，允许草稿里用单引号降低 JSON 转义风险。"""
+
+    match = re.fullmatch(r'Agent\s*\(\s*tag\s*=\s*([\'"])(.+?)\1\s*\)', target)
+    if not match:
+        return target
+    tag = match.group(2).strip()
+    escaped = tag.replace('\\', '\\\\').replace('"', '\\"')
+    return f'Agent(tag="{escaped}")'
+
+
 def _record_id(*, source_type: str, batch_id: str, request_id: str, index: int) -> str:
     if source_type == 'generated':
         return f'gen_{_safe_token(batch_id)}_{index:06d}'
diff --git a/src/openai_compat.py b/src/openai_compat.py
index 32cfeb1..2efbaf8 100644
--- a/src/openai_compat.py
+++ b/src/openai_compat.py
@@ -75,8 +75,11 @@ def _parse_tool_arguments(raw_arguments: Any) -> dict[str, Any]:
         try:
             parsed = json.loads(raw_arguments)
         except json.JSONDecodeError as exc:
+            preview = raw_arguments[:240].replace('\n', '\\n')
             raise OpenAICompatError(
-                f'Invalid tool arguments returned by model: {raw_arguments!r}'
+                'Invalid tool arguments returned by model: '
+                f'line {exc.lineno} column {exc.colno}: {exc.msg}; '
+                f'preview={preview!r}'
             ) from exc
         if not isinstance(parsed, dict):
             raise OpenAICompatError(
diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py
index 28dab35..7a49f29 100644
--- a/tests/test_agent_runtime.py
+++ b/tests/test_agent_runtime.py
@@ -769,6 +769,64 @@ class AgentRuntimeTests(unittest.TestCase):
         self.assertGreaterEqual(mutation_totals.get('assistant_tool_call_delta', 0), 2)
         self.assertEqual(mutation_totals.get('assistant_finalize', 0), 1)
 
+    def test_agent_reports_malformed_streaming_tool_arguments(self) -> None:
+        responses = [
+            [
+                {
+                    'choices': [
+                        {
+                            'delta': {
+                                'tool_calls': [
+                                    {
+                                        'index': 0,
+                                        'id': 'call_1',
+                                        'function': {
+                                            'name': 'data_agent_normalize_dataset_draft',
+                                            'arguments': '{"draft_text": "target: Agent(tag="地图导航")"}',
+                                        },
+                                    }
+                                ]
+                            },
+                            'finish_reason': None,
+                        }
+                    ]
+                },
+                {
+                    'choices': [{'delta': {}, 'finish_reason': 'tool_calls'}],
+                    'usage': {'prompt_tokens': 9, 'completion_tokens': 4},
+                },
+            ]
+        ]
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            workspace = Path(tmp_dir)
+            with patch(
+                'src.openai_compat.request.urlopen',
+                side_effect=make_streaming_urlopen_side_effect(responses),
+            ):
+                agent = LocalCodingAgent(
+                    model_config=ModelConfig(
+                        model='Qwen/Qwen3-Coder-30B-A3B-Instruct',
+                        base_url='http://127.0.0.1:8000/v1',
+                    ),
+                    runtime_config=AgentRuntimeConfig(
+                        cwd=workspace,
+                        stream_model_responses=True,
+                    ),
+                )
+                result = agent.run('Generate records')
+
+        self.assertEqual(result.stop_reason, 'backend_error')
+        self.assertIn('模型返回的工具参数不是合法 JSON', result.final_output)
+        self.assertIn('data_agent_normalize_dataset_draft', result.final_output)
+        malformed = [
+            message
+            for message in result.transcript
+            if message.get('stop_reason') == 'malformed_tool_arguments'
+        ]
+        self.assertEqual(len(malformed), 1)
+        self.assertNotIn('tool_calls', malformed[0])
+        self.assertEqual(result.transcript[-1]['content'], result.final_output)
+
     def test_transcript_entries_include_structured_blocks(self) -> None:
         responses = [
             {
diff --git a/tests/test_data_agent_records.py b/tests/test_data_agent_records.py
index db1b91f..4982681 100644
--- a/tests/test_data_agent_records.py
+++ b/tests/test_data_agent_records.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import csv
 import json
 import tempfile
 import unittest
@@ -15,6 +16,7 @@ from src.data_agent_records import (
     normalize_dataset_draft,
     prepare_generation_goal,
     prepare_generation_plan,
+    render_dataset_records_table_csv,
     update_generation_plan,
     validate_dataset_records,
 )
@@ -64,6 +66,22 @@ notes: 多轮承接附近生活服务查询
             ],
         )
 
+    def test_normalize_dataset_draft_canonicalizes_single_quote_agent_target(self) -> None:
+        payload = normalize_dataset_draft(
+            '''
+# dataset_label: 地图导航边界数据
+### case: 顺路停车场
+用户: 帮我找个顺路的停车场
+target: Agent(tag='地图导航')
+'''.strip(),
+            batch_id='demo',
+            base_timestamp=1_755_567_930_500,
+        )
+
+        record = payload['records'][0]
+        self.assertEqual(record['label']['target'], 'Agent(tag="地图导航")')
+        self.assertEqual(record['label']['target_type'], 'agent')
+
     def test_validate_dataset_records_reports_valid_payload(self) -> None:
         records = normalize_dataset_draft(
             '''
@@ -116,12 +134,50 @@ target: Agent(tag="life_service")
             )
             output_path = Path(tmp_dir) / payload['output_path']
             lines = output_path.read_text(encoding='utf-8').splitlines()
+            table_path = Path(tmp_dir) / payload['table_output_path']
+            table_rows = list(csv.DictReader(table_path.read_text(encoding='utf-8-sig').splitlines()))
 
         self.assertEqual(payload['output_format'], 'jsonl')
+        self.assertEqual(payload['table_output_format'], 'csv')
         self.assertEqual(payload['record_count'], 1)
         self.assertEqual(len(lines), 1)
         self.assertNotIn(': ', lines[0])
         self.assertEqual(json.loads(lines[0])['turn']['query'], '附近有什么好吃的')
+        self.assertEqual(len(table_rows), 1)
+        self.assertEqual(
+            list(table_rows[0].keys()),
+            ['request_id', 'timestamp', 'query', 'prev_session', 'context', 'label', '是否迁移Function', 'function'],
+        )
+        self.assertEqual(table_rows[0]['request_id'], 'aabbccdd')
+        self.assertEqual(table_rows[0]['query'], '附近有什么好吃的')
+        self.assertEqual(table_rows[0]['prev_session'], '[]')
+        self.assertEqual(table_rows[0]['context'], '{}')
+        self.assertEqual(table_rows[0]['label'], '地图和生活边界数据')
+        self.assertEqual(table_rows[0]['是否迁移Function'], '')
+        self.assertEqual(table_rows[0]['function'], 'Agent(tag="life_service")')
+
+    def test_render_dataset_records_table_csv_keeps_prev_session_json(self) -> None:
+        records = normalize_dataset_draft(
+            '''
+# dataset_label: 地图导航边界数据
+### case: 多轮顺路停车场
+用户: 查一下附近停车场
+小爱: 找到了附近停车场
+用户: 帮我找个最顺路的
+target: Agent(tag='地图导航')
+'''.strip(),
+            batch_id='demo',
+            base_timestamp=1_755_567_930_500,
+            timestamp_step_ms=60_000,
+        )['records']
+
+        rows = list(csv.DictReader(render_dataset_records_table_csv(records).splitlines()))
+
+        prev_session = json.loads(rows[0]['prev_session'])
+        self.assertEqual(prev_session[0]['query'], '查一下附近停车场')
+        self.assertEqual(prev_session[0]['tts'], '找到了附近停车场')
+        self.assertEqual(prev_session[0]['timestamp'], '1755567870500')
+        self.assertEqual(rows[0]['function'], 'Agent(tag="地图导航")')
 
     def test_normalize_online_draft_uses_real_request_metadata(self) -> None:
         records = normalize_dataset_draft(
@@ -532,6 +588,65 @@ target: Agent(tag="life_service")
         self.assertEqual(export_payload['record_count'], 1)
         self.assertEqual(len(exported_lines), 1)
 
+    def test_data_agent_tools_accept_draft_and_records_paths(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            context = build_tool_context(AgentRuntimeConfig(cwd=root))
+            draft_path = root / 'draft.txt'
+            draft_path.write_text(
+                '''
+# dataset_label: 地图导航边界数据
+### case: 顺路停车场
+用户: 帮我找个顺路的停车场
+target: Agent(tag='地图导航')
+'''.strip(),
+                encoding='utf-8',
+            )
+
+            normalize_result = execute_tool(
+                default_tool_registry(),
+                'data_agent_normalize_dataset_draft',
+                {
+                    'draft_path': str(draft_path),
+                    'batch_id': 'demo',
+                    'source_type': 'manual',
+                    'base_timestamp': 1_755_567_930_500,
+                },
+                context,
+            )
+            self.assertTrue(normalize_result.ok, normalize_result.content)
+            records = json.loads(normalize_result.content)['records']
+            self.assertEqual(records[0]['label']['target'], 'Agent(tag="地图导航")')
+
+            records_path = root / 'records.jsonl'
+            records_path.write_text(
+                '\n'.join(json.dumps(record, ensure_ascii=False) for record in records),
+                encoding='utf-8',
+            )
+            validate_result = execute_tool(
+                default_tool_registry(),
+                'data_agent_validate_dataset_records',
+                {'records_path': str(records_path)},
+                context,
+            )
+            export_result = execute_tool(
+                default_tool_registry(),
+                'data_agent_export_dataset_records',
+                {
+                    'records_path': str(records_path),
+                    'output_path': 'output/records.jsonl',
+                },
+                context,
+            )
+            self.assertTrue(export_result.ok, export_result.content)
+            export_payload = json.loads(export_result.content)
+            table_exists = (root / export_payload['table_output_path']).exists()
+
+        self.assertTrue(validate_result.ok, validate_result.content)
+        self.assertTrue(json.loads(validate_result.content)['ok'])
+        self.assertEqual(export_payload['record_count'], 1)
+        self.assertTrue(table_exists)
+
     def test_data_agent_tools_route_relative_outputs_to_session_output(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             root = Path(tmp_dir)
@@ -578,8 +693,10 @@ target: Agent(tag="life_service")
             export_payload = json.loads(export_result.content)
 
         expected = '.port_sessions/accounts/alice/sessions/s1/output/records.jsonl'
+        expected_table = '.port_sessions/accounts/alice/sessions/s1/output/records.csv'
         self.assertEqual(plan_payload['plan']['output_path'], expected)
         self.assertEqual(export_payload['output_path'], expected)
+        self.assertEqual(export_payload['table_output_path'], expected_table)
 
     def test_data_agent_session_records_path_stays_stable_when_file_exists(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir: