close

Main Chain B · Step 07

Inbox:双队列的 replay-once 投影

packages/core/agent/src/inbox.ts(220 行)。Inbox 是 agent 拥有的「持久待办工作」投影:内存状态只是缓存,真相是日志里的 agent/inbox/spliced 事件。所有输入方法(send/claim/splice)都汇到 mutate 这一个函数。

05 agent.ts:113 send → :119 inbox.splice agent/src/inbox.ts:139 splice → :158 mutate (下一步) 08 preStep:229 inbox.claim

示例本次示例:我们的消息在 inbox 里的三个 splice

示例轨迹 07-1 · 消息进入 → 认领 → 清空
# ① 用户点发送 → followup → send('next-turn', wakeup=true)(05 页 122 行)
#    → inbox.splice('next-turn', Infinity, 0, [msg]) → mutate(本页 158 行)
#    归一化:start=Infinity → 钳到队尾 0;插入 1 条 → 落盘:
{ "type":"agent/inbox/spliced", "seq":N, "data":{
    "target":"next-turn", "start":0, "inserted":[{"id":"m1",...}] } }
#    (无 removedCount → 纯插入;无 outcome)

# ② turn 1 的 preStep claim(08 页 229 行)→ claim('next-turn', 1)(本页 71 行)
#    先清 next-step(空)→ 再取 next-turn 队首 1 条 → 两条纯删除 splice 落盘:
{ "data":{ "target":"next-step",  "start":0 } }                       ← 空操作不落盘(170 行)
{ "data":{ "target":"next-turn",  "start":0, "removedCount":1 } }     ← 认领,无 outcome
#    → live 通知 agent/inbox/claimed {message, turn:1}(76 行)

# ③ 用户取消(假设)→ cancel → inbox.clear()(58 行)
#    清 next-step 再清 next-turn → 这次 discardRemoved=true:
{ "data":{ "target":"next-turn", "start":0, "removedCount":1, "outcome":"canceled" } }
#    outcome:'canceled' 是对账依据(§4 foldConsumedWork)
来源:inbox.ts 行级解读 + 05 页 send/claim 路径;splice 记录字段来自 inbox.ts:172-178
示例轨迹 07-2 · resume 时这页如何重放
# 进程重启 → 新 Inbox 构造(本页 32 行)
# 从 header.seedLength 起扫日志里的 agent/inbox/spliced
# 命中上面 ② 的认领记录 → apply() 重放 → next-turn 队列空
# 命中一个进程崩溃前没重放的插入 → 消息回到队列
# 这就是「待办不丢」:真相在日志,内存只是投影。
来源:inbox.ts:32-39 + 20 页持久化章

§1replay-once:构造器

