AskaEth
d64b26d499
迁移 0018：ALTER TABLE chat_messages ADD COLUMN turn_id TEXT NOT NULL DEFAULT ''
（与 persona_id / client_msg_id / document 同族：不用 NULL，空串 = 不属于任何已知回合 ⇒ 旧行零迁移、键省略）。
存储层全链路带 turn_id（列清单唯一一份 ⇒ List/ListAfter/Recent/Search/FindByClientMsgID 自动同源）；
新增 history/turn_id.go：ensureTurnIDColumn（既有库幂等前滚）+ SetMessagesTurnID
（第二段补写：单事务参数化、先到先得 WHERE turn_id = ''、幂等、未知 id 忽略）。
DTO：historyMessage.TurnID（json turnId,omitempty）⇒ REST 列表 / admin 渠道日志 / WS 握手回放 /
message_persisted / channel_persisted / message_edited 一次全覆盖。
写入面：persistUserMessages 读 ctx 回合标识；persistAssistantReply 增可选尾参（既有调用点零改动）；
withTurnID / turnIDFrom / assignTurnID / turnIDOfLeg。
⚠️ 编排器把 turnId 交给落库的 4 处（新起一路 / 续跑路 / 被 steer 吃进 / 助手回复）尚未接线
（那些文件当时被另一路占用）；当前生产行为 turn_id 仍为 ''。清单已在切片报告里精确到文件:行。
13 新用例 + 5 条变异（INSERT 不带列 / DTO 不回填 / 去掉 omitempty / 去掉先到先得 / 不回滚前滚）。
为守 session_persist.go ≤500 行，turnId 一节拆到新文件 session_turn_id.go。
120 lines
5.7 KiB
Go
120 lines
5.7 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
)
|
||
|
||
// 本文件归属消息行上的**回合标识**（turnId，design/15 §2.5.9）的持久化面：把编排器
|
||
// 为每一路生成的权威 id 落进 chat_messages.turn_id（迁移 0018），使客户端在重连 /
|
||
// 回放时仍能精确配对「回复 ↔ 提问」—— 流帧是易失的，历史行不是。
|
||
//
|
||
// 它从 session_persist.go 拆出来，使那个文件继续留在「单文件 ≤500 行」的硬约束内
|
||
// （与 session_events.go / session_document_status.go 同一理由）。
|
||
//
|
||
// 写入是**两段式**（与 document 列同一手法）：
|
||
// - 落库时所属路已经存在（新起的一路 + 它的助手回复）⇒ 随 INSERT 写入：用户消息
|
||
// 走 ctx 通道（withTurnID，与 deviceID / mode / 幂等键同一种机制），助手回复走
|
||
// 显式参数（persistAssistantReply 的可变尾参）；
|
||
// - 落库时还不知道所属路（忙时入队、被转向吃进）⇒ 由**实际处理它的那一路**在起跑 /
|
||
// 消费之后用 assignTurnID 补写。
|
||
//
|
||
// 空值一律是空操作：省略标识的路径（既有调用点、未接线该能力的构建）行为与加入本列
|
||
// 之前逐字相同。
|
||
|
||
// turnIDContextKey 是携带本路回合标识的 context 键（design/15 §2.5.9）。它由
|
||
// 「起了一路」的那一刻挂上，使这一路落的用户消息带上它。
|
||
type turnIDContextKey struct{}
|
||
|
||
// withTurnID 返回携带本路回合标识的 context。空 turnID 是空操作：没有回合标识的
|
||
// 路径（未接编排器的旧路径、存储替身）在 ctx 上不留任何痕迹。
|
||
func withTurnID(ctx context.Context, turnID string) context.Context {
|
||
if turnID == "" {
|
||
return ctx
|
||
}
|
||
return context.WithValue(ctx, turnIDContextKey{}, turnID)
|
||
}
|
||
|
||
// turnIDFrom 返回 ctx 携带的本路回合标识，未设置时返回 ""。
|
||
func turnIDFrom(ctx context.Context) string {
|
||
v, _ := ctx.Value(turnIDContextKey{}).(string)
|
||
return v
|
||
}
|
||
|
||
// firstTurnID 返回可选回合标识里的第一个非空值，没有时返回 ""。它服务
|
||
// persistAssistantReply 的可变形参：省略与空值等价（都不写这一列）。
|
||
func firstTurnID(turnID []string) string {
|
||
for _, id := range turnID {
|
||
if id != "" {
|
||
return id
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// turnIDWriter 是 HistoryStore 的一个**可选**更窄视图（与 clientMsgIDReader /
|
||
// documentStatusWriter 同一做法）：按消息 id 批量落定 turn_id。未实现该视图的
|
||
// 存储替身或后端退回「不补写」并留下带来源的 WARN —— 补写是增益，绝不能反过来
|
||
// 卡住会话。
|
||
type turnIDWriter interface {
|
||
SetMessagesTurnID(ctx context.Context, ids []string, turnID string) error
|
||
}
|
||
|
||
// turnIDLogSource 是回合标识补写的日志来源标签，使这条冷路径上的 WARN 能被追溯到
|
||
// 具体子功能（AGENTS §10 的「日志必须标来源」）。
|
||
const turnIDLogSource = "gateway/session-persist"
|
||
|
||
// turnIDOfLeg 返回该会话在途路 gen 的权威回合标识；该路已不在途（被打断清空、
|
||
// 已被取代、会话里没有它）时返回 ""。调用方用它把「这一刻算数的这一路」翻译成
|
||
// 要落库的标识 —— 它**不得**在持有 o.mu 时调用（本函数自己取锁）。
|
||
//
|
||
// ⚠️ 它有意**不用于**任何已经拿到 `*turnLeg` 的地方：InterruptScope(all) 会把
|
||
// st.legs 整个清掉，而被打断那一路的助手文本仍要按原样落库（design/15 §3 的
|
||
// 「内容是真的」），那时按 gen 回查会拿到空串、把配对丢掉。凡是手里有 leg 的
|
||
// 调用点都传 `leg.turnID`（出生后不可变），只有确实只拿得到 gen 的地方才走本函数。
|
||
func (o *Orchestrator) turnIDOfLeg(sessionID string, gen uint64) string {
|
||
o.mu.Lock()
|
||
defer o.mu.Unlock()
|
||
if leg := o.legLocked(o.states[sessionID], gen); leg != nil {
|
||
return leg.turnID
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// assignTurnID 把一组已落库消息的 turn_id 落定为 turnID —— 两段式写入的第二段。
|
||
// 它服务「落库那一刻还不知道自己属于哪一路」的消息：忙时入队（它的路要等轮到它
|
||
// 才起）与**被转向吃进**的消息（它由那个在飞的路回答）。两处都在路确定之后调用
|
||
// 本函数，因此这些行最终带上的正是**实际处理它们的那一路**的标识。
|
||
//
|
||
// 空 turnID、空消息列表、未接存储都直接返回（没有标识可落，也就没有写入）。存储
|
||
// 未实现该可选视图时同样返回 —— 但**写库失败绝不静默**：打一条带来源的 WARN，
|
||
// 日志只带会话、条数与错误，不带正文。写入用全新的后台 context（与
|
||
// persistAssistantReply / persistDocumentStatus 同一理由）：被取消回合的 ctx
|
||
// 绝不该中断这次补写。
|
||
//
|
||
// 它绝不改写已经带标识的行（存储侧只写给仍为空串的行，先到先得），因此重复调用
|
||
// 与「两条路径都想认领同一行」都不会让配对漂移。
|
||
//
|
||
// 它必须在**状态锁之外**调用：它写库（那是慢操作）。握着 o.mu 的调用点要先解锁。
|
||
func (o *Orchestrator) assignTurnID(sessionID, turnID string, msgs []history.Message) {
|
||
if o.history == nil || turnID == "" || len(msgs) == 0 {
|
||
return
|
||
}
|
||
writer, ok := o.history.(turnIDWriter)
|
||
if !ok {
|
||
return
|
||
}
|
||
ids := make([]string, 0, len(msgs))
|
||
for _, m := range msgs {
|
||
if m.ID != "" {
|
||
ids = append(ids, m.ID)
|
||
}
|
||
}
|
||
if err := writer.SetMessagesTurnID(context.Background(), ids, turnID); err != nil {
|
||
logger.WithSource(o.log, turnIDLogSource).Warn("turn id not persisted",
|
||
"err", err, "session", sessionID, "turn", turnID, "messages", len(ids))
|
||
}
|
||
}
|