Openclaw Workflow Publish

OC-Flow：为你的 OpenClaw 注入"确定性"灵魂。OC-Flow 完全嵌入在 OpenClaw 体系内，赋予 Agent 完整的流程控制能力：条件分支、循环遍历、精准等待、状态管理。通过 YAML 剧本实现固定流程、多步循环、严苛逻辑的任务。适用场景：财务办公、开发运维、个人助理。

# OpenClaw Workflow — 确定性工作流引擎

在不破坏 OpenClaw 灵活性的前提下，按 YAML 剧本执行 100% 确定性逻辑：判断、循环、脚本、LLM 调用、Skill 调用。

这是一个符合 OpenClaw / AgentSkills 目录约定的 Skill：
- `SKILL.md`：触发条件与使用说明
- `scripts/`：可执行入口与运行时代码
- `references/`：参考文档与示例工作流

## 使用方法

> **重要**: 脚本内置了 stdout 行缓冲，不需要额外设置 `PYTHONUNBUFFERED`。

```bash
# 运行工作流
python3 {baseDir}/scripts/openclaw_workflow.py execute <workflow.yaml>

# 验证工作流语法
python3 {baseDir}/scripts/openclaw_workflow.py validate <workflow.yaml>

# 列出可用工作流
python3 {baseDir}/scripts/openclaw_workflow.py list

# 查看历史运行
python3 {baseDir}/scripts/openclaw_workflow.py runs

# 从断点恢复
python3 {baseDir}/scripts/openclaw_workflow.py resume <run_id>

# 可视化面板
python3 {baseDir}/scripts/openclaw_workflow.py dashboard
```

## 工作流文件放在哪里

推荐目录（OpenClaw 自建/自维护流程）：

```bash
~/.openclaw/workspace/workflows
```

面板与 CLI 会优先发现以下位置：

1. `{baseDir}/references/examples`
2. `~/.openclaw/workspace/workflows`

说明：面板只扫描固定目录，不会递归读取整个 `~/.openclaw/workspace`。

**示例:**
```bash
# 运行示例工作流
python3 {baseDir}/scripts/openclaw_workflow.py execute {baseDir}/references/examples/basic_test.yaml

# 运行深度集成测试
python3 {baseDir}/scripts/openclaw_workflow.py execute {baseDir}/references/examples/deep_integration.yaml

# 查看补充设计说明
cat {baseDir}/references/readme.md
```

## 输出格式

JSON 运行记录，包含：
- `run_id` — 运行唯一 ID
- `flow_id` — 工作流名称
- `status` — `success` | `failed` | `aborted`
- `steps` — 每步执行结果（状态、输出、耗时、重试次数）
- `started_at` / `finished_at` — 起止时间

终端同时输出人可读的步骤进度日志。

## OpenClaw 深度绑定

- `llm` / `agent` / `skill` 节点通过 **Gateway RPC** 调用，不是本地直连模型。
- 同一工作流运行内所有调用共享同一个 **session**，Agent 拥有完整对话上下文。
- 模型由 OpenClaw Agent/Provider 配置统一管理，无需在工作流中指定。

## 工作流 YAML 格式

```yaml
name: "我的流程"
steps:
  - id: fetch
    type: script
    command: "curl -s https://api.example.com/data"

  - id: analyze
    type: llm
    prompt: "分析: {{fetch.output}}"

  - id: delegate
    type: subagent
    task: "根据以下分析结果生成一份详细报告: {{analyze.text}}"
    label: "报告生成器"
    wait: true
    timeout: 300

  - id: notify
    type: message
    channel: imessage
    target: "+861760051xxxx"
    message: "结果: {{delegate.result}}"
```

## 节点类型速查

| 类型 | 说明 | 关键参数 |
|------|------|----------|
| `script` | Shell/Python 脚本 | `command`, `inline`, `timeout` |
| `llm` | LLM 推理 (Gateway session) | `prompt`, `thinking`, `session` |
| `agent` | Agent 调用 (Gateway session) | `message`, `thinking`, `deliver` |
| `subagent` | 创建子代理执行独立任务 | `task`, `label`, `model`, `wait` |
| `wait_subagents` | 等待多个 Subagent 完成并收集结果 | `tracker`, `max_wait`, `poll_interval` |
| `skill` | 调用 OpenClaw Skill (Gateway session) | `action`, `args`, `instruction` |
| `condition` | If-Else 分支 | `if`, `then`, `else` |
| `loop` | 循环遍历 | `foreach`/`times`, `as`, `do` |
| `set` | 设置变量 | `var`, `value` |
| `log` | 日志输出 | `message`, `level` |
| `http` | HTTP 请求 | `url`, `method`, `headers`, `body` |
| `code` | 内联 Python (沙箱) | `python` |
| `wait` | 延时/等待条件 | `seconds`, `until` |
| `message` | 发送消息 | `channel`, `target`, `message` |