packages/core/agent/src/inbox.ts构造器与状态25-61
25export class Inbox {
26  private readonly state: InboxState = { 'next-turn': [], 'next-step': [] }
28  constructor(private readonly session, private readonly notifications) {
32    for (const event of session.events.slice(session.header.seedLength ?? 0)) {
33      if (event.type !== 'agent/inbox/spliced') continue
34      try { this.apply(event.data) }
35      catch (error) {
37        throw new Error(`invalid persisted inbox splice at session seq ${event.seq}`, { cause: error })
38      }
39    }
40  }
43  get nextTurn() { return this.state['next-turn'] }
48  get nextStep() { return this.state['next-step'] }
53  get hasPending() { return this.nextTurn.length > 0 || this.nextStep.length > 0 }
58  clear() {
59    this.splice('next-step', 0, this.nextStep.length, [])
60    this.splice('next-turn', 0, this.nextTurn.length, [])
61  }
26

两个队列就是全部状态:next-turn(等待各自轮次的消息)与 next-step(等待下一步边界的输入)。

32

replay-once 投影:构造时从 header.seedLength 起回放日志里的 agent/inbox/spliced 事件重建内存状态。为什么从 seedLength 开始?seed 前缀(父历史)已经含在投影里,不需要重放。这就是「resume 后 inbox 恢复」的机制。

34-38

回放校验失败 → 报错并带 cause(哪条 seq 坏了)。持久化的 splice 不可信,回放必须验证。

53

hasPending 是 turn() 324 行与 kick 220 行判断「还有没有活」的依据。

58-61

clear(cancel 时用):先清 next-step 再清 next-turn——顺序有意为之,保证任何时刻 next-turn 里的消息不被转向半途截胡。两个 splice 各自落盘(durable 的取消)。

§2mutate:先落盘,再改投影

packages/core/agent/src/inbox.ts所有变更的唯一出口139-193
139  splice(target, start, deleteCount, inserted): UserMessage[] {
145    return this.mutate(target, start, deleteCount, inserted, true)
146  }
158  private mutate(target, start, deleteCount, inserted, discardRemoved): UserMessage[] {
159    const inbox = this.state[target]
160    const truncatedStart = Math.trunc(start)
161    const offset = Number.isNaN(truncatedStart) ? 0 : truncatedStart
162    const actualStart = offset < 0
163      ? Math.max(inbox.length + offset, 0)
164      : Math.min(offset, inbox.length)
165    const truncatedDeleteCount = Math.trunc(deleteCount)
166    const actualDeleteCount = Math.min(
167      Math.max(Number.isNaN(truncatedDeleteCount) ? 0 : truncatedDeleteCount, 0),
168      inbox.length - actualStart,
169    )
170    if (actualDeleteCount === 0 && inserted.length === 0) return []
171    const outcome = discardRemoved && actualDeleteCount > 0 ? 'canceled' as const : undefined
172    const splice = {
173      target,
174      start: actualStart,
175      ...(actualDeleteCount === 0 ? {} : { removedCount: actualDeleteCount }),
176      inserted,
177      ...(outcome === undefined ? {} : { outcome }),
178    }
179    this.validate(splice)
180    const event = this.session.append('agent/inbox/spliced', splice)
181    const removed = inbox.splice(actualStart, actualDeleteCount, ...event.data.inserted)
182    if (discardRemoved) {
183      for (const message of removed) this.notifications.discarded(message)
184    }
185    for (const message of event.data.inserted) this.notifications.inserted(message)
186    return removed
187  }
139-146

公开的 splice 与内部 mutate 的差别只在 discardRemoved:公开 splice 传 true(被移除的消息是「取消」),claim 内部传 false(被移除的消息是「认领」,不算取消)。

160-170

归一化坐标:截断小数、NaN 归零、负 offset 从尾部算、越界钳到界内。任何输入都归一成「合法 splice」。零操作直接返回(不产生无意义事件)。

171

outcome: 'canceled' 只标在「真的删了东西」的公开 splice 上——这是 foldConsumedWork 对账的依据(§4)。

172-178

splice 记录是归一化后的坐标(target/start/removedCount/inserted),不是操作原始输入——回放只需机械重放,无需重算。

179

validate 在 append 之前——坏 splice 不进日志(§3)。

180-181

关键顺序:先 append(durable),再 splice(内存)。事件进日志后,内存投影才改。因此同步的 session/event 观察者读到的是 splice 前的列表,可用记录坐标重建被移除的消息——这正是 06 页 append 里「通知收集在 push 之前」的消费场景。

182-185

两个通知方向:被移除的消息发 discarded(仅公开 splice),插入的消息发 inserted。回放(apply,196-200 行)不通知——回放不是新事实。

§3claim 与 validate

packages/core/agent/src/inbox.tsclaim 与 validate71-78, 203-219
71  claim(target: InboxTarget, turn: number): UserMessage[] {
72    const claimed = this.mutate('next-step', 0, this.nextStep.length, [], false)
73    if (target === 'next-turn') {
74      claimed.push(...this.mutate('next-turn', 0, 1, [], false))
75    }
76    for (const message of claimed) this.notifications.claimed(message, turn)
77    return claimed
78  }
203  private validate(splice): void {
204    const inbox = this.state[splice.target]
205    const removedCount = splice.removedCount ?? 0
206    if (!Number.isSafeInteger(splice.start) || splice.start < 0 || splice.start > inbox.length
207      || !Number.isSafeInteger(removedCount) || removedCount < 0
208      || splice.start + removedCount > inbox.length) {
209      throw new Error('invalid inbox splice')
210    }
211    const candidate = inbox.toSpliced(splice.start, removedCount, ...splice.inserted)
212    const ids = new Set<string>()
213    for (const message of splice.target === 'next-turn'
214      ? [...candidate, ...this.nextStep]
215      : [...this.nextTurn, ...candidate]) {
216      if (ids.has(message.id)) throw new Error(`message "${message.id}" is already pending`)
217      ids.add(message.id)
218    }
219  }
72-74

claim 的语义:总是先清空 next-step(全部),仅当 target 是 next-turn 时再取一条排队 turn。返回「next-step 输入 + 排队 turn」的完整批——这就是 08 页 preStep 拿到的 claimed。

72, 74

两次 mutate 都传 discardRemoved=false——认领的删除不是取消,不落 outcome:'canceled',不发 discarded 通知。claim 的 durable 记录是「纯删除 splice」。

76

每条认领消息发 agent/inbox/claimed(带 owning turn)——被拒 step 的消息终点就是这里:它既不被 discarded 也不 re-emit 成 user/message。

206-209

坐标合法性:start/removedCount 必须是安全整数、非负、且在界内。

211-218

跨队列 id 唯一:插入后两个队列里任何消息 id 不能重复——同一消息不能同时排队又待转向。id 唯一性是 cancel/replace 按 id 定位的前提。

§4对账:foldConsumedWork

packages/core/agent/src/consumed-work.ts(108 行)的纯函数 foldConsumedWork 是「从日志可重建」原则的又一实例:只靠 session 日志(turn/step 边界 + inbox splice)就能判定一个 agent 到底消费/丢弃了哪些工作——不需要任何内存状态。它的对账依据就是本页落下的两类记录:

  • turn 的开合(turn/start、turn/end)圈定「哪些工作属于哪个 turn」;
  • splice 的 removedCount + outcome:'canceled' 区分「被认领消费」与「被取消丢弃」。

这解释了 171 行为什么 outcome 只在公开 splice 上标记:取消是对账需要的事实,认领不是。