PR #75 user-message-event：让事件流补上「用户自己这一轮」

figuretu/eyrie · main...feat/user-message-event @ 2af4015 · 2026-06-16 · 自包含，读完即弃

写法说明：本文按「逐跳走读」展开——每个机制都给真实代码片段（取自分支、经裁剪，青色斜体注释为解读所加，灰色斜体是源码原注释的保留或意译），每段代码标注所在文件。旅程结尾有一张「排查路标」：将来出问题时，症状对应去哪个文件看哪个函数。

1TL;DR

agent.events 是客户端重建一个会话时间线的唯一真相源——断线重连后，前端靠重放这串事件把对话画回来。但在此 PR 之前，这串事件里从来没有「用户说了什么」这一条：用户输入只被写进 runs.inputText 这一列（落库、但不进事件流），于是前端时间线里第一条业务事件是 run.started，用户自己的消息气泡无从渲染——这正是前端报上来的 Blocker。

本 PR 加一种事件类型 message.user：startTurn 在每一轮里把用户这轮输入作为一条事件持久化 + 广播，排在 run.started 之前，并且即使这一轮启动失败也保留。非测试改动只有 82 行（1 个事件类型 + 1 个 encoder + 1 处 emit + 同步 DB 的 CHECK 约束）；其余 221 行全是把这些行为钉死的测试。不动传输层、不动 client SDK、不出独立版本迁移脚本。

2形状先行

下潜前先认两个新形状，旅程里会反复用到。

第一，事件联合里多了一个成员。注意它身上没有 role 字段——role 是落库时由 encoder 钉成 'user' 的（见 A.3），联合成员本身只携带 text 和可选的 parts：

export type AgentProviderEvent = AgentProviderEventBase &
  (
    // ...assistant 流式增量、工具、审批等既有成员...
    | { type: 'message.delta'; role: 'assistant' | 'reasoning'; text: string; itemId?: string }
    | { type: 'message.user'; text: string; parts?: AgentInputPart[] }   // 新增：用户这一轮
    // ...
  )

第二，一个被提升为导出的类型守卫 allTextParts。它回答一个很窄的问题：「这串 parts 是不是全部都是纯文本块？」——这是本期唯一安全落库的 parts 形状（为什么这么窄见 A.2）。它被 service 和 codec 两处共用，所以从局部拷贝提成了共享导出：

export type AgentInputPart =
  | { type: 'text'; text: string }
  | { type: 'image'; path: string; mimeType: string }   // 注意 image 块带宿主 path

// 把一串 parts 收窄成「只含 text 块」，这是今天唯一安全持久化到用户事件上的形状。
export function allTextParts(
  parts: AgentInputPart[] | undefined,
): parts is Array<{ type: 'text'; text: string }> {
  return parts !== undefined && parts.length > 0 && parts.every((part) => part.type === 'text')
}

3旅程：一次 startTurn 如何把用户输入写进事件流

这是本 PR 唯一的旅程，也是它的全部价值所在。跟着一次 startTurn('session-1', { text: 'hello' }) 走一遍：从落点、到安全闸、到编码进库、到断线重放回原样。走通后你会知道用户气泡是从哪一行冒出来的、顺序由谁保证、图片为什么不进流。

A.1落点：为什么 emit 必须在 runner 之前

先摆清两件要同时满足的事：① message.user 的序号必须小于本轮任何 provider 事件，否则前端时间线里用户气泡会排在助手回复后面；② 这一轮哪怕启动失败，用户消息也得留痕——失败的 run 也该有一条诚实的时间线，告诉你「用户确实问了，但没跑起来」。

这两条共同把 emit 逼到了一个很具体的位置：在 beginTurn 拿到 run 之后、在 getOrCreate 构建 runner 之前，且整段仍在原有 try 内（emit 或构建任一抛错都走 terminateRun）。sink 的取法是「有活的就复用，没有就临时造一个 service-owned 的」——因为 runner 还没建，可能根本没有 live sink：

const run = await this.repo.beginTurn(sessionId, {
  id: createId(), inputText: normalized.text, startedAt: Date.now(),
})

try {
  const liveHandle = this.runnerManager.get(sessionId)
  const userEventSink =
    liveHandle?.sink ??
    new SessionSink(session.id, session.providerId, this.repo, this.broadcaster)  // 没活 runner 就临时造一个 sink
  userEventSink.setCurrentRun(run.id)
  await userEventSink.emit(buildUserMessageEvent(normalized))   // ← 必须 await 完，才保证 seq 在 run.started 之前

  const handle = await this.runnerManager.getOrCreate(/* ... */)  // 构建 runner（可能抛错）
  handle.sink.setCurrentRun(run.id)
  await handle.runner.startTurn(normalized)
} catch (err) {
  await this.repo.terminateRun(run.id, 'startTurnFailed')   // 此时 message.user 已落库，run 标 failed
  throw err
}