## 变量传递

- `{{variable}}` — 全局变量
- `{{step_id.output}}` / `{{step_id.text}}` — 步骤输出
- `{{item}}` — 循环当前元素
- `{{env.HOME}}` — 环境变量

## 错误处理

每步可配置: `retry: 3`, `retry_delay: 10`, `on_error: retry|skip|stop`

## 典型场景

- **每日简报**: 获取数据 → AI 分析 → 格式化 → 发送
- **发票归档**: 扫描邮件 → 识别发票 → 上传 NAS
- **固件打包**: 拉取代码 → 构建 → 测试 → 打包 → 通知
- **数据监控**: 定时检查 → 条件判断 → 告警
- **并行研究**: 创建多个 subagent 分头调研 → 汇总结果

## Subagent 节点

`subagent` 节点通过 OpenClaw `sessions_spawn` 工具创建独立子代理。

**与 agent 节点的区别:**
- `agent`: 在共享 session 中调用主 Agent，同一上下文
- `subagent`: 创建独立子代理，有自己的 session、系统提示和模型配置

### 创建流程

每个 subagent 的创建需要两层 session:

1. **Spawn session** — 用于承载 `sessions_spawn` 工具调用的"载体 session"
2. **Child session** — 由 `sessions_spawn` 在 Gateway 侧创建的实际子代理 session (`agent:main:subagent:<uuid>`)

`sessions_spawn` 是 Agent 工具 (不是 Gateway 直接 RPC)，所以必须通过 `agent_call` 在某个 session 中触发。spawn session 本身只是工具调用的容器，子代理真正执行任务的是 child session。

### 两种 Spawn Session 模式

#### 模式 A: 传统模式 (独立 session)

每个 spawn 创建一个临时的 `spawn:<hex>` session → `agent_call` 让 Agent 调用 `sessions_spawn` → 提取 childSessionKey → 立即删除 spawn session。

- **Session 数**: N 个 spawn (瞬态，创建后立删) + N 个 child = 持久 N 个
- **并发**: session 独立，天然支持并行
- **上下文**: 每次全新 session，无膨胀问题

#### 模式 B: 工厂模式 (共享 session) — 当前默认

所有 spawn 复用同一个 `factory:<hex>` session → 每 20 次轮换新 session → 用 `factory_lock` 序列化访问。

- **Session 数**: ~ceil(N/20) 个 factory + N 个 child ≈ N + 几个
- **并发**: `factory_lock` **强制所有 spawn 串行执行**，即使在并行循环中
- **上下文**: 累积增长，需要定期轮换

### ✅ 已解决问题 (2026-03-17 → 2026-03-18)

**原问题 1: 循环中工厂模式强制串行**

已通过批量 spawn 解决。循环中自动检测并使用批量创建。

**原问题 2: Gmail 24 封邮件超时**

批量 spawn 10 个 subagent 仅需 ~77s，24 个预计 ~90-120s，远低于超时限制。

**2026-03-18 架构更新: 子会话模式**

非循环场景下的单个 subagent 节点不再通过 sessions_spawn 间接创建，而是直接创建新会话执行任务。
会话就是 subagent，省去了 spawn 中间层。

**参数:**
| 参数 | 必需 | 默认值 | 说明 |
|------|------|--------|------|
| `task` | ✅ | — | 子代理要执行的任务描述 |
| `label` | — | (空) | 子代理显示名称 |
| `model` | — | (继承) | 覆盖使用的模型 |
| `thinking` | — | (关) | 思考级别: off/minimal/low/medium/high |
| `timeout` | — | 300 | 超时秒数 |
| `wait` | — | false | 是否等待子代理完成 |
| `poll_interval` | — | 15 | 等待时轮询间隔 (秒) |
| `mode` | — | run | `run` (一次性) 或 `session` (持久) |
| `cleanup` | — | auto | `keep` (保留) 或 `auto` (wait 完成后自动删除 session) |
| `throttle_timeout` | — | 300 | 并发等待超时 (秒)，达到 maxConcurrent 限制时等待空位 |
| `spawn_timeout` | — | 120000 | Gateway 调用超时 (ms) |
| `spawn_retries` | — | 2 | Gateway timeout 重试次数 |

