Cyrene_v2/internal/services/gateway/session_steer.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"
)

// 本文件实现「忙时新消息在回合内边界注入同一回合」的网关侧
// （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
	consumed     map[string]bool
}

// beginSteer 为一个刚启动的回合建一份转向游标。编排器把它挂到**该回合自己的 ctx**
// 上（loop.WithSteerSource），因此每个回合、每次续跑各有一份独立游标，上一个回合
// 已吃进的消息绝不会被下一个回合重复注入或漏掉。
func (o *Orchestrator) beginSteer(sessionID string, gen uint64) *steerTurn {
	return &steerTurn{
		orchestrator: o,
		sessionID:    sessionID,
		gen:          gen,
		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.(transcriptReader)
	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, transcriptContextCount)
	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
	}
	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。
		o.emitState(sessionID, wsSteerAck, map[string]any{"messageId": c.row.ID, "atStep": req.Step})
		out = append(out, transcriptMessage(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
}

// 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
}