Coding Agent Harness 拆解 · 消息与事件流：一条 Event 流驱动任意前端（三）

本文是《Coding Agent Harness 拆解》系列的一篇。整个系列逐个拆开一个真实在用的终端 AI 编程 agent，也就是 codebot，以及它底下的执行内核 agentcore（均为 Go 实现），看清一个"能每天用的 coding agent"到底由哪些子系统拼成。本篇聚焦一个最基础、也最容易被绕晕的机制：内核向上只暴露一条事件流。它由哪些事件组成、流式片段怎么一点点累积、Agent 如何把事件同步分发给监听者、Session 又如何把内核 Event 再加工成产品自己的 SessionEvent 往前端抛。

内核的全部对外输出，就是一个 chan Event。agentcore/doc.go 说得很直接：流式文本、工具执行、重试、最终总结，全都走这一条流。谁想驱动前端（TUI、web、日志），都只需要消费这一个 channel，而内核完全不知道对面是谁。下面就把这条流从产生到落屏拆开看。

消息不是字符串

事件里流动的是结构化的消息，不是纯文本。这里直接落到真实类型上（agentcore/message.go）。

内核区分两个层次：

AgentMessage 是一个接口，只要求 5 个方法：GetRole / GetTimestamp / TextContent / ThinkingContent / HasToolCalls。它是"应用层消息"的最小契约：harness 可以塞进自定义类型（状态提示、UI 标记），让它们随上下文流水线一起走，但在发给模型前被 DefaultConvertToLLM 过滤掉。
Message 是内核自带的 LLM 级实现。它的内容是一串 ContentBlock，不是一个 string。每个块带 Type，可能是 ContentText、ContentThinking、ContentToolCall、ContentImage 之一。TextContent() 就是把所有 text 块拼起来，ThinkingContent() 拼所有思考块。

// ContentBlock is a tagged union for message content.
type ContentBlock struct {
    Type     ContentType `json:"type"`
    Text     string      `json:"text,omitempty"`
    Thinking string      `json:"thinking,omitempty"`
    ToolCall *ToolCall   `json:"tool_call,omitempty"`
    Image    *ImageData  `json:"image,omitempty"`
    ToolName string      `json:"tool_name,omitempty"`
}

为什么用"块"来装内容？因为一条助手消息里可能同时有思考、有文本、有若干个 tool call，还挂着 Usage（token 用量）和 StopReason。字符串表达不了这些，而下游的渲染、审批、持久化全要精确读到它们。

一条流：事件类型全景

agentcore/event.go 用一个 EventType 常量枚举描述了一次运行会发生的所有生命周期信号。按"套娃"层级理解最清楚：

flowchart TB
  Start["EventAgentStart —— 一次 run 开始"]
  Start --> Turn["EventTurnStart —— 一轮开始"]
  Turn --> MS["EventMessageStart"]
  MS --> MU["EventMessageUpdate（流式增量，可 N 次）"]
  MU --> ME["EventMessageEnd（消息成型）"]
  ME --> TES["EventToolExecStart"]
  TES --> TEU["EventToolExecUpdate（preview / progress）"]
  TEU --> TEE["EventToolExecEnd"]
  TEE --> MR["EventModelResponse —— 一次模型调用收尾"]
  MR -->|"继续下一轮"| Turn
  MR -->|"停止"| End["EventAgentEnd（带 RunSummary）"]
  Retry["EventRetry / EventError 可在任意点插入"]

| 事件 | 触发时机 | 关键随附字段 | | --- | --- | --- | | EventAgentStart / EventAgentEnd | 整个 run 的起止 | End 带 NewMessages、Summary *RunSummary | | EventTurnStart | 每一轮循环开始 | — | | EventMessageStart / Update / End | 一条消息的流式生命周期 | Message、Delta、DeltaKind | | EventToolExecStart / Update / End | 一次工具执行 | ToolID、Tool、Args、Result、IsError | | EventModelResponse | 每次模型调用完成（含它触发的工具） | Message、ToolResults []ToolResult | | EventRetry | 一次重试前 | RetryInfo *RetryInfo | | EventError | 出错（随后必跟一个 EventAgentEnd） | Err error |

