fix(agent): validate durable inbox reconstruction

2026-09-09 04:02:35 +00:00 · 2026-08-31 13:35:49 +08:00
parent 4b1683c287
commit e5f2fbd9a2
11 changed files with 87 additions and 35 deletions
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write .agents/notes/implemented/architecture/2026-07-31-claimed-pre-step-inbox-lifecycle.md
-2026-07-31-claimed-pre-step-inbox-lifecycle.md: 8149c9df09ff69c017ba66a779d6fd3dfc314193
-2026-07-31-claimed-pre-step-inbox-lifecycle.zh.md: e855b88b26a55dd9f5c33c87a4b59d67496d00ed
+2026-07-31-claimed-pre-step-inbox-lifecycle.md: aee385bec30c572bd23d7aecac3a3515ad8cd98b
+2026-07-31-claimed-pre-step-inbox-lifecycle.zh.md: 79460b9abf25ef12925f459a8d13b78c7e63b604
@@ -18,7 +18,7 @@ Before every proposed step, `Inbox.claim(target)` atomically removes the complet

 The durable inbox remains two `UserMessage[]` lists addressed by `MessageId`. `append`, `prepend`, and `splice` take a target, while `replace(messageId, newMessage)` and `remove(messageId)` locate the pending message across both lists before committing a normalized splice. Replacement may change identity and emits the old message as discarded followed by the new message as inserted. Every insertion emits `agent/inbox/inserted { message }`; an ordinary removal records `outcome: 'canceled'` and emits `agent/inbox/discarded { message }`. Claiming records pure deletions without an outcome and emits claimed events from Inbox itself. These live events add no placement, outcome, or batch fields.

-The two event surfaces have separate consumers. Observers following one message use `agent/inbox/inserted`, `claimed`, and `discarded`. `AgentRegistry` contributes the standard `inbox` projection over the durable `agent/inbox/spliced` stream whenever the projection registry is composed; UI edits and removals route through an Inbox mutation method so the same projection records every change.
+The two event surfaces have separate consumers. Observers following one message use `agent/inbox/inserted`, `claimed`, and `discarded`. `AgentRegistry` contributes the standard `inbox` projection over the durable `agent/inbox/spliced` stream whenever the projection registry is composed; UI edits and removals route through an Inbox mutation method so the same projection records every change. When that projection reconstructs durable history, it rejects unsafe or out-of-range coordinates and duplicate `MessageId` values across both lists, and reports the offending event seq.

 Plugins that need current-step atomic rewriting return messages from `agent/pre-step`. Plugins that only need later context may mutate `agent.inbox` directly. Workspace context uses both paths: asynchronous filesystem projections stage one replaceable `next-step` item, while the next entering pre-step folds that item or a newly composed baseline into its final batch and removes the pending copy. Rejection keeps the item queued.