**示例 — fire-and-forget:**
```yaml
- type: subagent
  task: "整理今天的新闻摘要"
  label: "新闻助手"
```

**示例 — 等待结果:**
```yaml
- id: research
  type: subagent
  task: "研究 {{topic}} 并写一份 500 字的分析报告"
  label: "研究员"
  wait: true
  timeout: 600
  poll_interval: 20

- type: log
  message: "研究结果: {{research.result}}"
```

## Wait Subagents 节点

`wait_subagents` 节点实现标准的 subagent fan-in (汇合) 模式:
- 在 loop 中通过 `subagent` 节点 (wait: false) 创建多个并行子代理
- 用 code 节点在 loop 中收集 spawn 信息到 tracker 列表
- `wait_subagents` 节点轮询所有子代理的 JSONL completion event，全部完成后返回结果列表

### 完成检测 (两级策略)

- **策略 A (精确)**: 读取 spawn session 的 JSONL，查找 auto-announce completion event。仅在 spawn session 未被清理时可用 (wait=true 或工厂模式)
- **策略 B (直读)**: 直接读取 child session 的 JSONL，检查是否有 assistant 回复。这是 wait=false 模式的主要检测路径，因为 spawn session 在创建后立即清理

### 自动清理 (三层保障)

- **即时清理:** 子会话模式 `wait=true` 完成后立即删除 child session
- **汇合清理:** `wait_subagents` 节点完成后批量清理所有 child session (cleanup: auto)
- **兆底清理:** 工作流结束时 (无论成功/失败/中断)，engine.py 的 finally 块会清理:
  - 工作流主 session (`agent:main:openclaw-workflow:<ns>`)
  - 所有残留的 spawn/factory/child session
  - tracker 中记录的 child session
  - 工厂模式内部追踪的 child session (via `get_factory_child_session_keys`)

### 并发控制

`subagent` 节点在创建时会检查两个配置:
- `agents.defaults.subagents.maxConcurrent` (子代理专属限制，默认 20)
- `agents.defaults.maxConcurrent` (Gateway 全局嵌入式运行并发限制，默认 4)

取两者较小值作为实际限制。并发计数仅统计 `:spawn:` session (瞬态占位)，不统计 `:subagent:` session (因为完成后仍残留在 sessions.json 中会导致误判)。

### 内置重试

`subagent` 节点对 Gateway timeout 错误内置 2 次重试 (spawn_retries=2)，spawn 超时时间 120s (spawn_timeout=120000ms)。

**参数:**
| 参数 | 必需 | 默认值 | 说明 |
|------|------|--------|------|
| `tracker` | ✅ | — | spawn 信息列表 (需包含 spawn_session_key, child_session_key) |
| `max_wait` | — | 600 | 最大等待秒数 |
| `poll_interval` | — | 5 | 轮询间隔秒数 |
| `extra_fields` | — | [] | 从 tracker item 透传到结果中的额外字段名 |
| `cleanup` | — | auto | Session 清理策略: `auto` (全部清理), `completed` (仅清理已完成的), `keep` (不清理) |

**示例 — 并行分类:**
```yaml
# 1) 初始化 tracker
- id: init_tracker
  type: code
  python: "result = []"

# 2) loop 中 fire-and-forget spawn + 收集信息
- type: loop
  foreach: "{{items}}"
  as: item
  do:
    - id: spawn_task
      type: subagent
      task: "分析: {{item.data}}"
      wait: false
    - id: collect
      type: code
      python: |
        tracker = init_tracker if isinstance(init_tracker, list) else []
        tracker.append({
            "item_id": item.get("id", ""),
            "spawn_session_key": spawn_task.get("spawn_session_key", ""),
            "child_session_key": spawn_task.get("child_session_key", ""),
        })
        result = tracker

# 3) 等待全部完成
- id: wait_all
  type: wait_subagents
  tracker: "{{init_