有两个容易踩的点，源码注释专门写了：

EventModelResponse 不等于"一次用户对话"。 一次 model 调用产生一个 EventModelResponse：steering 插入、length 恢复都会在同一个 run 里多产生几个。它是"一次模型调用收尾"的信号，不是"本轮对话结束"。
没有独立的 EventTurnEnd 常量。 一轮的工具结果通过 EventModelResponse 的 ToolResults 字段带出来，EventAgentEnd 才是整个 run 的终点，并附上事实性的 RunSummary（TurnCount / ToolCalls / ToolErrors / EndReason）。

一个结构体，装下所有事件

Event 是一个典型的联合结构体：字段很多，但每种 Type 只填其中几个。这是刻意的设计：一个 channel 类型要能承载所有语义，就得让 Event 成为所有可能载荷的超集。

type Event struct {
    Type        EventType
    Message     AgentMessage    // message_* 事件
    Delta       string          // message_update 的文本增量
    DeltaKind   DeltaKind       // 这个增量是什么：text / thinking / toolcall
    ToolID      string          // tool_exec_* 的调用 ID
    Tool        string          // 工具名
    Args        json.RawMessage // tool_exec_start/update 的入参
    Result      json.RawMessage // tool_exec_end / preview 的结果
    Progress    *ProgressPayload
    UpdateKind  ToolExecUpdateKind
    IsError     bool
    ToolResults []ToolResult   // model_response：本轮所有工具结果
    Err         error          // error 事件
    NewMessages []AgentMessage // agent_end：本次 loop 新增的消息
    RetryInfo   *RetryInfo     // retry 事件
    Summary     *RunSummary    // agent_end：事实性总结
}

::: tip 事件带背压，不是“最大努力广播” agentcore/event.go 里的 eventSink.emit 有讲究：run 存活期间，channel 满了就阻塞，这是背压，不丢事件；只有当 run 的 context 已经取消，投递才降级为"尽力而为"，避免一个被遗弃的 channel 把 loop goroutine 泄漏掉。也就是说，只要你还在正常消费，就不会漏掉任何一个事件。 :::

流式增量如何累积

最容易想错的一环是流式。直觉以为 message_update 的载荷是"这次新增的几个字"，于是消费者要自己把它们一段段拼起来。真实设计更省心：看 agentcore/loop.go 的 callLLMStream：

case StreamEventTextDelta, StreamEventThinkingDelta, StreamEventToolCallDelta:
    partial = ev.Message // partial is the FULL message-so-far, cumulative
    ...
    var dk DeltaKind
    switch ev.Type {
    case StreamEventThinkingDelta: dk = DeltaThinking
    case StreamEventToolCallDelta: dk = DeltaToolCall
    }
    sink.emit(Event{Type: EventMessageUpdate, Message: partial, Delta: ev.Delta, DeltaKind: dk})

关键在于：Message 字段挂的是累积到当前为止的完整 partial 消息，而 Delta 只是这一步的增量。两种信息同时给出：

想"覆盖式"渲染的消费者，直接读 ev.Message.TextContent()，拿到的就是完整快照，不用自己拼。
想"追加式"处理的消费者，读 ev.Delta 拿增量。
DeltaKind 告诉你这段增量是正文（DeltaText，空串）、思考（DeltaThinking）还是工具调用参数 JSON（DeltaToolCall），从而分流渲染。

一条消息的流式序列因此非常规整：EventMessageStart（首个片段到达时才发）→ 若干 EventMessageUpdate → EventMessageEnd（StreamEventDone 时发，带最终成型的 Message）。这套"先 Start、增量 Update、末尾 End"的三段式，是后面所有 UI 累积逻辑的地基。

Agent 如何同步分发

Agent（agentcore/agent.go）是内核对循环的有状态包装。它自己就是循环事件的第一个消费者：consumeLoop 一边读事件更新内部状态，一边把每个事件转发给所有 Subscribe 进来的监听者。

先看它维护的三块流式状态，全在 consumeLoop 的 switch 里更新：

case EventMessageStart, EventMessageUpdate:
    partial = ev.Message
    a.streamMessage = ev.Message          // 外部可见的“正在流式的消息”

