gobrave是如何做到GET /events?after=sequence 原生增量回放

Searched for regex events|GetTaskEvents|agentEventsQuery (**/gobrave/internal/handler/**), 15 results

Searched for regex /events|after= (**/gobrave/**), 50 results

Read , lines 770 to 830

Searched for regex agent/task/events|after (**/brave-ui/src/**), 35 results

Searched for regex getAgentTaskEventsApi (**/brave-ui/src/**), 9 results

Searched for regex attachTaskStream\(|GetAgentTask\(|GetTask\b|parseAgentID (**/gobrave/internal/handler/agent.go), 20 results

Read , lines 240 to 360

Read , lines 1 to 120

Read , lines 120 to 240

Searched for regex interface AgentEventItem (**/brave-ui/src/api/agent/index.ts), 1 result

Read , lines 120 to 150

Read , lines 210 to 240

一、机制概览

「原生增量回放」不是靠前端缓存或时间戳，而是事件表自带一个 per-task 单调递增的 sequence，读取端只传「我已经收到的最大 sequence」，服务端返回 sequence > after 的行。

二、写入端：sequence 是怎么产生的

1. 数据结构（internal/agent/event.go）

type AgentEvent struct {
    ID        int64          // 雪花 ID，永远唯一
    TaskID    int64
    Sequence  int64          // per-task 单调递增，增量游标
    Type      AgentEventType
    Payload   any            `gorm:"serializer:json"`
    CreatedAt time.Time
}

GORM tag 上有关键的复合索引：

TaskID   ... `gorm:"index:idx_agent_events_task_seq,priority:1"`
Sequence ... `gorm:"index:idx_agent_events_task_seq,priority:2"`

即 (task_id, sequence)，正好服务 WHERE task_id=? AND sequence>? ORDER BY sequence 这个查询。

2. 单一出口 AgentService.publish

e := &AgentEvent{
    ID:        utils.GenerateID(), // 广播事件（不落库的 delta）也必须分配
    TaskID:    taskID,
    Type:      typ,
    CreatedAt: time.Now(),
}
if persist && s.events != nil {
    _ = s.events.Append(ctx, e)
}
s.bus.Publish(ctx, *e)

注意 ID 在广播前就分配了：delta 事件不落库、Sequence 恒为 0，如果 ID 也为 0，前端按 id:sequence 去重会把同一任务的全部 delta 判成重复而丢弃（代码注释里记录了这个坑）。

3. Append：事务内分配 sequence（internal/agent/repository/event.go）

return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
    var maxSeq int64
    tx.Model(&agent.AgentEvent{}).
        Where("task_id = ?", event.TaskID).
        Select("COALESCE(MAX(sequence), 0)").
        Scan(&maxSeq)
    event.Sequence = maxSeq + 1
    return tx.Create(event).Error
})

因此 sequence 是稠密、无洞、从 1 开始的。只有「需要进入时间线的事件」才走这条路（isBlockEvent 为真的块事件）；delta 事件不写库，也就没有 sequence。

三、读取端：一次 SQL 完成的增量

1. 路由（internal/router/router.go:466）

r.GET("/agent/task/events", handler.GetTaskEvents)

2. Handler（internal/handler/agent.go:789）

type agentEventsQuery struct {
    TaskID string `form:"task_id" binding:"required"` // 雪花 ID 以字符串传输
    After  int64  `form:"after"`                      // 上次收到的最大 sequence，缺省 0
}

func (h *AgentHandler) GetTaskEvents(c *gin.Context) {
    if _, ok := getCurrentUserID(c); !ok { return }   // 仅鉴权登录
    c.ShouldBindQuery(&q)
    taskID, _ := parseAgentID(q.TaskID)               // 字符串 → int64，避免 JS 精度丢失
    events, err := h.svc.GetEvents(c.Request.Context(), taskID, q.After)
    c.JSON(http.StatusOK, events)
}

3. 查询（repository/event.go:39）

r.db.WithContext(ctx).
    Where("task_id = ? AND sequence > ?", taskID, after).
    Order("sequence ASC").
    Find(&events)

只有两件事：range scan on 复合索引 + 服务端排序。 没有游标编码、没有 offset 漂移、没有时间戳歧义——这就是「原生」的含义。

客户端只要保存 max(lastEvent.sequence)，重连时传 after 即可，天然幂等：重复请求同一 after 返回同一集合，插入新事件也不会改变历史行的 sequence（append-only），不像 offset 分页会被并发写入挤歪。

四、前端如何把它和 WS 合起来（agent-task-stream.tsx）

关键在「先订阅、后回放、再合并」的水位线协议：

1. 先挂 WS：sseClient.onMessage 在 hydratedRef.current === false 时，把实时事件推进 liveBufferRef，不直接渲染。
2. 再 GET 回放：getAgentTaskEventsApi(taskId, 0) 拉全量历史。
3. 合并去重：历史 + 缓冲用 put() 以 `${id}:${sequence}` 为 key 去重，再按 created_at → sequence 稳定排序。
4. 同一 reducer 回放：合并后的数组逐条喂给 applyEvent，历史与实时走完全相同的代码路径，不存在两套渲染逻辑；回放结束后 closeOpen("assistant_final") 之类收口一下未闭合的 streaming 消息，置 hydratedRef = true。
5. 之后实时：新 WS 事件直接 applyEvent，seenRef 的 id:sequence 集合防止重复应用。

这个「缓冲 + 水位线 + 统一 reducer」是漂移最少的重连写法：既不丢 WS 早期事件（已缓冲），也不重复（去重键），还不依赖事件到达顺序（排序）。

五、设计上的取舍与现存问题

优点

• append-only + 稠密 sequence = 天然幂等、可断点续传，多端/刷新/重连无状态恢复
• 索引与查询一一对应，回放成本是 O(新增事件数)
• ID 与 sequence 双键去重，广播事件也有唯一 ID
• 时间线完全由事件日志驱动，而不是「从消息快照反推」

坑（值得注意）

1. delta 不可回放：text / reasoning_delta 只广播不落库，刷新后只能看到落库的 message / reasoning 完整块（设计如此，但意味着「逐字打字机效果」在回放时会跳变）。
2. MAX(sequence)+1 依赖单写者假设：注释写明「同一 task 的 Append 不会并发」。在多进程/多实例部署下，同一 task 若被两个实例写（例如恢复机制重复启动），会产生重复 sequence，且索引不是 unique，DB 不会拦。要真正安全需改 UPDATE ... RETURNING / 行锁 / redis INCR / 或把 (task_id, sequence) 设为唯一键并重试。
3. 无分页：GetTaskEvents 没有 limit，长任务的首次 after=0 会一次拉回全部落库事件。
4. 无归属校验：handler 只调了 getCurrentUserID(c) 拿用户 ID，但没有用它校验 task 是否属于该用户（GetEvents 直接按 taskID 查），存在 IDOR 风险；对比 GetTask 等接口才是按 userID 过滤的。
5. 无保留策略：agent_events 只追加不清理，需要 TTL / 归档。
6. 前端排序以 created_at 为主键、sequence 为次键：服务端已按 sequence 有序，前端排序理论上可能因时钟回拨造成与 sequence 不一致（实际影响很小，但纯函数式做法应只按 sequence 排）。