这处落点和最初计划不同（计划是放在 getOrCreate 之后走 handle.sink）。改的原因是个真实的丢消息缺口，见第 4 节偏差 ②。把两版并排看：

构造事件的 helper 很短，但它替整条旅程做了关键裁剪——parts 只在「全 text」时才挂上去：

function buildUserMessageEvent(
  input: AgentInput,
): Extract<AgentProviderEvent, { type: 'message.user' }> {
  const parts = allTextParts(input.parts) ? input.parts : undefined   // 非全 text 一律置 undefined
  return {
    type: AgentEventType.MessageUser,
    text: input.text,
    ...(parts ? { parts } : {}),   // 没有就连 key 都不放，而不是放个空数组
  }
}

A.2安全闸：为什么图片输入不进 parts

这是本 PR 最容易被忽略、却最有设计判断的一处。用户输入可能是图文混合（AgentInputPart 里 image 块带一个 path——那是宿主机上的本地文件路径）。如果原样把 parts 落进持久事件，等于把宿主 path 焊进了一份会被断线重放、会被前端消费的契约里。

本期的处理是：text 永远在（图文混合时由 normalizeInput 用 [image: image/png] 这样的占位拼出诚实回显），但 parts 只在全部都是 text 块时才落库；只要掺了一个 image/ref 块，整组 parts 省略——而不是给消费者一个「缺了图片的半截 parts」。text 在这种情况下就是权威回显。

const text = input.parts
  .map((part) => (part.type === 'text' ? part.text : `[image: ${part.mimeType}]`))  // 图片只露 mime，不露 path
  .join('\n')
return { text, parts: input.parts }

A.3编码进库：role、不脱敏、与 input.requested 的反向

事件落库走 SessionSink.emit：它先 encodeEvent 把联合成员翻译成行字段，再 appendEvent 在一个串行化事务里分配单调的 seq/sessionSeq 并插入，最后只广播已提交的那一行。这条链路是既有的（本 PR 没碰 sink），新增的只是 message.user 的 encoder。

encoder 干三件事，每件都有判断：role 钉成 'user'（落到索引列，replay 时再读回）；parts 这里再过一次 allTextParts（双保险，即便有人绕过 helper 直接 emit）；rawJson 置 null——这点和 input.requested 正好相反：

function encodeMessageUser(providerId, event, _context, _rawJson): EncodedEvent {
  return eventResult(providerId, {
    role: 'user',
    payload: { text: event.text, parts: allTextParts(event.parts) ? event.parts : undefined },
    // 与 input.requested 脱敏其 raw body 不同：用户这一轮不含 provider 敏感数据，
    // 所以 text 原样持久化作为时间线内容。image ref 在 upload-ref 契约 opaque 之前一律不带——
    // 混合 parts 否则会泄漏宿主路径或半截块，所以 parts 仅在每块都是 text 时才持久化。
    rawJson: null,
  })
}

这个 encoder 必须被登记进派发表 encodeByType。这张表带 satisfies Record<AgentEventType, EventEncoder>——也就是说，加了新事件类型却忘了登记 encoder，编译期就会红，不会留到运行时：

const encodeByType = {
  // ...
  [AgentEventType.MessageCompleted]: encodeMessageCompleted,
  [AgentEventType.MessageUser]: encodeMessageUser,   // 漏了这行，satisfies 会让 tsc 报缺键
  [AgentEventType.ToolStarted]: encodeToolStarted,
  // ...
}

A.4重放：DB 这道 CHECK 闸，与解码回事件

事件行真正落库前还要过一道 SQLite CHECK 约束——agent_events 表对 type 和 role 两列都枚举了允许值。这意味着新增 message.user / role='user' 不只是 TS 层的事，DB schema 不放行就会在 INSERT 时直接撞约束。所以本 PR 在三个层同步加值：Drizzle 表定义、baseline SQL、以及 Drizzle 的快照 JSON：

// type 白名单加一个成员：
'message.completed',
'message.user',          // ← type CHECK 三处都加
'tool.started',

// role 白名单从 (assistant, reasoning) 扩到含 user：
roleValid: check(
  'agent_events_role_valid',
  sql`${table.role} IS NULL OR ${table.role} IN ('assistant', 'reasoning', 'user')`,  // ← 多了 'user'
),

