--- name: guanwf description: 当用户要创建、编辑、保存工作流引擎中的工作流（含数据流节点、Python 节点、子工作流编排、参数赋值等多节点 DAG），或需要预览数据流节点、校验/试运行 Python 脚本、运行工作流、配置定时调度、从失败处恢复续跑，或需要查询工作流/数据流列表和详情时使用。也适用于把一批已有作业工作流按依赖串成统一调度的主工作流（编排）：从血缘自动推导依赖生成编排（deps plan）、对账编排依赖与血缘是否一致（deps verify）、配置 cron 调度上下线（schedule）、失败后增量恢复（run --recover）。数据流是 BI ETL 的扩展，运行在独立的工作流引擎中。适用于用户说"创建一个数据流""创建一个带 Python 节点的工作流""编辑这个工作流""保存数据流""预览数据流节点""运行工作流""验证这个 Python 脚本能不能跑""列出所有数据流""查看这个工作流的定义""把这批作业串起来统一调度""按血缘生成调度依赖""给工作流配个每天凌晨的调度""调度先下线""编排里有作业挂了修好后接着跑"等场景。 compatibility: "Requires Node.js 14+. Install via npm link --foreground-scripts (local) or npm install -g --foreground-scripts @guandata/guanwf (from internal Nexus registry) so the AI skill refresh result is visible. CLI command: guanwf." --- # guanwf 工作流引擎工作流（ProcessDefinition）的编辑闭环工具。一个工作流是一个节点 DAG， 节点可以是数据流（Dataflow）、Python 脚本、参数赋值等类型。 这是一个执行型 skill，不是只读分析 skill。 - `guancli workflow` 负责通用工作流、数据流、目录和线上结构查询；v0.1.820（8.2.0 兼容线）实例诊断使用 `guanwf instance list/tasks/logs/latest`，可直接解析 `--dir` 工作区。 - `guanwf` 负责把目标工作流拉到本地工作目录，修改事实源，再完成 `export -> preview/validate -> save-draft/save -> run`。 ## 核心概念 - **工作流 (Workflow / PROCESS)**: 顶层容器，包含任务节点 DAG - **数据流 (Dataflow / DATAFLOW)**: 嵌入在工作流中的 ETL 流程，actions 结构与 BI ETL 相同 - **数据库数据流 (DB_DATAFLOW)**: 数据流的变体，支持 SQL 下推 - **Python 节点 (SECURE_PYTHON)**: 工作流级 Python 任务，沙箱内运行脚本，可读输入数据集并创建输出数据集 - **子工作流节点 (SUB_PROCESS + PROCESS)**: 引用另一个已存在的工作流作为节点，用于把一批作业编排成统一调度的主工作流 - **节点依赖**: 节点间用 `After`（成功后）/ `AfterFailure`（失败后）/ `AfterAny`（完成后）声明执行顺序 - **工作流参数**: `Workflow.Params` 声明，使用 `ParamRef` / `DynamicParamRef` / `DataDrivenParamRef` 引用 - **结构化任务节点**: DATASET、HTTP、SQL、PARAMETER_ASSIGNMENT、SWITCH 均有 8.2.0 可执行的强类型 DSL，未知字段由 `task.json` 保留；SHELL/LOOP 仅保留导入模型，8.2.0 不可创建、导出、保存或运行 ### 8.2.0 兼容边界 - 8.2.0 后端没有注册 `SHELL`、`LOOP` TaskParameters，CLI 会在脚手架、导出、保存前明确拒绝。 - `DatasetNode` 可刷新数据库抽取/直连数据集；不能刷新 `DATA_SET_OFFLINE_DEV`，该路径在 8.2.0 会把 BI 文本响应误当 JSON。需要更新离线开发产出时，使用 `SubWorkflowNode` 调用产出工作流。 - `ParameterAssignmentNode` 的 `DATASET` 来源不能选数据库直连数据集（BI 返回 60006）；三张 StarRocks 表应使用 `DataSourceType: "DATABASE"` 和数据账户。抽取/产出数据集可使用 `DATASET` 来源及 `input1` 查询。 ## 关键不变式 **保存工作流时，必须携带完整的工作流上下文（所有 tasks + 所有 dataflowJson）。** `guanwf save` 内部会以服务端最新版本为基底做字段级合并，AI 不需要也不应该手拼保存 payload。 **编辑 ≠ 删除重建。** 编辑已有工作流永远走 `edit → export → save` 原地保存（保留 workflowId）， 改名直接修改 `workflow.go` 中 `Workflow.Name` 后正常 save。guanwf 没有提供工作流删除命令； 用户要求删除时直接说明 CLI 暂不支持。**禁止**用"新建一个工作流替代旧工作流"来模拟编辑或 改名——workflowId 变化会丢失定时/事件调度配置、权限与协作者设置、运行实例历史，且所有通过 `SubWorkflowNode` 引用它的编排工作流会断链。只有用户明确指示"新建替代、再废弃旧的"时才允许， 且执行前必须列出上述影响并获得确认。 ## Harness 工作法：源文件驱动，不手写最终 JSON 先改可读、可复现的源文件，再用明确命令重新生成派生产物。不要为了让保存请求看起来正确而直接改最终 JSON。 **如果任务是生成或修改工作流，但本轮没有修改下面任一事实源，不要汇报完成，也不要直接 save：** - `workflow.go`（工作流结构：节点 + 依赖） - `nodes/<节点ID>/etl.go`、`nodes/<节点ID>/*.sql`（数据流节点） - `nodes/<节点ID>/script.py`（Python 节点脚本） - `nodes/<节点ID>/task.json`（RAW 透传节点） - `nodes/<节点ID>/node_*.json`（被 etl.go 里 `LoadNodeFromFile(...)` 引用的透传节点） 查询、预览、运行、排查已有导出结果时可以不修改源文件；但只要目标是"生成/调整工作流逻辑"，最终变更必须落在上述事实源中。 ### 工作目录结构（Artifact contract） ``` / workflow.go # 可编辑事实源；含凭据时 export 自动收紧为 0600/当前用户 ACL 并加入 .gitignore nodes/<节点ID>/ etl.go # 可编辑事实源：DATAFLOW/DB_DATAFLOW 节点（guanetl DSL） *.sql # 可编辑事实源：SQL 节点外置文件（ReadSQLFile 引用） node_*.json # 可编辑事实源：LoadNodeFromFile 透传节点配置 script.py # 可编辑事实源：PYTHON 节点脚本正文 python.json # 服务端字段透传（edit 生成，勿手改；改配置去 workflow.go 的 PythonConfig） task.json # 0600/当前用户 ACL；RAW 节点为完整事实源，结构化节点保留未知字段 script.sh # SHELL 节点脚本 query.sql # SQL / PARAMETER_ASSIGNMENT 节点 SQL body.json/body.txt # HTTP 请求体 _input.json / meta.json # 导入快照/派生产物，不手改 _layout.json # 画布坐标（edit 保留线上坐标；新节点自动布局补位） _python_runtime_context.json # Python Runtime 结构化上下文（命令生成，只读） _wf_state.json # 系统状态，不手改 _parent_snapshot.json # 脱敏服务端快照，不手改 _parent_merge_snapshot.json # 受限权限的精确三方合并基线，不手改 .gitignore # 自动忽略含原始凭据的精确合并基线 _exported_workflow.json # 派生产物，不手改；由 export 重新生成 ``` ### 禁止的偷懒路径 - 不要直接编辑 `_exported_workflow.json` 来"修好"预览或保存。 - 不要直接编辑 `_parent_snapshot.json` 或 `_parent_merge_snapshot.json` 来拼保存 payload。 - 不要手改 `python.json`；输入/输出/资源配置改 `workflow.go` 的 `PythonConfig`。 - 不要跳过 `export`，拿上一次的导出结果去 `preview` 或 `save`。 如果 `export`、`preview`、`save` 或 `run` 失败，修复顺序固定为：读失败输出 -> 判断是 `workflow.go`、节点源文件、认证/网络还是线上任务问题 -> 修改最小责任源文件 -> 从失败步骤往后重跑。 ## 认证 复用 guancli 的认证链路： ```bash guancli auth login # 先确保已登录 guancli auth status # 检查连接状态 ``` 若由上游托管 OIDC Token，当前进程同时设置 `GUANCLI_OIDC_BASE_URL` 和 `GUANCLI_OIDC_ACCESS_TOKEN` 后直接运行 `guancli auth status`，无需本地 profile，也不执行浏览器登录或自动刷新。 ## 固定工作方式 ### 0. 先分场景 - `只读查询`: 用 `guancli workflow`，不要创建工作目录。 - `新建工作流`: `guanwf create --name ... --parent-dir ...`（8.2.0 支持 `--type DATAFLOW/DB_DATAFLOW/PYTHON/HTTP/SQL/ORCHESTRATION`），然后编辑 `workflow.go` 和 `nodes/`。 - `编辑已有工作流`: `guanwf edit ` 导入整个工作流（数据流节点 → etl.go，Python 节点 → script.py）。 - `编排一批作业`: `guanwf create --type ORCHESTRATION` + `guanwf deps plan` 从血缘自动生成依赖，见「编排工作流」一节。 - `修复失败`: 先定位失败发生在 `export`、`preview`、`save` 还是 `run`，只修最小责任源文件。 ### v0.1.820 参数、节点和实例命令 ```bash guanwf node add http --id http_1 --name "通知下游" --dir guanwf node add sql --id sql_1 --name "执行SQL" --dir guanwf run --dir --param biz_date=2026-07-14 --confirm guanwf instance list --dir --state FAILURE -f json guanwf instance tasks --dir -f json guanwf instance logs --dir --node "<节点名>" --full --log-type ALL -f json guanwf instance latest --dir --logs -f json ``` `node add` 只生成节点目录和可粘贴 DSL 片段，不自动修改 `workflow.go` AST。节点真实字段、 参数类型兼容和外置文件约定见 `references/TASK_NODE_CONTRACTS.md`。 `instance logs --full -f json` 固定返回 `complete/attemptsExpected/attemptsReturned/missingAttempts/fallbackUrl`， 无法证明重跑/恢复历史完整时明确 `complete=false` 并非零退出；诊断同时返回 `firstFailure/rootCause/secondaryErrors` 及 task/log evidence refs。实例和任务 ID 全程按不透明字符串 处理，状态统一输出 canonical `RUNNING_EXECUTION`，后端历史拼写仅保留在 `backendState`。 ### 1. 缺上下文先查，不要猜 查询工作流和数据流定义时使用 `guancli workflow`。工作流/数据流由独立的 workflow 引擎管理。 ```bash guancli workflow tree # 工作流目录树 guancli workflow list # 列出工作流 + 内嵌数据流（默认） guancli workflow list --show-embedded=false # 只列工作流，不展示内嵌数据流 guancli workflow list 销售 # 按关键词搜索 guancli workflow list --page-size 50 # 调整每页条数 guancli workflow list --parent-dir # 指定父目录 guancli workflow get # 查看工作流详情（含数据流节点详情） guancli workflow get --raw # 输出原始 JSON guancli workflow get -f json # JSON 格式输出 guanwf get -f json # 原始定义 + dataflow sidecar 可达性 guanwf status -f json # 轻量状态 + raw/reachable/orphan 分类 ``` `workflow tree` 调用 `/api/directory/MASTER_FLOW/authorized-tree` 获取工作流目录树。 数据流以 SUB_PROCESS 节点形式内嵌在工作流中（`dataFlowNode=true`），不作为独立记录存在。 `list` 默认展示内嵌数据流（遍历每个工作流并汇总），使用 `--show-embedded=false` 可关闭。 顶层 `dataflowJson` 只是 raw sidecar 容器，不能用其 length 判断可执行节点数。只把当前 SUB_PROCESS task 引用的条目视为 `reachableDataflows`；其余为 `orphanSidecars`，get/edit/status 会显式列出。保存会原样保留孤儿 sidecar 及未知扩展字段，但不会把它们放入执行拓扑，也不会默认清理。 涉及 Python 节点时，在编写或修改 `script.py` **之前**查询目标环境的真实 Runtime： ```bash guanwf python-runtime --dir --format json ``` 命令返回默认 Runtime、完整镜像目录和工作区各 Python 节点使用的 `runtimeMode`、`pythonVersion`、 `pythonVersionSource`、`imageId` 及解析状态，并写入 `/_python_runtime_context.json`。`guanwf edit` 检测到 Python 节点时也会自动获取并返回同一结构化上下文。生成代码时只以其中的 `pythonRuntime.nodes[].runtime.pythonVersion`（或 `pythonRuntime.selected`）为兼容目标； CLI 优先通过 `/api/python-images/list` 读取 `pythonVersion`；旧接口不可用时降级到 `/api/system-images/list?type=PYTHON` 并解析 `runtimeEnv`。两类结构化版本字段均缺失或无法识别时， 使用产品兼容契约 Python 3.8，并明确标记 `pythonVersionSource: "compatibility_fallback"`；**禁止从镜像描述、名称或 URL 推断版本**。 结构化上下文中的 `imageUrl` 会移除 userinfo、fragment 和敏感 query；落盘文件使用 0600 并自动忽略。 固定 Runtime 输出 `runtimeMode: "FIXED_RUNTIME"`，配置镜像的动态 Pod 输出 `runtimeMode: "DYNAMIC_POD"`； 目录接口不可用或镜像/版本字段缺失时仍保守生成 Python 3.8 兼容代码，同时保留 warning 和 fallback 来源。 Python 3.8 fallback 下禁止直接生成 `zoneinfo`、`dict[str, ...]` / `list[...]`、`X | None` 等 3.9/3.10+ 写法；分别使用经验证可用的时区依赖、`typing.Dict` / `typing.List`、`typing.Optional`。 不得用固定 `/tmp/...` 路径在 Python 节点之间交换中间结果：不同节点/并发实例可能运行在不同 UID、进程或 Pod。跨节点数据必须通过 `save_outputN` + 下游 `load_inputN` 的正式数据集/DAG 传递； 单节点临时文件使用 `tempfile.TemporaryDirectory()` 并在上下文退出时自动清理。 ### 2. 真正改文件前先读这些 至少先读： - `workflow.go` - 目标节点的源文件（`nodes/<节点ID>/etl.go` / `script.py` / `task.json`） - `references/WORKFLOW_DSL.md`（workflow.go 写法、PythonConfig、依赖声明） - `references/ETL_AI_DEVELOP.md`（数据流节点 etl.go 写法） - `references/WORKFLOW_NODES.md`（数据流扩展节点） 必要时再读： - `nodes/<节点ID>/_input.json`（只读，排查导入前结构） - `_exported_workflow.json`（只读，对比导出结果） - `guancli workflow get -f json` 的线上结构 ### 3. 改完永远按这个顺序 1. `guanwf export --dir ` 2. 数据流节点要看数据结果时 `guanwf preview --dir `；Python 节点要校验时 `guanwf run --node "<节点名>" --validate --dir --dry-run`，确认计划后追加 `--confirm`（需先保存草稿） 3. 先用 `guanwf save-draft --dir --dry-run` 或 `guanwf save --dir --dry-run` 查看计划，确认后追加 `--confirm` 4. 需要执行时先运行 `guanwf run --wait --dir --dry-run`，确认后追加 `--confirm` 如果 `export` 没过，不要直接 `preview` 或 `save`。 ## 默认流程 ### 新建工作流 ```bash guanwf create --name "我的工作流" --parent-dir --dir # 数据流节点骨架（默认） guanwf create --name "Python清洗" --type PYTHON --parent-dir --dir # Python 节点骨架 guanwf python-runtime --dir --format json # 写 Python 前读取目标环境 Runtime # 编辑 workflow.go（节点 + 依赖）和 nodes/ 下源文件后： guanwf export --dir guanwf save-draft --dir --dry-run guanwf save-draft --dir --confirm guanwf preview --dir # 数据流节点：预览数据 guanwf run --node "<节点名>" --validate --dir --confirm # Python 节点：校验（不创建输出数据集） guanwf save --dir --confirm # 正式发布 ``` ### 编辑已有工作流 ```bash guanwf edit --dir # 生成 workflow.go + nodes/<节点ID>/；含 Python 节点时同时生成 _python_runtime_context.json # 修改源文件后与新建流程相同：export → preview/validate → save-draft/save ``` `edit` 会把数据流节点的线上 actions 经 guanetl json2go 导入为 `etl.go`（SQL 外置为 `*.sql`， 复杂节点外置为 `node_*.json` + `LoadNodeFromFile(...)`），这些都是可编辑源文件。 ### 修复失败 1. 先确认失败发生在 `export`、`preview`、`save` 还是 `run`。 2. 如果失败来自 `export`，只改 `workflow.go` 或节点源文件。 3. 如果失败来自 `preview`，先确认 `_exported_workflow.json` 是否由最新源文件生成；必要时先重跑 `export`。 4. 如果失败来自 `save`，不要手写保存 payload；检查工作流是否能重新拉取、本地导出是否来自最新源文件。 5. 从失败步骤往后重跑，不要把整个工作目录推倒重来。 ### 运行工作流 ```bash guanwf run --dir --confirm # 触发运行（已发布版本） guanwf run --wait --dir --dry-run # 查看执行计划 guanwf run --wait --dir --confirm # 确认并等待完成 guanwf run --wait --timeout 600 --dir --confirm # 自定义超时 guanwf run --wait --output-timeout 120 --dir --confirm # 输出注册/更新独立超时 guanwf run --wait --logs --dir --confirm # 完成后输出任务日志 guanwf run --node "<节点名>" --validate --dir --confirm # 校验 Python 脚本（不创建输出数据集） guanwf run --node "<节点名>" --draft --wait --logs --dir --confirm # 单节点真实运行草稿 guanwf run --recover --wait --dir --confirm # 从最近失败实例的失败节点续跑 guanwf instance resume-failed --dom-id --wait --timeout 900 --confirm ``` `--node` 按节点显示名只运行指定节点（TASK_ONLY）；`--draft` 运行 save-draft 的草稿版本； `--logs` 输出任务实例日志（失败时自动拉取，失败的子工作流节点自动下钻子实例日志）； `--recover` 从失败处续跑（见「失败恢复」一节）。 普通 `run` 会在提交前执行 Python 输入规模/内存预检并返回 `PASS/NO_GO/UNKNOWN`；`NO_GO` 默认阻断，`UNKNOWN` 只有在阅读 dry-run 缺失证据后显式追加 `--allow-unknown-memory` 才继续。 预检只读取数据集元数据，不触发 preview 或全量物化。TASK_ONLY + CREATE_NEW 依据版本化能力矩阵 fail closed；用 `guanwf output capabilities` 查询当前矩阵。 启动实例前会从同一份最终执行快照重新解析 `--param`、输出后置条件与 Python 内存门禁； 确认期间发生参数、Python 节点或输出槽位漂移时会在 `start-process-instance` 前失败关闭。 `run --recover` 和 `instance resume-failed` 的最外层实例只有传 `--wait` 才等待完成；嵌套 PROCESS 子实例始终等待成功后才恢复父实例。长任务用 `--timeout` 调整等待上限。 **`--validate` 是验证 Python 脚本的默认方法**：工作流必须先有草稿。CLI 先对最终完整源码 执行 WORKFLOW_PARAMS 唯一入口、类型、引号上下文和残留检查（dry-run 只显示类型/来源，不显示 值），再在临时定义中清空目标 Python 节点的输出注册配置，并把 `save_outputN` 替换为 serializer stub。stub 会在任务唯一临时目录中用真实 PyArrow/Parquet 完成 `Table.from_pandas` 和 `write_table`，随后清理目录且不注册数据集；结束后自动恢复原始草稿。发布态工作流必须先执行 `save-draft --confirm`，CLI 不会为了校验擅自创建无法废弃的临时草稿。日志中可看到 dtype、 head、Arrow schema 和 `serializer=SUCCESS`。详见 `references/WORKFLOW_DSL.md`。 ### 预览数据流节点 ```bash guanwf preview --dir # 只有一个数据流节点时自动选中，预览其终端 action guanwf preview --dir # 多个数据流节点时按 action id 定位 ``` `preview` 基于 export 的本地定义直接发起预览，无需先 save。Python 节点没有 preview API， 校验用 `run --node "<节点名>" --validate`。 ## 编排工作流（把一批作业统一调度） 场景：已有一批作业工作流（每个作业产出数据集），要按依赖串成一个主工作流统一调度。 用户可能这样下达任务（自然语言 → 执行路径映射）： | 用户说 | Agent 执行 | |---|---| | "把 xslh 开头的这批作业串成一个统一调度，每天凌晨 2 点跑" | create ORCHESTRATION → deps plan --search "xslh" → verify → save → schedule set --cron "0 0 2 * * ?" | | "这三个作业 A、B、C 编排起来，A 必须在 B 之前" | deps plan --jobs ,, + hints.json 里 before:[{first:A,then:B}] | | "作业C 是读昨天的快照，别给它连边" | hints.json 里 cross_batch:[{consumer:C,dataset:}] | | "统一调度里有个作业挂了，帮我修好后接着跑，成功的别重跑" | 下钻日志定位失败作业 → 回作业目录修复+save → 编排目录 run --recover --wait | | "调度先别上线/先停掉" | schedule set ... --offline / schedule disable | | "看下现在的调度配置" | schedule info | **依赖的事实源是平台血缘**（作业定义中的输入/输出数据集），不是文档/Excel 的口头描述。 用 `deps plan` 从血缘自动推导，用 `deps verify` 对账，人工只补血缘表达不了的提示 （先后顺序 hint、跨批次 CrossBatch、剔除 exclude）。 ```bash # 1. 建编排工作区（无初始节点） guanwf create --name "统一调度" --type ORCHESTRATION --parent-dir --dir # 2. 从血缘自动生成 workflow.go（SubWorkflowNode + After 边，每条边注释依据） guanwf deps plan --search "<作业名模式>" --yes --dir # 或 --jobs ,,... guanwf deps plan ... --hints hints.json # 人工提示（before/cross_batch/exclude） # 3. 人工微调后对账（ERROR 必须修复；有 ERROR 时 exit 1） guanwf export --dir guanwf deps verify --dir # 4. 保存并运行 guanwf save --dir --confirm guanwf run --wait --dir --confirm ``` ```bash # 5. 配置定时调度并上线 guanwf schedule set --cron "0 0 2 * * ?" --dir --confirm guanwf schedule set --cron "0 0 2 * * ?" --param biz_date=2026-07-14 --dir --confirm guanwf schedule event set --rule : --param biz_date=2026-07-14 --dir --confirm ``` 规则要点： - 一个数据集只能有一个产出作业（DUP_PRODUCER 是 ERROR）。 - A 读 B 的产出则编排中必须有 B→A 路径，否则 verify 报 MISSING_EDGE； 刻意读上一批（T-1）产物时在 `Workflow.CrossBatch` 中声明，verify 降级为 INFO。 - 子作业失败 → 主实例失败，下游不执行；失败日志会自动下钻到子实例定位真实报错。 - **失败恢复**：修复失败作业并 `guanwf save --confirm` 后，在编排目录执行 `guanwf run --recover --wait --confirm`——自底向上恢复失败的子实例和主实例， 成功祖先和无关分支不重跑；恢复节点的全部 DAG 后继会重跑（包括先前为 SUCCESS 的后继）。 例如 job_a/job_b 是 job_c 的成功祖先、job_c 失败时只从 job_c 及其后继开始重跑。 - 修改编排结构改 `workflow.go`（`SubWorkflowNode` / `After` / `CrossBatch`）， 不要去改被引用作业的内容；作业内容问题回作业自己的工作目录修。 `SubWorkflowNode` DSL、hints.json 结构、verify 全部检查项见 `references/WORKFLOW_DSL.md` 「编排工作流」一节。 ## 定时调度 ```bash guanwf schedule info --dir # 查看当前调度 guanwf schedule set --cron "0 0 2 * * ?" --dir --confirm # 创建/更新并上线 guanwf schedule set --cron "0 0 2 * * ?" --offline --dir --confirm # 只保存不上线 guanwf schedule enable --dir --confirm # 上线 guanwf schedule disable --dir --confirm # 下线 ``` - crontab 是 Quartz 6/7 段表达式（秒 分 时 日 月 周 [年]），如 `0 30 6 * * ?` 每天 06:30。 - `--failure-strategy` 是已废弃的兼容参数，仅接受 `CONTINUE`；新离线开发固定提交 `CONTINUE`，失败后的调度行为由节点间 `FAILURE` / `ALL` 连线配置。 - `set` 可选 `--warning NONE|SUCCESS|FAILURE|ALL`（默认 FAILURE）、`--start/--end` 生效区间、 `--auto-cancel-queued`（上一批未跑完时自动取消堆积实例）。 - 已有调度时未指定的参数保留现状；事件调度使用 `guanwf schedule event get/set/disable`。 ## 失败恢复（run --recover） ```bash guanwf run --recover --wait --dir --confirm ``` 找到最近一次 FAILURE 实例，从恢复节点处续跑；成功祖先和无关分支不重跑，恢复节点的 全部 DAG 后继会重跑。8.2.0 的恢复种子状态包括 FAILURE、STOP、KILL、 NEED_FAULT_TOLERANCE。适用于长链路编排中 个别作业失败：先回失败作业的工作目录修复并 `guanwf save --confirm`，再回编排目录 `--recover --confirm`。 平台语义（已实测）：失败的子工作流节点会复用原子实例，且子实例持有失败时的旧定义快照。 `--recover` 已处理这一点——自底向上先恢复失败子实例（`update=true` 吸收失败节点及后继 的最新发布定义），并对恢复范围内已成功的 PROCESS 后继显式执行 `REPEAT_RUNNING`，再恢复 父实例。若子工作流引用已切换、编排 DAG 结构已变更（加减节点/改依赖），CLI/平台会拒绝 恢复，此时直接 `guanwf run --confirm` 重跑整链。 ## save 的工作机制 `save` / `save-draft` 不是简单把本地 JSON 上传。内部流程是： 1. 读取本地 `_exported_workflow.json`（必须由 `export` 重新生成） 2. 从服务端重新拉取工作流最新版本作为基底 3. tasks 按节点 id 做字段级合并：本地定义的字段覆盖，服务端独有字段（dsId、运行时注册信息等）保留 ；Python 输出顶层 `dsId` 与 `dataSource.dsId` 同时存在时必须一致，冲突会在写请求前失败 ；本地显式绑定已有 dsId 的 Python 输出会 fresh 查询数据集并补齐 `parentDirId`、`parentDirName`、 `dirPath`，同时校验物理对象存在性、`DATA_SET_OFFLINE_DEV/MASTER_FLOW` 类型、producer lineage 与输出槽身份；不存在、无权限、类型/血缘/目录/槽位冲突均在写请求前按维度失败。 仅当 `_wf_state.json` 已记录同一 task/slot 的可信 placeholder lineage，或该 slot 是本次保存 明确新增的输出时，服务端回填、`created=false` 且物理对象尚不存在的临时 dsId 才继续按 “待首次运行物化”处理。fresh `edit` 无法仅凭 404 区分 placeholder 与已删除的显式外部绑定， 因而按显式绑定 fail closed；用户改绑后的新 ID 也不会继承旧 pending 来源 4. 节点成员按三方判断（服务端最新 + 本地 `_parent_merge_snapshot.json` + 本地导出）： 本地快照里有、导出里删了的节点 → 删除；服务端有、本地快照不知道的节点（他人并发新增）→ 保留并打印提示 5. dataflowJson 按 dataflow id 经 etlmerge 做 actions 字段级合并（成员判断同上） 6. dry-run 固定输出 `baseVersion/currentVersion/changedFields/noOp/conflicts`；可用 `--require-change` 阻断静默 no-op，用 `--expected-base-version` 做 optimistic CAS。调用 `/process/save-draft` 或 `/process/save` 后必须 fresh readback 所有目标字段，否则命令非零退出； 随后把**本次提交的定义**刷新为本地快照 并同步进 `python.json`（不是服务端回写值——服务端在保存/运行后对输出 dataSource/dsId 等字段的回写需要重新 `guanwf edit` 才能拿到），随后回读服务端做输出落位诊断： 目录被纠正时提示 `save.output_dir_rectified`，输出尚未物化时提示 `save.output_pending_materialization`（此时 dsId 是临时值，首次成功运行前每次保存都会 重新生成，勿用于下游引用）。`run` 等待成功后会校验所有输出的数据集实体线上存在 （`created=true` 直接通过；`created=false` 会查询数据集本身——绑定其他流程创建的 已有数据集时 DSL 要求保持 `Created=false`，该状态合法）。`run --wait` 会冻结每个 slot 的 CREATE_NEW/UPDATE_EXISTING 后置条件，逐项返回 compute/file/discovery/registerOrUpdate/readback； CREATE_NEW 要求唯一 dsId、lineage 和可读实体，UPDATE_EXISTING 还要求版本/commit token/摘要 发生变化。确认后、启动 POST 前的最后执行快照会重新生成输出和内存契约；并发新增/删除 task、slot 或资源漂移不会绕过 preflight。任一失败时 `businessResult != SUCCESS` 且命令非零退出。 因此用户和 AI 都不需要手写服务端保存 payload。多人/多端并发修改同一工作流时，以服务端 最新版本为基底合并，不会把别人新加的节点冲掉；但同一节点的并发修改仍是后保存者覆盖。 只修改一个 Python 输出绑定时，不要编辑完整 DSL；先确保当前版本已有草稿，再使用精确命令： ```bash guanwf output bind --task-id --slot-id --dataset-id --dry-run guanwf output bind --task-id --slot-id --dataset-id --confirm guanwf output rebind --task-id --slot-id --expected-old --dataset-id --confirm guanwf output unbind --task-id --slot-id --expected-old --confirm guanwf output bind-registered-outputs --workflow-id --dom-id --dry-run guanwf output bind-registered-outputs --workflow-id --dom-id --confirm guanwf output capabilities -f json guanwf output bootstrap-plan --task-id --slot-id --update-mode APPEND --primary-key id -f json ``` 命令以 task id + output slot id 精确定位，只保存草稿；确认前再次 fresh read 并执行 optimistic CAS，`expected-old` 不一致或在确认窗口中已检测到定义变化时不会发起写入。后端当前没有暴露原子 条件写接口，最后一次 fresh read 到整份 `save-draft` POST 之间仍遵循上文的同节点后写覆盖语义； 并发编辑同一工作流时应串行执行精确绑定命令。`unbind` 只清除定义中的绑定，不删除物理数据集。 `bind-registered-outputs` 只接受实例中带 `taskId + slotId + dsId` 的精确映射，拒绝按 `dsIds` 数组顺序猜测；`processInstanceJson` 只用于识别 task/slot 定义形状，每个回写还必须有当前实例运行结果中同 task/slot/dsId 的成功状态或与 fresh 数据集精确一致的 commit token，但显式 `FAILURE` 终态永远优先于 token。 所有 slot 先校验后用一份 save-draft 提交，缺失/重复/冲突时零写入，重复执行幂等。 首次 `CREATE_NEW` 运行成功后，CLI 会把精确实例输出标记为“等待绑定发布”；在执行 `bind-registered-outputs` 并通过 `save --confirm` 完成发布前，后续 `run`（包括 dry-run）会以 `output_binding.publish_required` 失败关闭，避免运行时回写的 dsId 被误判为已进入发布版本。 未绑定输出不能一步直接 APPEND/PK；先按 bootstrap plan 以 OVERWRITE 创建，fresh 取得并绑定唯一 dsId，再用 expected-old 切换增量模式和主键。 若他人并发新增的节点依赖了本地这次删除/改名的节点，保存会检测到断链冲突并中止， 此时按提示重新 `guanwf edit <工作流ID>` 同步最新版本后再修改。 ## API 路由 所有 API 通过 `/api/offline-dev/*` 前缀走 BFF 代理到工作流引擎。 | 操作 | 方法 | 路径 | |------|------|------| | 分页查询 | POST | `/api/offline-dev/process/list-paging` | | 查询详情 | GET | `/api/offline-dev/process/{id}/select-by-id` | | 保存草稿 | POST | `/api/offline-dev/process/save-draft` | | 保存发布 | POST | `/api/offline-dev/process/save` | | 预览普通数据流 | POST | `/api/offline-dev/dataflow/preview-async` | | 查询/取消普通预览 | GET/POST | `/api/offline-dev/dataflow/task/{taskId}`、`/cancel` | | 预览 DB Dataflow | POST | `/api/offline-dev/dataflow/v2/preview-async` | | 查询/取消 DB Dataflow 预览 | GET/POST | `/api/task/{taskId}`、`/cancel` | | 运行工作流 | POST | `/api/offline-dev/process/{id}/start-process-instance` | | 查询运行实例 | GET | `/api/offline-dev/process/instance/{id}/select-by-id?domId=` | | 实例任务列表 | GET | `/api/offline-dev/process/instance/{id}/task-list-by-process-id?domId=` | | 任务日志 | GET | `/api/offline-dev/log/detail?taskInstId=&logType=UI&domId=` | | 实例列表分页 | GET | `/api/offline-dev/instance/list-paging?processDefinitionId=&domId=` | | 恢复/停止实例 | POST | `/api/offline-dev/process/{id}/instance/{instId}/execute` | | 子实例定位 | GET | `/api/offline-dev/process/instance/{instId}/task/{taskInstId}/select-sub-process?domId=` | | 调度查询 | GET | `/api/offline-dev/process/{id}/schedule/info?domId=` | | 调度创建/更新 | POST | `/api/offline-dev/process/{id}/schedule/create`、`/schedule/{scheduleId}/update` | 实例/日志查询接口要求 `domId` 参数（工作流详情 `select-by-id` 响应中的 `domId` 字段）， `guanwf run` 内部自动处理。`list-paging` 请求体的搜索字段是 `searchVal`。 ## 与其他 skill 的关系 | 任务 | 工具 | |------|------| | BI ETL 编辑 | guanetl | | 工作流/数据流只读查询 | guancli workflow（隐藏子命令，需手动输入） | | 工作流编辑闭环 | guanwf（本工具） | | 数据集管理 | guands | | workflow.go DSL / Python 节点参考 | guanwf/references/WORKFLOW_DSL.md | | 通用 ETL 节点函数参考 | guanwf/references/ETL_AI_DEVELOP.md（symlink → guanetl） | | 工作流扩展节点参考 | guanwf/references/WORKFLOW_NODES.md | ## 工作流特有节点（数据流节点内部） 工作流数据流是 BI ETL 的超集，额外支持以下节点类型： | 节点类型 | 函数 | 说明 | |---|---|---| | `INPUT_DATASOURCE` | `LoadNodeFromFile` / `BasicInputDatasource` | 数据库直连输入（无上游） | | `OUTPUT_DATASOURCE` | `LoadNodeFromFile` / `BasicOutputDatasource` | 数据回写到数据库 | | `INCREMENT_OUTPUT_DATASET` | `LoadNodeFromFile` / `BasicIncrementOutputDataset` | 增量输出数据集 | | `TRIM_COLUMNS` | `BasicTrimColumns` | 去空格 | | `VALUE_MAPPER` | `LoadNodeFromFile` | 值映射/替换 | | `SINGLE_VALUE_MAPPER` | `LoadNodeFromFile` | Null 值替换 | | `COMBINE_COLUMNS` | `BasicCombineColumns` | 合并列 | `guanwf edit` 导入时： - INPUT_DATASOURCE / OUTPUT_DATASOURCE / INCREMENT_OUTPUT_DATASET / VALUE_MAPPER / SINGLE_VALUE_MAPPER → 保存为 `node_.json` + `LoadNodeFromFile("node_.json")` - TRIM_COLUMNS / COMBINE_COLUMNS → 直接生成结构化 Go DSL 代码 详细配置参数见 `references/WORKFLOW_NODES.md`。 ## 注意事项 - 编辑时自动获取并保存工作流 snapshot，save 时从服务端重新获取最新版本合并 - 数据流 actions 结构与 BI ETL 相同，节点函数和约束复用 guanetl 的 framework - 节点显示名在工作流内必须唯一（依赖声明与 `run --node` 都按名字引用） - DB_DATAFLOW 涉及 SQL 下推和单账号约束，后续可能需要额外处理 - `guancli workflow` 子命令在 `guancli --help` 中不显示（隐藏命令），但手动输入可正常使用 - 旧版（单数据流 `etl/` 格式）工作目录已不支持，遇到时用 `guanwf edit` 在新目录重新拉取