case EventMessageEnd:
    a.streamMessage = nil
    if ev.Message != nil {
        a.messages = append(a.messages, ev.Message)   // 落进历史
        if msg, ok := ev.Message.(Message); ok && msg.Usage != nil {
            a.totalUsage.Add(msg.Usage)                // 累加 token 用量
        }
    }

case EventToolExecStart:
    a.pendingToolCalls[ev.ToolID] = struct{}{}         // 记在飞的工具
case EventToolExecEnd:
    delete(a.pendingToolCalls, ev.ToolID)

于是任何时候调用 a.State()，都能拿到一致的快照：StreamMessage（正在流的半成品，空闲时为 nil）、PendingToolCalls（正在执行的工具 ID 集合）、TotalUsage（累计用量）。EventError 会丢弃残缺的 partial 并清掉 streamMessage，避免 State() 暴露一个永远不会完成的半成品，这些兜底都写在同一个 switch 里。

分发的契约是这套设计里最该记住的一点，Subscribe 的注释把它列为稳定 API 保证：

sequenceDiagram
    participant Runner as AgentLoop (goroutine)
    participant CL as Agent.consumeLoop
    participant L1 as 监听者1 (注册在前)
    participant L2 as 监听者2 (注册在后)

    Runner-->>CL: Event
    CL->>CL: 加锁更新内部状态
    CL->>CL: 复制 listeners 快照后解锁
    CL->>L1: fn(ev) 同步调用
    L1-->>CL: 返回后
    CL->>L2: fn(ev) 同步调用

三条规则：监听者被同步调用、按注册顺序、在同一个事件消费 goroutine 上。注册在前的监听者总是先看到每个事件：顺序敏感的消费者（比如必须先否决的预算哨兵）可以依赖它。代价也直白：某个监听者慢，会拖慢后面所有监听者、并给循环的事件 channel 造成背压。所以重活要自己另开 goroutine。consumeLoop 里的做法是先在锁内复制一份 listeners 快照，解锁后再逐个回调，回调期间不持锁。

Session 把 Event 再加工成 SessionEvent

codebot 的 internal/agent.Session 就是上面那个"注册在前的监听者"：它 Subscribe 到内核 Agent，回调函数是 handleAgentEvent（internal/agent/session_state.go）。它做两件事：先按事件类型驱动产品副作用，再把内核 Event 包一层往 UI 抛。

func (s *Session) handleAgentEvent(ev agentcore.Event) {
    s.runtime.handleEvent(ev)
    // ... 按 ev.Type 分派产品逻辑 ...
    s.emit(SessionEvent{
        Type:       SEAgentEvent,
        AgentEvent: &ev,          // 内核事件被透明包进 SessionEvent
    })
}

SessionEvent（internal/agent/events.go）是对内核事件的扩展：当 Type == SEAgentEvent 时，AgentEvent 字段就是原封不动的内核 Event；此外它还定义了一批纯产品层事件：SEAutoCompactionStart/End（自动压缩）、SEAutoRetryStart/End（重试）、SEModelChanged、SEGoalUpdated、SEError 等。内核不知道"压缩""目标"这些概念，它们是 harness 在事件流上长出来的。

handleAgentEvent 里几个典型的"借事件做副作用"：

| 监听到的内核事件 | Session 顺手做了什么 | | --- | --- | | EventMessageStart（assistant） | 记 lastAssistantStart，用于算这轮延迟 | | EventMessageEnd | 持久化消息、写 llm_call 观测记录、自动命名会话、抽取会话记忆 | | EventError（配额类） | 把目标标记为 usage-limited | | EventRetry | 转成 SEAutoRetryStart 通知 UI | | EventAgentEnd | 收尾遥测、flush 待落盘消息、跑 hooks、清技能增量 |

这就是架构上那层"缝合层"的真面目：内核的事件流进 Session，codebot 的持久化、压缩、遥测、hooks 全都挂在这条流上，再以 SessionEvent 的形式统一流向前端。

TUI 如何累积并落屏

最终消费者之一是 TUI。internal/ui/tui/events.go 的 HandleAgentEvent 订阅的是 SessionEvent 里透传出来的内核事件，它的职责是：把"进行中"的内容画在活动区（View()），把"已完成"的内容打印进滚动区（scrollback）。

