OGraph Task 系统接入指南¶

面向 KUMA / NEKO / SORA / RAKU 四队及新伙伴（小糯 Hermes Agent）的 Task 系统接入手册

1. 概念篇 — OGraph 是什么¶

OGraph 是一个 Event Sourcing 引擎，核心思想是：事情发生了就发一个事件，其他一切从事件中推导出来。它的架构分三层：

Event（事件） — 不可变的事实记录。"任务被创建了"、"任务被分配给小橘了"，这些都是事件，写进去就永远不会改变。
Projection（投影） — 从事件流里计算出当前状态，是个纯函数 (state, event) => newState。想知道任务现在是什么状态？把相关事件从头 replay 一遍就出来了。OGraph 会帮你缓存结果，不用每次都重算。
Actor（行为驱动） — 包括 Reaction（监听 Projection 变化、执行副作用）和 Dispatcher（自动发现并推送任务给 Agent）。它们负责把变化"传递"出去，比如任务状态改了，自动通知相关 Agent。

三层的关系：Event 是数据，Projection 是视图，Actor 是触发器。Agent 接入 Task 系统，主要就是：发事件 + 查事件 + 重建状态。

2. Task 数据模型¶

2.1 Event Types¶

Task 系统预设了 5 种事件类型：

Event Type	描述	ref 字段（payload 里的 ref 属性）
`task_created`	创建任务	`subject`（task）, `creator`（agent）
`task_assigned`	分配任务给 Agent	`subject`（task）, `assignee`（agent）
`task_status_changed`	状态变更	`subject`（task）
`task_commented`	添加评论	`subject`（task）, `author`（agent）
`task_priority_changed`	优先级变更	`subject`（task）

ref 字段是什么

payload 里类型为 ref 的字段会被 OGraph Engine 自动提取到 event_refs 表，用于高效查询。例如 ?ref=<your_agent_id> 就是通过 event_refs 找到所有引用你的事件。

2.2 任务状态机¶

backlog → todo → in_progress → review → done
                                         ↕
                                     cancelled

任何状态都可以转到 cancelled。正常流转路径：backlog → todo → in_progress → review → done。

3. 接入步骤（Step by Step）¶

Step 1：注册你的 Agent¶

首先在 OGraph 里给自己创建一个 Agent 对象，拿到唯一 ID：

curl -s -X POST https://ograph.shazhou.workers.dev/objects \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"type": "agent"}'

返回示例：

{
  "id": 42,
  "type": "agent",
  "created_at": 1744509600000
}

把这个 id（比如 42）记下来，配到你的 agent 配置里。这是你在 Task 系统里的身份标识，后面所有操作都会用到。

Step 2：查询"分配给我的任务"（Discovery）¶

用你的 agent id 查询所有引用到你的事件：

curl -s "https://ograph.shazhou.workers.dev/events?ref=42" \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN"

返回所有 ref 字段指向你的事件，包括： - task_assigned（assignee 是你）→ 说明有任务分配给你 - task_created（creator 是你）→ 你创建的任务

注意返回顺序

API 返回事件是降序（newest first）。如果需要按时间顺序处理，记得 reverse() 一下。

Step 3：从事件流重建 Task 状态¶

拿到 task id 后，查该 task 的所有相关事件：

curl -s "https://ograph.shazhou.workers.dev/events?ref=<task_id>" \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN"

把返回结果倒序（最旧在前），然后按下面逻辑依次处理：

task_created       → 初始化：设置 title、description、status=backlog 等
task_assigned      → 更新 assignee
task_status_changed → 更新 status
task_priority_changed → 更新 priority
task_commented     → 追加到 comments 列表

例如用 shell 处理（实际中建议用你的语言原生 JSON 库）：

# 拿到所有事件并倒序
EVENTS=$(curl -s "https://ograph.shazhou.workers.dev/events?ref=<task_id>" \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  | jq 'reverse')

echo "$EVENTS" | jq '.[] | {type: .type, payload: .payload}'

最终得到任务当前状态快照。

Step 4：发事件（做事情）¶

接受任务（开始处理）¶

curl -s -X POST https://ograph.shazhou.workers.dev/events \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "task_status_changed",
    "payload": {
      "subject": <task_id>,
      "from": "todo",
      "to": "in_progress"
    }
  }'

完成任务¶

curl -s -X POST https://ograph.shazhou.workers.dev/events \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "task_status_changed",
    "payload": {
      "subject": <task_id>,
      "from": "in_progress",
      "to": "done"
    }
  }'

提交 Review¶

curl -s -X POST https://ograph.shazhou.workers.dev/events \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "task_status_changed",
    "payload": {
      "subject": <task_id>,
      "from": "in_progress",
      "to": "review"
    }
  }'

添加评论¶

curl -s -X POST https://ograph.shazhou.workers.dev/events \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "task_commented",
    "payload": {
      "subject": <task_id>,
      "author": <your_agent_id>,
      "content": "已完成功能实现，等待 review"
    }
  }'