重放侧不需要为新类型写任何特判——这是既有 codec 设计的红利。decodeEventRow 是通用的：role 从行的索引列读回（readIndexedPayloadFields），text/parts 从 payload JSON 通用展开，再拼出事件联合。于是断线后重放出来的就是 { type:'message.user', role:'user', text, parts? }，和落库前一致。

event: {
  runId: row.runId, sessionId: row.sessionId, providerId: row.providerId,
  type: row.type,
  ...readIndexedPayloadFields(row),    // role 从索引列读回，message.user 在此自动拿回 'user'
  ...omitRowOwnedPayloadFields(payload),  // text / parts 从 payload 通用展开
  raw,
} as AgentEvent

4计划 vs 实现的偏差

照 spec/plan 直接做成的部分（加事件类型、加 encoder、加 emit）不在这里——那些读者从计划就能预期。这一节只收实质偏差：计划写错或漏判、实现期被迫改道的地方，这些才是认知裂缝。

5心智模型补丁

6新词表

7测试与风险地图

纯事实陈述。本 PR 测试占比 ~73%，行为钉得相当密。

有兜底的（测试钉住的行为）

• 🟢 事件词表长度：agent-contract.test.ts 把 vocabulary 从 18 改 19 并断言含 'message.user'——加错/漏加类型必红。
• 🟢 编解码 round-trip：agent-event-codec.test.ts 钉死 role 落 'user' 索引列、rawJson 为 null、全 text parts 入 payload、含 image 时 payload 省略 parts（且断言落库 JSON 不含 /cache/secret.png 这种宿主 path）。
• 🟢 emit 在 runner 之前：agent-service-methods.test.ts 用记录式 repo 断言 append 顺序——message.user 先于 runner.startTurn。
• 🟢 两条失败路径都留痕：startTurn 抛错（runner 拒绝）和 runner 构建失败（FakeRunnerManager(null)）两个用例，都断言 message.user 已 append、随后 terminate:startTurnFailed。
• 🟢 真 SQLite 重放顺序：agent-integration.test.ts 断言 sessionSeq [1,2,3]、首条是 message.user、次条 run.started、三条 run.completed。
• 🟢 DB 约束仍拒非法 role：db.test.ts 原来用非法 role='user' 测拒绝路径，现 user 合法了，改用仍非法的 'system' 继续覆盖。
• 🟡 真 Claude e2e：session-loop-e2e.test.ts 给两轮各加一句「首事件是 message.user」，但整文件 skipIf(!EYRIE_CLAUDE_E2E)——默认 CI 不跑，靠上面的 integration 测试覆盖重放。

薄冰（无测试 / 已知遗留）

• 🟠 image/ref part 的正式形状未定：本期靠「省略 parts + text 占位」绕开，真正的图片重放要等 Slice 2 的 upload-ref 闭环。前端今天拿不到结构化图片块。
• 🟡 前端拿不到强类型：传输层 @eyrie/api 的 event 仍是 unknown（本 PR 没碰），前端要靠运行时判别 type==='message.user'，没有编译期窄化。导出传输安全的事件联合是独立后续。

8验收提示 + 覆盖声明

别被这些吓到

• 2264 行的快照 JSON 没动几个字：meta/0000_snapshot.json 的 diff 只是两条 CHECK value 字符串各加了一个枚举值——不是重写 schema。
• session-loop-e2e.test.ts 默认不跑：skipIf(!EYRIE_CLAUDE_E2E) 是设计，不是漏测；重放由 integration 测试在真 SQLite 上覆盖。
• 传输层/SDK 没改是有意的：本 PR 范围只到 daemon 契约；前端强类型是独立 slice，不在此。
• commit 只有 1 个：原实现曾误叠在 discovery 分支栈顶，已 cherry-pick 摘成独立分支，所以这里看到的是干净单 commit。

覆盖声明

本报告由主理 agent 全量精读：12 个改动文件的 diff 全部过目，并亲自 Read 了 service.ts（startTurn + normalizeInput + buildUserMessageEvent）、session-sink.ts（emit 全文）、event-codec.ts（encodeEvent / decodeEventRow / eventResult / 索引字段读写 / encodeMessageUser）、types.ts（联合 + allTextParts）原文裁剪代码，未经 subagent 转述。DB 三层改动经 diff 核对一致。无抽样、无略读。报告为一次性理解辅助，不维护、不作真相源。

figuretu/eyrie · feat/user-message-event @ 2af4015 · +284/−19 · 12 文件 · ~73% 测试 · 本报告阅后即弃。