流式文本的累积就落在这里，注意它怎么用上一节说的"完整快照 + 增量"双份信息：

case agentcore.EventMessageUpdate:
    if ev.Message != nil {
        if text := ev.Message.TextContent(); text != "" {
            m.Streaming.Reset()          // 有完整快照就整段覆盖
            m.Streaming.WriteString(text)
        }
        ...
    } else if ev.Delta != "" {
        m.Streaming.WriteString(ev.Delta) // 没快照才退化为追加增量
    }

因为 partial 是累积的，TUI 直接 Reset 再写整段最省事、也不会因丢过某个 delta 而错位；只有 Message 为 nil 时才退回按 Delta 追加。"落屏"发生在 EventMessageEnd：此时把 Streaming / Thinking 缓冲清空，用最终的 TextContent() / ThinkingContent() 渲染成一个块 printBlock 进 scrollback，并顺手把 Usage 累加进 RunStats。

工具事件是另一套缓冲。TUI 在 EventToolExecStart 时把工具头缓存进 ToolHeaders、结果缓冲进 ToolOutputBuf，到 EventToolExecEnd 才把"头 + 结果"作为一个整块打印，这样并行工具不会交错。EventToolExecUpdate 则按 UpdateKind 分流：ToolExecUpdatePreview 画 diff 预览，ToolExecUpdateProgress 把 ProgressPayload 里的 ProgressToolDelta / ProgressThinking 累积进子 agent 的流式缓冲（工具执行管线本身另有专篇展开）。

::: tip End 事件为什么"丢" Args EventToolExecEnd 上 agentcore 不再带 Args（只在 Start 带）。所以 TUI 需要跨事件保存的每-调用状态（比如 plan 文件标记、错误时要复现的命令），都得在 EventToolExecStart 时存进 PendingTools 之类的 map，End 时再取。这是读 events.go 时最容易困惑的一处，本质是"事件是瘦的，状态在消费者侧攒"。 :::

一个增量从模型到屏幕

把三层串起来，一个文本片段的完整旅程是这样：

sequenceDiagram
    participant Model as ChatModel
    participant Stream as callLLMStream
    participant Agent as Agent.consumeLoop
    participant Session as Session.handleAgentEvent
    participant TUI as TUI.HandleAgentEvent

    Model-->>Stream: StreamEventTextDelta("好")
    Stream->>Agent: Event{MessageUpdate, Message:partial, Delta:"好"}
    Agent->>Agent: streamMessage = partial
    Agent->>Session: 同步回调(ev)
    Session->>TUI: SessionEvent{SEAgentEvent, AgentEvent:ev}
    TUI->>TUI: Streaming.Reset + 写入完整快照
    Note over Model,TUI: 循环到 StreamEventDone
    Model-->>Stream: StreamEventDone
    Stream->>Agent: Event{MessageEnd, Message:final}
    Agent->>Agent: messages append + totalUsage.Add
    Agent->>Session: 同步回调
    Session->>Session: 持久化消息 / 写 llm_call
    Session->>TUI: SEAgentEvent
    TUI->>TUI: printBlock 落进 scrollback

同一条流，三层各取所需：内核攒对话状态，Session 攒产品状态，TUI 攒渲染缓冲。谁都没有"反向查询"对方，全靠这一条单向事件流对齐。

三问回顾

这个机制解决了什么痛点？ 让内核用唯一一条事件流表达全部生命周期信号，任何前端只需消费一个 channel 就能驱动；结构化消息 + 累积式增量让消费者既能覆盖式渲染也能追加式处理，还不必自己拼字符串。
如果没有它，会在哪里崩？ 若用多路回调或裸文本，顺序、背压、状态一致性会散落各处：UI 看不到流式过程、State() 会暴露永不完成的半成品、慢监听者悄悄拖垮循环、工具结果与消息错位。
它和前后怎么接上？ 向上，Agent 把事件同步分发给 Session，Session 再加工成 SessionEvent 喂给 TUI；向下，事件由 AgentLoop 在跨越 LLM 边界和工具边界时逐点 emit。事件从哪 emit、工具事件那条支流、谁消费 SessionEvent，分别是下面三篇的主题。