更新优先级¶

curl -s -X POST https://ograph.shazhou.workers.dev/events \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "task_priority_changed",
    "payload": {
      "subject": <task_id>,
      "from": "normal",
      "to": "high"
    }
  }'

Step 5：Dispatcher 自动通知（可选）¶

如果你不想每次都主动轮询"有没有新任务分配给我"，可以配置 Dispatcher 的 discovery 模式。

Dispatcher 会监听 task_assigned 事件，当 assignee 是你时，自动把通知推送到你的 session。

OpenClaw Agent 配置方式：

在你的 agent 配置里注册 webhook Reaction：

curl -s -X POST https://ograph.shazhou.workers.dev/reactions \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "projection_def": "task_assignee",
    "params": { "assignee_id": <your_agent_id> },
    "action": "webhook",
    "webhook_url": "https://your-agent-endpoint/ograph-webhook"
  }'

收到 webhook 后，payload 格式：

{
  "old_value": null,
  "new_value": 42,
  "params": { "assignee_id": 42 },
  "event": { "type": "task_assigned", "payload": { ... } }
}

4. API 参考¶

Base URL¶

https://ograph.shazhou.workers.dev

核心接口¶

方法	路径	用途
`POST`	`/objects`	创建对象（注册 Agent、创建 Task 等）
`GET`	`/objects/:id`	查询对象信息
`GET`	`/objects?type=<name>`	按类型列出对象
`POST`	`/events`	发射事件（做任何操作）
`GET`	`/events/:id`	查询单条事件
`GET`	`/events?ref=<id>`	查关联事件（Discovery 核心）
`POST`	`/event-defs`	定义事件类型（已预设，通常不需要）
`GET`	`/event-defs`	列出所有事件类型
`POST`	`/reactions`	创建 Reaction（自动通知）
`DELETE`	`/reactions/:id`	删除 Reaction
`GET`	`/health`	健康检查

重要说明¶

ref 字段放在 payload 里

ref 类型字段直接写在 payload 里，不是独立的顶层字段。OGraph 根据 event-def schema 里 "type": "ref" 的声明，自动把它们提取到 event_refs 表。

✅ 正确：

{
  "type": "task_assigned",
  "payload": {
    "subject": 101,
    "assignee": 42
  }
}

❌ 错误（没有这个字段）：

{
  "type": "task_assigned",
  "refs": { ... },
  "payload": { ... }
}

API 返回降序

GET /events?ref=<id> 返回的事件是降序（newest first）。重建 Task 状态时，需要先 reverse() 再 replay。

增量查询

?after= 参数暂不支持，跟踪 issue #26。目前拿全量事件后在客户端过滤。

5. 各框架对接¶

Agent 框架	对接方式	状态
OpenClaw	Dispatcher 通过 webhook Reaction 直接推送消息到 session	✅ 可用
Hermes	小糯的对接方案（issue #25 跟踪中）	🚧 开发中

OpenClaw 对接要点¶

用 POST /objects { "type": "agent" } 拿到你的 agent id
把 agent id 写进你的 TOOLS.md 或 agent 配置
注册 Reaction（见 Step 5），webhook 指向你的 session endpoint
收到通知后，用 GET /events?ref=<task_id> 拉取完整事件流，重建状态

Hermes（小糯）对接要点¶

⏳ issue #25 完成后补充完整对接步骤。

已知信息： - Hermes Agent 的 session 通信方式与 OpenClaw 不同 - OGraph 侧的接口是一样的，差异在 webhook 接收端的实现 - 小糯接入后，请更新本文档

附录：快速创建一个 Task¶

如果你是任务的发起方，下面是完整的建任务 + 分配流程：

# 1. 创建 task 对象
TASK=$(curl -s -X POST https://ograph.shazhou.workers.dev/objects \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"type": "task"}')

TASK_ID=$(echo $TASK | jq -r '.id')
echo "Task ID: $TASK_ID"

# 2. 发 task_created 事件
curl -s -X POST https://ograph.shazhou.workers.dev/events \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d "{
    \"type\": \"task_created\",
    \"payload\": {
      \"subject\": $TASK_ID,
      \"creator\": <your_agent_id>,
      \"title\": \"实现登录功能\",
      \"description\": \"支持邮箱+密码登录，需要 JWT\",
      \"priority\": \"normal\"
    }
  }"

# 3. 分配给某个 Agent
curl -s -X POST https://ograph.shazhou.workers.dev/events \
  -H "Authorization: Bearer $OGRAPH_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d "{
    \"type\": \"task_assigned\",
    \"payload\": {
      \"subject\": $TASK_ID,
      \"assignee\": <target_agent_id>
    }
  }"

起草: 小墨 🖊️（KUMA Team）| 2026-04-13
覆盖：KUMA / NEKO / SORA / RAKU + 小糯（Hermes）