Cell Protocol v1（M4c）

Coordinator ↔ cell worker。TCP NDJSON，一行请求/一行响应。 与 wire-protocol.md（flow 级）不同：这里是 agent cell 级，worker 持有 cell 状态。

1. 传输

• TCP；coordinator 主动连接；一个连接一个请求。
• 请求/响应各一行 UTF-8 JSON + \n。
• 超时：15s（cluster demo）/ 2s（测试）。
• 无 TLS；共享 token 鉴权。

2. Envelope / Outbound

type Envelope = {
  run: string;          // run id；与 id 一起做 dedupe
  id: u64;              // coordinator 单调递增
  channel: string;      // "<from>:<port>"，per-edge 通道
  seq: u64;             // per-channel 单调序号
  capability: string;   // target cell 的 capability
  from: string;         // cell id 或 "input"
  to: string;           // target cell id
  port: string;         // sender port name
  message: Message;     // { tag, payload, from? }
};

type Outbound = {
  from: string;
  port: string;
  to?: string | null;   // null = terminal output
  message: Message;
};

3. Request

type CellRequest =
  | { type: "ping"; token?: string }
  | { type: "spawn"; token?: string; cell: AgentState; capability: string }
  | { type: "deliver"; token?: string; envelope: Envelope }
  | { type: "snapshot"; token?: string; cell: string; capability: string }
  | { type: "remove"; token?: string; cell: string; capability: string }
  | { type: "set_capability"; token?: string; cell: string; capability: string }
  | { type: "stats"; token?: string };

4. Response

type CellResponse =
  | { type: "pong"; node: string; version: 1 }
  | { type: "spawned"; node: string }
  | { type: "ack"; node: string; id: u64; duplicate: bool;
      state: AgentState; sends: Outbound[]; usage: Usage }
  | { type: "snapshot"; node: string; cell: AgentState }
  | { type: "removed"; node: string }
  | { type: "stats"; node: string; cells: u64; deliveries: u64;
      failures: u64; uptime_secs: u64 }
  | { type: "error"; node: string; error: string };

5. 语义

deliver

1. 若 envelope.id 已在 worker seen 集合 → ack { duplicate: true, sends: [], usage: 0 }。
2. 取 to cell，调用 behavior；更新 state。
3. 若 action 有 effect：worker 本地执行 EffectBus，usage 累加；把 Message("effect", output) 再交给同一 cell（最多 8 轮）。
4. action.send 经 cell.ports 解析为 Outbound.to；null 表示终端输出。
5. 返回 ack { state, sends, usage }。

spawn / snapshot / remove

• spawn：插入 AgentState；behavior kind 必须已知。
• snapshot：返回 cell 当前状态（迁移用）。
• remove：删除 cell。
• 迁移：snapshot(source) → spawn(target) → remove(source) → 更新 placement。

dedupe / at-least-once

• id 由 coordinator 唯一分配；worker 丢弃重复。
• 失败重试可能重复投递；id 相同则不会重复执行。
• 但 effect 本身可能已执行（at-least-once），需要 effect idempotency key 才能 exactly-once（M4d）。

5.1 per-edge 通道与 capability（M4d）

• worker 按 (run, channel) 维护 last_seq：
  • seq == last+1 接受；
  • seq <= last 返回 duplicate:true；
  • seq > last+1 返回错误（gap）。
• capability 在 spawn 时绑定到 cell；deliver/snapshot/remove 必须匹配。
• set_capability 用于撤销/轮换；旧 capability 立即失效。
• 有 --secret 时 capability 是 v1:<run>:<epoch>:<hmac>（HMAC-SHA256），worker 在 spawn 时校验 MAC；无 secret 时退回 run-cell-seq 字符串。
• revoke 增加 epoch，旧 capability 立即失效。
• mTLS 仍未实现；TCP 明文，依赖共享 token + 受信网络。

5.2 effect 幂等缓存

• worker 以 run:cell:op:input_hash 缓存 effect output。
• 同输入再次出现（不同 envelope id）不重复调用 handler，不计费。
• 缓存是 per-worker；跨节点 failover 后丢失（需要持久化幂等键）。

6. 错误

• unauthorized：token 不匹配。
• unknown cell / unknown behavior / effect 错误。
• coordinator 对传输错误执行 failover：用 last_states 在存活节点重新 spawn。

7. 限制

• 无 mTLS / capability per cell。
• 无背压；coordinator 队列无界。
• 无 worker 自动心跳/再入；kill/healthy_nodes 手动。
• 无持久化 dedupe；worker 重启后 seen 丢失。
• cell 状态快照只在 coordinator 的 last-ack；没有 WAL。

7.1 队列与持久化

• coordinator with_queue_limit(n) 提供 admission control（超限 Backpressure）。
• effect cache 是 per-worker 内存；worker 重启/failover 后丢失。
• audit hash = sha256(events)，--save-audit 落盘。

8. 版本

• version: 1（pong.version）。
• 测试：crates/ain-cluster/tests/cluster.rs。

在 GitHub 编辑本文