@@ -34,7 +34,7 @@ The archived [addressable queue occurrence decision](../../archived/feature/2026

 ## Verification

-Agent-loop coverage pins turn-start-before-claim-before-pre-step ordering, exact live event payloads, balanced no-step rejection, final-batch rewriting, input inserted after a claim, listener failure, and cancellation. Inbox and consumer tests pin pure claim deletions, canceled ordinary removals, agent-instructions staging, replacement, and same-step entry, plan/goal/hook behavior, UI cleanup, compaction, checkpointing, and resumed durable projection. Generated event and type catalogs expose only the new waterfall and payloads.
+Agent-loop coverage pins turn-start-before-claim-before-pre-step ordering, exact live event payloads, balanced no-step rejection, final-batch rewriting, input inserted after a claim, listener failure, and cancellation. Inbox and consumer tests pin pure claim deletions, canceled ordinary removals, agent-instructions staging, replacement, and same-step entry, plan/goal/hook behavior, UI cleanup, compaction, checkpointing, resumed durable projection, and rejection of invalid persisted coordinates or cross-list identities. Generated event and type catalogs expose only the new waterfall and payloads.

 ## Consequences

@@ -18,7 +18,7 @@ Status: implemented

 持久 inbox 仍是两份通过 `MessageId` 寻址的 `UserMessage[]` 列表。`append`、`prepend` 与 `splice` 接受 target；`replace(messageId, newMessage)` 与 `remove(messageId)` 则在提交规范化 splice 前，通过 `MessageId` 跨两份列表定位待处理消息。替换可以改变标识，并先将旧消息作为 discarded 发布，再将新消息作为 inserted 发布。每次插入发出 `agent/inbox/inserted { message }`；普通删除记录 `outcome: 'canceled'` 并发出 `agent/inbox/discarded { message }`。领取记录不带 outcome 的纯删除，并由 Inbox 自行发出 claimed 事件。这些实时事件不增加 placement、outcome 或批次字段。

-两类事件接口服务不同消费方。跟踪单条消息的观察方使用 `agent/inbox/inserted`、`claimed` 与 `discarded`。`AgentRegistry` 会在投影注册表已组合时，在持久 `agent/inbox/spliced` 流上贡献标准 `inbox` 投影；UI 编辑与移除通过 Inbox 变更方法处理，从而让同一投影记录所有变化。
+两类事件接口服务不同消费方。跟踪单条消息的观察方使用 `agent/inbox/inserted`、`claimed` 与 `discarded`。`AgentRegistry` 会在投影注册表已组合时，在持久 `agent/inbox/spliced` 流上贡献标准 `inbox` 投影；UI 编辑与移除通过 Inbox 变更方法处理，从而让同一投影记录所有变化。该投影重建持久历史时，会拒绝不安全或越界的坐标，以及跨两份列表重复的 `MessageId`，并报告出错事件的 seq。

 必须对当前步骤进行原子改写的插件从 `agent/pre-step` 返回消息。只需要稍后上下文的插件可以直接修改 `agent.inbox`。Workspace context 同时使用两条路径：异步文件系统投影会暂存一条可替换的 `next-step` 消息，而下一次进入步骤的 pre-step 会把该消息或新组合的基线折入最终批次，并移除仍待处理的副本。reject 会让该条目继续排队。

@@ -34,7 +34,7 @@ Status: implemented

 ## 验证

-agent loop（智能体循环）覆盖固定先 `turn/start`、再领取、后 pre-step 的顺序、实时事件的确切载荷、边界平衡的无步骤 reject、最终批次改写、领取后插入的输入、监听器失败与取消。Inbox 和消费方测试固定纯领取删除、普通删除的 canceled 结果、agent-instructions 的暂存、替换与同一步骤进入、plan/goal/钩子行为、UI 清理、压缩（compaction）、检查点以及恢复后的持久投影。生成的事件与类型目录只公开新的 waterfall 与载荷。
+agent loop（智能体循环）覆盖固定先 `turn/start`、再领取、后 pre-step 的顺序、实时事件的确切载荷、边界平衡的无步骤 reject、最终批次改写、领取后插入的输入、监听器失败与取消。Inbox 和消费方测试固定纯领取删除、普通删除的 canceled 结果、agent-instructions 的暂存、替换与同一步骤进入、plan/goal/钩子行为、UI 清理、压缩（compaction）、检查点、恢复后的持久投影，以及对非法持久坐标或跨列表重复标识的拒绝。生成的事件与类型目录只公开新的 waterfall 与载荷。

 ## 后果

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write docs/subsystems/core.md
-core.md: 03299a114987d9508b8d58ec24a3c4f919006648
-core.zh.md: bee87294670a7f7ea23169e00beb0ff324846025
+core.md: 2d80985bb16ea2858bffd53e4ab584e096c2dc7a
+core.zh.md: 6c731e07decbc50cdd552d81615b511f42266aef
@@ -177,7 +177,7 @@ The inbox is the delivery vocabulary — two ordered pending-message lists the a
 type InboxTarget = 'next-turn' | 'next-step'
 ```

-Every pending occurrence is its `UserMessage`; `MessageId` is the sole identity. `Inbox.append`, `prepend`, `replace`, `remove`, `clear`, `splice`, and `claim` record normalized durable `agent/inbox/spliced` mutations and reject duplicate pending ids. `replace(messageId, newMessage)` and `remove(messageId)` locate the pending message across both lists; replacement may change identity and emits the old message as discarded followed by the new message as inserted. Ordinary removals and `clear()` are cancellations. `claim(target)` removes the proposed step batch — all `next-step` input plus, at a turn boundary, one `next-turn` message — through pure deletion splices without emitting discarded notifications, then Inbox emits per-message claimed notifications. `AgentRegistry` contributes the standard `inbox` projection whenever the projection registry is composed; its cell is the sole live state and the same fold serves cold consumers. Consumers following one message use the exact `agent/inbox/inserted`, `claimed`, and `discarded` notifications.
+Every pending occurrence is its `UserMessage`; `MessageId` is the sole identity. `Inbox.append`, `prepend`, `replace`, `remove`, `clear`, `splice`, and `claim` record normalized durable `agent/inbox/spliced` mutations and reject duplicate pending ids. `replace(messageId, newMessage)` and `remove(messageId)` locate the pending message across both lists; replacement may change identity and emits the old message as discarded followed by the new message as inserted. Ordinary removals and `clear()` are cancellations. `claim(target)` removes the proposed step batch — all `next-step` input plus, at a turn boundary, one `next-turn` message — through pure deletion splices without emitting discarded notifications, then Inbox emits per-message claimed notifications. `AgentRegistry` contributes the standard `inbox` projection whenever the projection registry is composed; its cell is the sole live state and the same fold serves cold consumers. That fold rejects unsafe or out-of-range splice coordinates and duplicate identities across both lists, identifying malformed durable history by event seq. Consumers following one message use the exact `agent/inbox/inserted`, `claimed`, and `discarded` notifications.

 Cancellation:

@@ -181,7 +181,7 @@ inbox 即投递词汇——agent 以持久投影形式拥有的两条有序待
 type InboxTarget = 'next-turn' | 'next-step'
 ```

-每个待处理入队项就是其 `UserMessage`；`MessageId` 是唯一标识。`Inbox.append`、`prepend`、`replace`、`remove`、`clear`、`splice` 与 `claim` 会记录规范化的持久 `agent/inbox/spliced` 变更，并拒绝重复的待处理 id。`replace(messageId, newMessage)` 与 `remove(messageId)` 通过 `MessageId` 跨两份列表定位待处理消息；替换可以改变标识，并先将旧消息作为 discarded 发布，再将新消息作为 inserted 发布。普通删除和 `clear()` 都表示取消。`claim(target)` 通过纯删除 splice 移除拟进入步骤的批次——全部 `next-step` 输入，外加轮次边界上的一条 `next-turn` 消息——且不发出 discarded 通知，随后由 Inbox 逐条发出 claimed 通知。`AgentRegistry` 会在投影注册表已组合时贡献标准 `inbox` 投影；其 cell 是唯一 live 状态，同一份折叠也服务于冷消费方。跟踪单条消息的消费方使用精确的 `agent/inbox/inserted`、`claimed` 与 `discarded` 通知。
+每个待处理入队项就是其 `UserMessage`；`MessageId` 是唯一标识。`Inbox.append`、`prepend`、`replace`、`remove`、`clear`、`splice` 与 `claim` 会记录规范化的持久 `agent/inbox/spliced` 变更，并拒绝重复的待处理 id。`replace(messageId, newMessage)` 与 `remove(messageId)` 通过 `MessageId` 跨两份列表定位待处理消息；替换可以改变标识，并先将旧消息作为 discarded 发布，再将新消息作为 inserted 发布。普通删除和 `clear()` 都表示取消。`claim(target)` 通过纯删除 splice 移除拟进入步骤的批次——全部 `next-step` 输入，外加轮次边界上的一条 `next-turn` 消息——且不发出 discarded 通知，随后由 Inbox 逐条发出 claimed 通知。`AgentRegistry` 会在投影注册表已组合时贡献标准 `inbox` 投影；其 cell 是唯一 live 状态，同一份折叠也服务于冷消费方。该 fold 会拒绝不安全或越界的 splice 坐标，以及跨两份列表重复的标识，并通过事件 seq 指出格式错误的持久历史。跟踪单条消息的消费方使用精确的 `agent/inbox/inserted`、`claimed` 与 `discarded` 通知。

 取消：

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/core/agent/README.md
-README.md: d56a917976e07d99e947703294bfe1e195533f8c
-README.zh.md: 3cc0cec48861161bdc61550d9f8b13c78109e94a
+README.md: 90e6aa0f570f5bde4a120f2d8213fb308cb7792a
+README.zh.md: 8922d78591211a6b48976f2c5632766d90eb3186
@@ -54,7 +54,7 @@ Most interception points are cooperative waterfalls. `agent/pre-step` receives a

 `PreStepDecision` is either `{ kind: 'reject' }` or `{ kind: 'enter', messages, startsRequestSeries? }`. The enter branch is the complete identified, frozen batch for the proposed step. `startsRequestSeries: true` declares that this admitted batch begins a distinct model-message series; ordinary follow-ups leave it absent. A listener that wraps downstream entry preserves both that declaration and the batch unless it intentionally replaces either one; additions follow the waterfall's natural return order. Claiming already removed the offered messages from the inbox, so rejection does not retain them. Messages inserted after the claim remain pending for a later boundary.

-`AgentRegistry` contributes the standard `inbox` session projection whenever the projection registry is composed. The registry folds durable `agent/inbox/spliced` events once and remains the sole owner of the live `{ 'next-turn', 'next-step' }` state; Inbox is a command facade that reads that unit rather than replaying or copying the fold. Inbox live notifications are deliberately per-message and minimal: `agent/inbox/inserted { message }`, `agent/inbox/claimed { message, turn }`, and `agent/inbox/discarded { message }`. Inbox emits them as it commits the corresponding mutation, without adding another lifecycle envelope.
+`AgentRegistry` contributes the standard `inbox` session projection whenever the projection registry is composed. The registry folds durable `agent/inbox/spliced` events once and remains the sole owner of the live `{ 'next-turn', 'next-step' }` state; Inbox is a command facade that reads that unit rather than replaying or copying the fold. Reconstruction rejects unsafe or out-of-range splice coordinates and duplicate `MessageId` values across both pending lists, reporting the offending event seq instead of accepting malformed durable history. Inbox live notifications are deliberately per-message and minimal: `agent/inbox/inserted { message }`, `agent/inbox/claimed { message, turn }`, and `agent/inbox/discarded { message }`. Inbox emits them as it commits the corresponding mutation, without adding another lifecycle envelope.

 Turn and step boundaries and the model token stream are durable `session/event` facts rather than mirrored `agent/*` notifications. Consumers read `turn/*`, `step/*`, and `assistant/chunk` from the session feed; tool policy and outcome observation belong to the complete pipeline documented by [`dsh-tools`](../tools/README.md).

@@ -56,7 +56,7 @@ Agent *创建* 由实现 `AgentFactory` 的插件（`dsh-agent-loop`）提供，

 `PreStepDecision` 要么是 `{ kind: 'reject' }`，要么是 `{ kind: 'enter', messages, startsRequestSeries? }`。enter 分支是拟进入步骤的完整、带标识且冻结的批次。`startsRequestSeries: true` 声明该接纳批次会开启一个独立的模型消息序列；普通 follow-up 不设置它。包装下游 enter 的监听器会同时保留该声明和消息批次，除非有意替换其中一项；新增消息遵循 waterfall 的自然返回顺序。领取操作已经把候选消息从 inbox 删除，因此 reject 不会保留它们；领取后插入的消息仍等待后续边界。

-`AgentRegistry` 会在投影注册表已组合时贡献标准 `inbox` 会话投影。注册表只折叠一次持久 `agent/inbox/spliced` 事件，并继续作为 live `{ 'next-turn', 'next-step' }` 状态的唯一所有者；Inbox 是读取该单元的命令 facade，不会重新回放或复制折叠结果。Inbox 的实时通知刻意采用逐消息的最小载荷：`agent/inbox/inserted { message }`、`agent/inbox/claimed { message, turn }` 与 `agent/inbox/discarded { message }`。Inbox 在提交对应变更时自行发出这些通知，不引入另一层生命周期封套。
+`AgentRegistry` 会在投影注册表已组合时贡献标准 `inbox` 会话投影。注册表只折叠一次持久 `agent/inbox/spliced` 事件，并继续作为 live `{ 'next-turn', 'next-step' }` 状态的唯一所有者；Inbox 是读取该单元的命令 facade，不会重新回放或复制折叠结果。重建过程会拒绝不安全或越界的 splice 坐标，以及跨两份待处理列表重复的 `MessageId`，并报告出错事件的 seq，而不会接受格式错误的持久历史。Inbox 的实时通知刻意采用逐消息的最小载荷：`agent/inbox/inserted { message }`、`agent/inbox/claimed { message, turn }` 与 `agent/inbox/discarded { message }`。Inbox 在提交对应变更时自行发出这些通知，不引入另一层生命周期封套。

 轮次和步骤边界以及模型 token 流是持久 `session/event` 事实，而不是镜像的 `agent/*` 通知。消费方从会话事件流读取 `turn/*`、`step/*` 和 `assistant/chunk`；工具策略与结果观测属于 [`dsh-tools`](../tools/README.zh.md) 记录的完整流水线。

@@ -3,6 +3,7 @@
 import type { UserMessage } from '@deepseek-ai/dsh-llm/types'
 import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
 import { z } from 'zod'
+import type { InboxState, InboxWireState } from './types.ts'

 /** Wire validation for pending agent input reconstructed from durable inbox splices. */
 export const inboxProjectionSchema = z.object({
@@ -10,10 +11,7 @@ export const inboxProjectionSchema = z.object({
  'next-step': z.array(z.custom<UserMessage>()).readonly(),
 }).readonly()

-/** Complete pending Inbox value reconstructed from durable splices. */
-export type InboxState = z.infer<typeof inboxProjectionSchema>
-
-/** Standard fold that reconstructs pending agent input from durable splices. */
+/** Standard fold that reconstructs pending input and rejects invalid durable splice history. */
 export const inboxProjectionDefinition = {
  key: 'inbox',
  stateSchema: inboxProjectionSchema,
@@ -21,25 +19,35 @@ export const inboxProjectionDefinition = {
  apply(state: InboxState, event) {
    if (event.type !== 'agent/inbox/spliced') return state
    const splice = event.data
-    const next = state[splice.target].toSpliced(
-      splice.start,
-      splice.removedCount ?? 0,
-      ...splice.inserted,
-    )
-    return splice.target === 'next-turn'
-      ? { 'next-turn': next, 'next-step': state['next-step'] }
-      : { 'next-turn': state['next-turn'], 'next-step': next }
+    try {
+      const inbox = state[splice.target]
+      const removedCount = splice.removedCount ?? 0
+      if (!Number.isSafeInteger(splice.start) || splice.start < 0 || splice.start > inbox.length
+        || !Number.isSafeInteger(removedCount) || removedCount < 0
+        || splice.start + removedCount > inbox.length) {
+        throw new Error('invalid inbox splice')
+      }
+      const next = inbox.toSpliced(splice.start, removedCount, ...splice.inserted)
+      const ids = new Set<string>()
+      for (const message of splice.target === 'next-turn'
+        ? [...next, ...state['next-step']]
+        : [...state['next-turn'], ...next]) {
+        if (ids.has(message.id)) throw new Error(`message "${message.id}" is already pending`)
+        ids.add(message.id)
+      }
+      return splice.target === 'next-turn'
+        ? { 'next-turn': next, 'next-step': state['next-step'] }
+        : { 'next-turn': state['next-turn'], 'next-step': next }
+    } catch (error: unknown) {
+      throw new Error(`invalid persisted inbox splice at session seq ${event.seq}`, { cause: error })
+    }
  },
  wire: {
-    viewSchema: inboxProjectionSchema,
-    view: (state: InboxState) => state,
+    // The wire value is the fold state itself: every pending message already
+    // round-trips the session log as lossless JSON. Only the static type
+    // narrows to the JSON-safe projection table entry.
+    viewSchema: inboxProjectionSchema as unknown as z.ZodType<InboxWireState>,
+    view: (state: InboxState) => state as unknown as InboxWireState,
  },
  stateVersion: 1,
 } satisfies ProjectionDefinition<'inbox', InboxState>
-
-declare module '@deepseek-ai/dsh-session-projection/types' {
-  interface SessionProjectionMap {
-    /** Pending agent input reconstructed from durable inbox splices. */
-    inbox: InboxState
-  }
-}
@@ -54,7 +54,51 @@ async function inboxAgent(rawId: string): Promise<{ ctx: Context; session: Sessi
  return { ctx, session, agent }
 }

+async function reconstructPersistedInbox(
+  rawId: string,
+  populate: (session: Session) => void,
+): Promise<Error> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  const session = ctx.sessions.create(SessionId(rawId))
+  populate(session)
+  await ctx.plugin(SessionProjectionRegistry)
+  await ctx.plugin(AgentRegistry)
+  try {
+    ctx.sessionProjections.stateOf(session, 'inbox')
+  } catch (error: unknown) {
+    if (error instanceof Error) return error
+    throw error
+  }
+  throw new Error('persisted inbox reconstruction unexpectedly succeeded')
+}
+
 describe('Inbox', () => {
+  it('rejects invalid durable coordinates and duplicate identities during reconstruction', async () => {
+    const outOfRange = await reconstructPersistedInbox('invalid-inbox-range', (session) => {
+      session.append('agent/inbox/spliced', {
+        target: 'next-turn', start: 0, removedCount: 1, inserted: [],
+      })
+    })
+    expect(outOfRange.message).toBe('invalid persisted inbox splice at session seq 0')
+    expect((outOfRange.cause as Error).message).toBe('invalid inbox splice')
+
+    const pending = createUserMessage({
+      content: [{ type: 'text', text: 'duplicate' }],
+      source: { kind: 'user' },
+    })
+    const duplicate = await reconstructPersistedInbox('invalid-inbox-duplicate', (session) => {
+      session.append('agent/inbox/spliced', {
+        target: 'next-turn', start: 0, inserted: [pending],
+      })
+      session.append('agent/inbox/spliced', {
+        target: 'next-step', start: 0, inserted: [pending],
+      })
+    })
+    expect(duplicate.message).toBe('invalid persisted inbox splice at session seq 1')
+    expect((duplicate.cause as Error).message).toBe(`message "${pending.id}" is already pending`)
+  })
+
  it('projects inherited Inbox events in a forked session', async () => {
    const { ctx, session: parent, agent: parentAgent } = await inboxAgent('inbox-fork-parent')
    const inherited = createUserMessage({