271 lines
12 KiB
Go
271 lines
12 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"log/slog"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript"
|
||
)
|
||
|
||
// 本文件实现「忙时新消息在回合内边界注入同一回合」的网关侧
|
||
// （design/15 §2.0/§2.1/§2.4/§5，裁决 13.4.5 目标态）。它是 loop.SteerSource
|
||
// 的生产实现，与 loop/steer.go 的调用点合成一条链：
|
||
//
|
||
// loop 在每个回合内边界（一批工具调用结束之后、下一次模型调用之前）
|
||
// → 本文件重读该会话转录，把已落库、且在本会话 pending 里的用户消息交出去
|
||
// → 循环把它们并入**同一回合**的下一份模型视图（绝不新开第二回合，I1）
|
||
// → 本文件把这些消息从 pending 移除并下发 steer_ack（I2：注入＝已消费）
|
||
//
|
||
// 候选集必须**同时**满足两个条件，缺一不可：
|
||
//
|
||
// - **在转录里**（history.Recent 重读）＝最新性与可去重：消息真的落库了，我们才
|
||
// 能拿到它的权威 id 与 created_at。模型消息本身没有 id 字段，因此「按
|
||
// created_at + id 去重」只能在持有转录的这一侧做（design/15 §2.4）。
|
||
// - **在 pending 里**＝恰好一次（I2）：只有本编排器排队过的消息才允许被本回合
|
||
// 接手；注入后立刻从 pending 移除，因此它既不会被第二个回合再答一遍，也不会丢。
|
||
// 由别的写入方（IM 适配器自有路径）落库、本编排器从未排队的消息**保持不注入**
|
||
// —— 那条路径正在自己作答它，注入会让同一条消息被答两次。
|
||
//
|
||
// 未接历史存储、读转录失败、消息尚未落库、或本回合的世代已被打断/新回合取代时一律
|
||
// 返回空（该消息留在 pending，照旧作为自己的回合被回答）：转向是增益，任何一处失败
|
||
// 都不该让用户已经在跑的回合失败，也绝不该让一条消息静默消失。
|
||
|
||
// steerLog 返回标记来源 gateway/session-steer 的 logger（design/15 转向链的日志
|
||
// 能按子功能定位，而不是混在 gateway 的裸日志里）。
|
||
func steerLog(o *Orchestrator) *slog.Logger {
|
||
return logger.Scoped(o.log, source, "session-steer")
|
||
}
|
||
|
||
// steerTurn 是一个在飞回合的转向游标。**每个回合一份**，绝不跨回合复用：它记录该
|
||
// 回合已吃进的消息 id（I2 的去重键）与回合世代号（被中断/被新回合取代后立即失效）。
|
||
//
|
||
// 它只在回合自己的 goroutine 上被调用（loop 的边界调用是同步的），因此自身的
|
||
// consumed 无需加锁；pending 的读写与世代校验则一律走编排器互斥锁。
|
||
type steerTurn struct {
|
||
orchestrator *Orchestrator
|
||
sessionID string
|
||
gen uint64
|
||
// turnID 是本路（＝本回合）的权威回合标识：`steer_ack` 的**帧壳**必须带上它
|
||
// （忙时并行规范草案 §2.5）。真并行下同一会话有两条交错的路，只按 sessionId
|
||
// 分桶的话客户端不知道这条消息被**哪一路**吃进，灰态也就配不上正确的
|
||
// `turn.end`。它在出生后不可变，因此这里存下来即可，无需在发帧时回查状态锁。
|
||
turnID string
|
||
consumed map[string]bool
|
||
}
|
||
|
||
// beginSteer 为一个刚启动的回合建一份转向游标。编排器把它挂到**该回合自己的 ctx**
|
||
// 上（loop.WithSteerSource），因此每个回合、每次续跑各有一份独立游标，上一个回合
|
||
// 已吃进的消息绝不会被下一个回合重复注入或漏掉。turnID 是该路的权威回合标识，
|
||
// 与 `turn_begin` 帧壳同值（客户端靠它把 steer_ack 配到正确的 `turn.end`）。
|
||
func (o *Orchestrator) beginSteer(sessionID string, gen uint64, turnID string) *steerTurn {
|
||
return &steerTurn{
|
||
orchestrator: o,
|
||
sessionID: sessionID,
|
||
gen: gen,
|
||
turnID: turnID,
|
||
consumed: make(map[string]bool),
|
||
}
|
||
}
|
||
|
||
// steerCandidate 是本次边界要注入的一条消息：它的 pending 队列序号（移除认领用）
|
||
// 与它已落库的转录行（id/created_at/内容/说话人归属都从行上来）。
|
||
type steerCandidate struct {
|
||
seq uint64
|
||
row history.Message
|
||
}
|
||
|
||
// Steer 是 loop.SteerSource 的实现（design/15 §2.4）：在一次回合内边界重读该会话
|
||
// 转录，返回本次要注入同一回合的用户消息，并把它们从 pending 移除（注入＝已消费）。
|
||
// 它绝不返回错误——失败一律退化为「本次边界没有可注入的消息」。
|
||
func (s *steerTurn) Steer(ctx context.Context, req loop.SteerRequest) []model.Message {
|
||
if s == nil || s.orchestrator == nil || ctx.Err() != nil {
|
||
return nil
|
||
}
|
||
sessionID := req.SessionID
|
||
if sessionID == "" {
|
||
sessionID = s.sessionID
|
||
}
|
||
o := s.orchestrator
|
||
reader, ok := o.history.(transcript.Reader)
|
||
if !ok {
|
||
return nil
|
||
}
|
||
pending := o.pendingSnapshot(sessionID, s.gen)
|
||
if len(pending) == 0 {
|
||
return nil
|
||
}
|
||
// 每个边界都重读转录：设计要求的「最新性＝下一个边界即生效」正是靠这次重读，
|
||
// 而不是靠回合开始时的那一次播种（design/15 §2.4）。
|
||
recent, err := reader.Recent(ctx, sessionID, transcript.ContextCount)
|
||
if err != nil {
|
||
steerLog(o).Debug("steer transcript read", "session", sessionID, "step", req.Step, "err", err)
|
||
return nil
|
||
}
|
||
candidates := s.match(pending, recent)
|
||
if len(candidates) == 0 {
|
||
return nil
|
||
}
|
||
// 认领失败（世代已被中断/被更新的顶层回合取代）时把这条消息留给那条新路径：
|
||
// 既不注入、也不发 steer_ack —— 否则它会被本回合与下一个回合各答一次（I2）。
|
||
if !o.consumePending(sessionID, s.gen, candidateSeqs(candidates)) {
|
||
return nil
|
||
}
|
||
// 被吃进的消息在落库那一刻还不知道属于哪一路（那时它还在排队），因此认领成功
|
||
// 之后立刻按本路补写 turn_id（两段式第二段，design/15 §2.5.9）：它与本路的
|
||
// 助手回复同值，客户端才能把灰态气泡与正确的回复配到一起。
|
||
o.assignTurnID(sessionID, s.turnID, steerRows(candidates))
|
||
out := make([]model.Message, 0, len(candidates))
|
||
for _, c := range candidates {
|
||
s.consumed[c.row.ID] = true
|
||
// steer_ack：该消息已被本回合吃进、尚未被正面回答（前端据此把它渲染成
|
||
// 灰态，design/15 §5）。它携带具体消息身份，因此不进 frameReachesAll 的
|
||
// 状态机信号白名单，落 fail-closed 默认档：只发 owner + admin。
|
||
//
|
||
// 帧壳必须带本路的 turnId（忙时并行规范草案 §2.5）：真并行下客户端要
|
||
// 靠它把这条消息配到**正确那一路**的 `turn.end`，否则灰态收不了口。
|
||
o.emitTurnState(sessionID, s.turnID, wsSteerAck, map[string]any{"messageId": c.row.ID, "atStep": req.Step})
|
||
out = append(out, transcript.Message(c.row))
|
||
}
|
||
steerLog(o).Info("pending steered", "session", sessionID, "step", req.Step, "messages", len(out))
|
||
return out
|
||
}
|
||
|
||
// match 把 pending 里的用户消息与转录行配对（design/15 §2.4 的「按 created_at + id
|
||
// 去重」）。它按 pending 的 FIFO 顺序（＝到达顺序，与转录的 created_at/rowid 顺序
|
||
// 一致）逐条找**最旧的、尚未被认领的**同行；找不到（尚未落库、或落库失败、或该行
|
||
// 已在本回合更早的边界被吃进）就跳过，让那条消息留在 pending 里等自己的回合。
|
||
//
|
||
// 用 (内容, 图片) 配对而不是 id 配对的原因：编排器不持有自己在 pending 路径上写入
|
||
// 的行的 id（写入在持久化钩子内部完成，本回合不该为了拿到 id 去改那条路径的契约）。
|
||
// 同一文本连发两条仍能正确配对——配对逐行认领、且已吃进的行记在 consumed 里，
|
||
// 因此第二条只会配到第二行。
|
||
func (s *steerTurn) match(pending []pendingMsg, recent []history.Message) []steerCandidate {
|
||
rows := make([]history.Message, 0, len(recent))
|
||
for _, r := range recent {
|
||
if r.Role != history.RoleUser || r.ID == "" || s.consumed[r.ID] {
|
||
continue
|
||
}
|
||
rows = append(rows, r)
|
||
}
|
||
if len(rows) == 0 {
|
||
return nil
|
||
}
|
||
claimed := make([]bool, len(rows))
|
||
out := make([]steerCandidate, 0, len(pending))
|
||
for _, p := range pending {
|
||
text, image := pendingUserContent(p)
|
||
if text == "" && image == "" {
|
||
continue
|
||
}
|
||
idx := -1
|
||
for i, r := range rows {
|
||
if !claimed[i] && r.Content == text && r.Image == image {
|
||
idx = i
|
||
break
|
||
}
|
||
}
|
||
if idx < 0 {
|
||
continue
|
||
}
|
||
claimed[idx] = true
|
||
out = append(out, steerCandidate{seq: p.seq, row: rows[idx]})
|
||
}
|
||
return out
|
||
}
|
||
|
||
// pendingUserContent 返回一条待处理消息里第一条用户消息的文本与图片，供与转录行
|
||
// 配对。没有用户内容（理论上不可达：pending 只收非空用户回合）时返回两个空串。
|
||
func pendingUserContent(p pendingMsg) (text, image string) {
|
||
for _, m := range p.messages {
|
||
if m.Role != model.RoleUser {
|
||
continue
|
||
}
|
||
if t, img := messageTextImage(m); t != "" || img != "" {
|
||
return t, img
|
||
}
|
||
}
|
||
return "", ""
|
||
}
|
||
|
||
// candidateSeqs 把候选消息收集为 pending 序号集合，供锁内一次性认领移除。
|
||
func candidateSeqs(cs []steerCandidate) map[uint64]bool {
|
||
seqs := make(map[uint64]bool, len(cs))
|
||
for _, c := range cs {
|
||
seqs[c.seq] = true
|
||
}
|
||
return seqs
|
||
}
|
||
|
||
// steerRows 把候选消息收成它们**已落库的转录行**，供按路补写 turn_id 用：这些行被
|
||
// 吃进之前都是以「还不知道属于哪一路」的状态落库的（那时列里是空的），因此这一刻
|
||
// 才谈得上给它们一个归属。
|
||
func steerRows(cs []steerCandidate) []history.Message {
|
||
rows := make([]history.Message, 0, len(cs))
|
||
for _, c := range cs {
|
||
rows = append(rows, c.row)
|
||
}
|
||
return rows
|
||
}
|
||
|
||
// pendingSnapshot 在状态锁下取该会话待处理队列的一份副本（世代不符时返回 nil：
|
||
// 那条路已失效，例如被打断、被取代或会话已清空）。取副本而不是
|
||
// 持锁做转录读取，是为了不让有界存储读取堵住 Submit。
|
||
func (o *Orchestrator) pendingSnapshot(sessionID string, gen uint64) []pendingMsg {
|
||
o.mu.Lock()
|
||
defer o.mu.Unlock()
|
||
st := o.states[sessionID]
|
||
if o.legLocked(st, gen) == nil || len(st.pending) == 0 {
|
||
return nil
|
||
}
|
||
return append([]pendingMsg(nil), st.pending...)
|
||
}
|
||
|
||
// consumePending 在状态锁下把已注入的消息从待处理队列移除（I2：注入＝已消费，绝不
|
||
// 被第二个回合再答一遍）。它按 pendingMsg.seq 认领，因此即便队列在同一时刻被追加
|
||
// （只会加在队尾）或被中断清空，也绝不会误删别的消息。
|
||
//
|
||
// 返回 false 表示这次认领**没有生效**（队列里已无这些序号，或本路已失效
|
||
// ——被打断、被取代）。调用方此时必须放弃注入：否则这条消息会被本回合与那条
|
||
// 新路径各答一次。它必须在把消息交给循环**之前**调用。
|
||
//
|
||
// 多路口径（design/15 §2.5.2 I2）：认领仍只走这一把锁与这一套序号，因此
|
||
// 「新起的第 2 路」与「在飞路的下一个 steer 边界」抢同一条消息时也只有一个赢家
|
||
// ——升级路径本就不把消息放进 pending（它直接登记为一条新路），而 steer 只能
|
||
// 认领 pending 里的条目，两者不可能同时拿走同一条。
|
||
func (o *Orchestrator) consumePending(sessionID string, gen uint64, seqs map[uint64]bool) bool {
|
||
if len(seqs) == 0 {
|
||
return false
|
||
}
|
||
o.mu.Lock()
|
||
defer o.mu.Unlock()
|
||
st := o.states[sessionID]
|
||
if o.legLocked(st, gen) == nil {
|
||
return false
|
||
}
|
||
kept := make([]pendingMsg, 0, len(st.pending))
|
||
removed := make([]model.Message, 0, len(seqs))
|
||
for _, p := range st.pending {
|
||
if seqs[p.seq] {
|
||
removed = append(removed, p.messages...)
|
||
// 被注入的若是语音请求，它的音频要由本回合结算时一并送回（协同 C1）；
|
||
// 否则那条请求的回合已被本回合接手，它却永远等不到 voice_output。
|
||
o.addVoiceSinkLocked(st, gen, p.voice)
|
||
continue
|
||
}
|
||
kept = append(kept, p)
|
||
}
|
||
st.pending = kept
|
||
if len(removed) == 0 {
|
||
return false
|
||
}
|
||
// 被吃进的消息同时并入续跑尾部：本回合的回答紧随其后，因此若有下一条消息接着
|
||
// 被续跑，它看得到「这句问过、她在这里答了」，而不是一条无头的回答（design/15
|
||
// §2.2 的增量上下文续接）。尾部有界，绝不因此无限增长。
|
||
st.tail = o.boundedTail(concatMessages(st.tail, removed...))
|
||
return true
|
||
}
|