feat(chat): chat_messages.turn_id 列 + DTO(写入接线留待下一步)
迁移 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。
This commit is contained in:
1 parent
354b08a45f
commit
d64b26d499
12 files changed
+950
-16
No files matched your search
@@ -93,7 +93,21 @@ type historyMessage struct {
|
||||
// 这类脆弱匹配),也能在重连后从读接口收敛。它为空时省略,而空意味着
|
||||
// 「这条消息没有幂等键」(老客户端、IM 镜像、历史行)。
|
||||
ClientMsgID string `json:"clientMsgId,omitempty"`
|
||||
CreatedAt int64 `json:"createdAt"`
|
||||
// TurnID 是这条消息所属的那一**路**回合的权威标识(design/15 §2.5.9):
|
||||
// 编排器生成,与该路全部流帧 / `turn_begin` 帧壳上的 turnId **同值**。
|
||||
// 同一路的 user 行与 assistant 行因此带同一个值,客户端据此
|
||||
// - 精确配对「回复 ↔ 提问」;
|
||||
// - 在**重连 / 回放**时从历史行恢复回复侧的分组小标(真并行下同一会话有两个
|
||||
// 交错的回答,只按顺序渲染会配错)。
|
||||
//
|
||||
// 它为空时省略,而空意味着「这一行不属于任何已知的回合」:加入该列之前的历史
|
||||
// 行、以及未经编排器的写入路径(IM 镜像等)都是这种情形,读起来与以往逐字
|
||||
// 相同。排队 / 被转向吃进的消息在**落库那一刻**也还没有自己的路(那时它由哪
|
||||
// 一路回答尚未确定),因此它们的那一帧 `message_persisted` 不带 turnId ——
|
||||
// 客户端从 `turn_begin`(帧壳 turnId + data.messageIds)得知归属,行本身随后
|
||||
// 由它实际被处理的那一路补写。
|
||||
TurnID string `json:"turnId,omitempty"`
|
||||
CreatedAt int64 `json:"createdAt"`
|
||||
}
|
||||
|
||||
// handleHistorySessions 先返回主会话,再返回每个子会话(最旧
|
||||
@@ -432,7 +446,11 @@ func toHistoryMessage(m history.Message) historyMessage {
|
||||
// (协同 G4),读接口(GET /sessions/{id}/messages)与
|
||||
// `message_persisted` 帧因此给出同一个答案。
|
||||
ClientMsgID: m.ClientMsgID,
|
||||
CreatedAt: m.CreatedAt,
|
||||
// TurnID 同理一并回填:读接口、WS 握手回放、`message_persisted` 与
|
||||
// `message_edited` 四条面共用本映射,因此它们对「这条消息属于哪一路」
|
||||
// 永远给出同一个答案(design/15 §2.5.9)。
|
||||
TurnID: m.TurnID,
|
||||
CreatedAt: m.CreatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,337 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"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"
|
||||
)
|
||||
|
||||
// 本文件锁定 chat_messages.turn_id 的四条出口与两条写入路径(design/15 §2.5.9):
|
||||
//
|
||||
// - 出口:历史 DTO(REST `GET /sessions/{id}/messages`)、WS 握手 `history` 回放、
|
||||
// 以及经 toHistoryMessage 构造的 `message_persisted` / `message_edited` 帧;
|
||||
// - 写入:落库时已知所属路(新起的一路 + 它的助手回复,随 INSERT)、以及落库时
|
||||
// 还不知道所属路(忙时入队 / 被转向吃进,由实际处理它的那一路用 assignTurnID
|
||||
// 补写)。
|
||||
//
|
||||
// 语义口径全部是**可选**的:空 ⇒ 该键从 JSON 里消失,与同族可选字段
|
||||
// (personaId / clientMsgId / documents)逐字同形。
|
||||
|
||||
// turnsOf 把一页消息按内容映射到它们的 turn_id,便于断言配对。
|
||||
func turnsOf(msgs []history.Message) map[string]string {
|
||||
out := make(map[string]string, len(msgs))
|
||||
for _, m := range msgs {
|
||||
out[m.Content] = m.TurnID
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// TestHistoryMessageTurnIDCarriedAndOmitted 钉住 DTO 形状:有标识就带出来,空则
|
||||
// 整个键消失(客户端据「键在不在」区分「属于某一路」与「不属于任何已知回合」)。
|
||||
func TestHistoryMessageTurnIDCarriedAndOmitted(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
withTurn := toHistoryMessage(history.Message{ID: "m1", SessionID: "web", Role: history.RoleUser, Content: "问", TurnID: "turn-a"})
|
||||
b, err := json.Marshal(withTurn)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
var decoded map[string]any
|
||||
if err := json.Unmarshal(b, &decoded); err != nil {
|
||||
t.Fatalf("unmarshal: %v", err)
|
||||
}
|
||||
if decoded["turnId"] != "turn-a" {
|
||||
t.Fatalf("turnId = %v, want turn-a (%s)", decoded["turnId"], b)
|
||||
}
|
||||
|
||||
// 空标识(旧行、IM 镜像、落库时还不知道所属路的行)⇒ 键缺席。
|
||||
legacy := toHistoryMessage(history.Message{ID: "m2", SessionID: "web", Role: history.RoleUser, Content: "问"})
|
||||
b, err = json.Marshal(legacy)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal legacy: %v", err)
|
||||
}
|
||||
if strings.Contains(string(b), `"turnId"`) {
|
||||
t.Fatalf("空 turn_id 应当省略该键,得到 %s", b)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHistoryMessagesEndpointCarriesTurnID 端到端演练读接口:同一路的两条消息
|
||||
// 带同一个 turnId,旧行(空)连键都不出现 —— 客户端因此能在重连后按 turnId 把
|
||||
// 「回复 ↔ 提问」配对起来,而不是靠顺序猜。
|
||||
func TestHistoryMessagesEndpointCarriesTurnID(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
_, ts, jwt, st := newHistoryEnv(t)
|
||||
for _, m := range []history.Message{
|
||||
{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "这条路的问题", TurnID: "turn-leg1", CreatedAt: 1000},
|
||||
{SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "这条路的回答", TurnID: "turn-leg1", CreatedAt: 1500},
|
||||
{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "另一路的问题", TurnID: "turn-leg2", CreatedAt: 2000},
|
||||
{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "本列之前的老行", CreatedAt: 2500},
|
||||
} {
|
||||
if err := st.Append(ctx, m); err != nil {
|
||||
t.Fatalf("seed %q: %v", m.Content, err)
|
||||
}
|
||||
}
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
// 先按原始 JSON 断言「键存在与否」,再按 DTO 断言取值。
|
||||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("messages status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
var raw []map[string]any
|
||||
decodeDataSlice(t, body, &raw)
|
||||
seen := make(map[string]string, len(raw))
|
||||
for _, m := range raw {
|
||||
content, _ := m["content"].(string)
|
||||
id, ok := m["turnId"]
|
||||
if !ok {
|
||||
seen[content] = "<absent>"
|
||||
continue
|
||||
}
|
||||
seen[content], _ = id.(string)
|
||||
}
|
||||
for content, want := range map[string]string{
|
||||
"这条路的问题": "turn-leg1",
|
||||
"这条路的回答": "turn-leg1",
|
||||
"另一路的问题": "turn-leg2",
|
||||
"本列之前的老行": "<absent>",
|
||||
} {
|
||||
if got := seen[content]; got != want {
|
||||
t.Fatalf("%q 的 turnId = %q, want %q(整页=%s)", content, got, want, body)
|
||||
}
|
||||
}
|
||||
|
||||
var page []historyMessage
|
||||
decodeDataSlice(t, body, &page)
|
||||
turns := turnsOfWire(page)
|
||||
if turns["这条路的问题"] != "turn-leg1" || turns["这条路的回答"] != "turn-leg1" {
|
||||
t.Fatalf("同一路两条 = %q / %q,want 都是 turn-leg1", turns["这条路的问题"], turns["这条路的回答"])
|
||||
}
|
||||
}
|
||||
|
||||
// turnsOfWire 是 historyMessage 版的内容 → turnId 映射。
|
||||
func turnsOfWire(msgs []historyMessage) map[string]string {
|
||||
out := make(map[string]string, len(msgs))
|
||||
for _, m := range msgs {
|
||||
out[m.Content] = m.TurnID
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// TestWSHandshakeReplayCarriesTurnID 钉住 WS 握手 `history` 回放这一面:它与读
|
||||
// 接口共用同一份映射,因此客户端刚连上就能按 turnId 恢复回复侧的分组小标。
|
||||
func TestWSHandshakeReplayCarriesTurnID(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
_, ts, jwt, st := newHistoryEnv(t)
|
||||
if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "回放里的提问", TurnID: "turn-replay", CreatedAt: 1000}); err != nil {
|
||||
t.Fatalf("seed user: %v", err)
|
||||
}
|
||||
if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "回放里的回答", TurnID: "turn-replay", CreatedAt: 1500}); err != nil {
|
||||
t.Fatalf("seed reply: %v", err)
|
||||
}
|
||||
|
||||
conn := dialWS(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner))
|
||||
defer conn.Close()
|
||||
_, raw, err := conn.ReadMessage()
|
||||
if err != nil {
|
||||
t.Fatalf("read handshake frame: %v", err)
|
||||
}
|
||||
var frame struct {
|
||||
Type string `json:"type"`
|
||||
Data struct {
|
||||
Messages []map[string]any `json:"messages"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &frame); err != nil {
|
||||
t.Fatalf("decode frame %s: %v", raw, err)
|
||||
}
|
||||
if frame.Type != "history" {
|
||||
t.Fatalf("first frame = %q, want history (%s)", frame.Type, raw)
|
||||
}
|
||||
if len(frame.Data.Messages) != 2 {
|
||||
t.Fatalf("replayed %d messages, want 2 (%s)", len(frame.Data.Messages), raw)
|
||||
}
|
||||
for _, m := range frame.Data.Messages {
|
||||
if m["turnId"] != "turn-replay" {
|
||||
t.Fatalf("回放消息 = %v, want turnId=turn-replay", m)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestPersistTurnCarriesTurnIDOnBothRows 锁定第一段写入:新起的一路的用户消息与
|
||||
// 它的助手回复带**同一个** turn_id,且 `message_persisted` 那一帧(由同一条已存
|
||||
// 消息构造)也带着它 —— 客户端因此不需要等一次历史拉取就能配对。
|
||||
func TestPersistTurnCarriesTurnIDOnBothRows(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
_, _, _, st := newHistoryEnv(t)
|
||||
var broadcast []history.Message
|
||||
o := &Orchestrator{history: st, log: testLogger(), onPersisted: func(m history.Message) { broadcast = append(broadcast, m) }}
|
||||
|
||||
legCtx := withTurnID(ctx, "turn-solo")
|
||||
o.persistUserMessages(legCtx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "单路的问题")}, "web:test", "chat", "cyrene")
|
||||
o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "单路的回答"}, "web:test", "chat", "cyrene", "turn-solo")
|
||||
|
||||
msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
turns := turnsOf(msgs)
|
||||
if turns["单路的问题"] != "turn-solo" || turns["单路的回答"] != "turn-solo" {
|
||||
t.Fatalf("两条 = %q / %q,want 都是 turn-solo(同一路两条同值)", turns["单路的问题"], turns["单路的回答"])
|
||||
}
|
||||
if len(broadcast) != 2 || broadcast[0].TurnID != "turn-solo" || broadcast[1].TurnID != "turn-solo" {
|
||||
t.Fatalf("message_persisted 载荷 = %+v, want 两条都带 turn-solo", broadcast)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPersistParallelLegsCarryDistinctTurnIDs 锁定并行两路:每一路的两条消息
|
||||
// 同值,两路之间不同值 —— 这正是「回复 ↔ 提问」精确配对的前提(按顺序渲染会
|
||||
// 把两条交错的回答配错)。
|
||||
func TestPersistParallelLegsCarryDistinctTurnIDs(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
_, _, _, st := newHistoryEnv(t)
|
||||
o := &Orchestrator{history: st, log: testLogger()}
|
||||
|
||||
o.persistUserMessages(withTurnID(ctx, "turn-1"), history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "第一路的问题")}, "web:test", "chat", "cyrene")
|
||||
o.persistUserMessages(withTurnID(ctx, "turn-2"), history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "第二路的问题")}, "web:test", "chat", "cyrene")
|
||||
o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "第一路的回答"}, "web:test", "chat", "cyrene", "turn-1")
|
||||
o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "第二路的回答"}, "web:test", "chat", "cyrene", "turn-2")
|
||||
|
||||
msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
turns := turnsOf(msgs)
|
||||
for content, want := range map[string]string{
|
||||
"第一路的问题": "turn-1",
|
||||
"第一路的回答": "turn-1",
|
||||
"第二路的问题": "turn-2",
|
||||
"第二路的回答": "turn-2",
|
||||
} {
|
||||
if got := turns[content]; got != want {
|
||||
t.Fatalf("%q 的 turn_id = %q, want %q(两路各自两条同值、两路不同值)", content, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestAssignTurnIDStampsQueuedAndSteeredRows 锁定第二段写入与它的归属口径:忙时
|
||||
// 入队(以及被转向吃进)的消息在落库那一刻还不知道自己属于哪一路,由**实际处理
|
||||
// 它的那一路**补写;已经带上标识的行不被后来者改写(先到先得,一行只属于一路)。
|
||||
func TestAssignTurnIDStampsQueuedAndSteeredRows(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
_, _, _, st := newHistoryEnv(t)
|
||||
o := &Orchestrator{history: st, log: testLogger()}
|
||||
|
||||
// 忙时入队:落库时 ctx 上没有回合标识 ⇒ 列里留空。
|
||||
queued := o.persistUserMessages(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "排队的一句")}, "web:test", "chat", "cyrene")
|
||||
// 被转向吃进的一句:同样在它自己的那一轮之前就落了库。
|
||||
steered := o.persistUserMessages(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "被转向吃进的一句")}, "web:test", "chat", "cyrene")
|
||||
if len(queued) != 1 || len(steered) != 1 {
|
||||
t.Fatalf("落库条数 = %d / %d, want 各 1", len(queued), len(steered))
|
||||
}
|
||||
before, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
for _, m := range before {
|
||||
if m.TurnID != "" {
|
||||
t.Fatalf("消息 %q 在补写之前就有 turn_id %q,want 空", m.Content, m.TurnID)
|
||||
}
|
||||
}
|
||||
|
||||
// 这两条最终由同一路回答(转向把它并进了在飞的那一路)⇒ 补写同一个标识。
|
||||
o.assignTurnID(history.MainSessionID, "turn-inflight", append(queued, steered...))
|
||||
after, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||||
if err != nil {
|
||||
t.Fatalf("list after assign: %v", err)
|
||||
}
|
||||
for _, m := range after {
|
||||
if m.TurnID != "turn-inflight" {
|
||||
t.Fatalf("消息 %q 的 turn_id = %q, want turn-inflight(按实际处理它的那一路写)", m.Content, m.TurnID)
|
||||
}
|
||||
}
|
||||
|
||||
// 另一路(例如续跑)试图认领同一行:不得改写既有的归属。
|
||||
o.assignTurnID(history.MainSessionID, "turn-later", queued)
|
||||
again, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||||
if err != nil {
|
||||
t.Fatalf("list after re-assign: %v", err)
|
||||
}
|
||||
for _, m := range again {
|
||||
if m.TurnID != "turn-inflight" {
|
||||
t.Fatalf("消息 %q 的 turn_id 被改写成 %q, want 仍是 turn-inflight(先到先得)", m.Content, m.TurnID)
|
||||
}
|
||||
}
|
||||
|
||||
// 空标识 / 空列表是空操作(未接线的旧路径与「这一路没有落库任何消息」)。
|
||||
o.assignTurnID(history.MainSessionID, "", queued)
|
||||
o.assignTurnID(history.MainSessionID, "turn-x", nil)
|
||||
}
|
||||
|
||||
// TestTurnIDOfLegResolvesInFlightLeg 钉住补写路径要用的那个查询:在途路的 gen →
|
||||
// 该路的权威 turnId;不在途(被打断 / 已被取代 / 会话里没有它)⇒ 空串。
|
||||
func TestTurnIDOfLegResolvesInFlightLeg(t *testing.T) {
|
||||
t.Parallel()
|
||||
o := &Orchestrator{log: testLogger(), states: map[string]*sessionState{
|
||||
history.MainSessionID: {legs: map[uint64]*turnLeg{7: {turnID: "turn-7"}, 8: {turnID: "turn-8"}}},
|
||||
}}
|
||||
if got := o.turnIDOfLeg(history.MainSessionID, 8); got != "turn-8" {
|
||||
t.Fatalf("turnIDOfLeg(8) = %q, want turn-8", got)
|
||||
}
|
||||
if got := o.turnIDOfLeg(history.MainSessionID, 9); got != "" {
|
||||
t.Fatalf("turnIDOfLeg(未知 gen) = %q, want 空串", got)
|
||||
}
|
||||
if got := o.turnIDOfLeg("web:unknown", 7); got != "" {
|
||||
t.Fatalf("turnIDOfLeg(未知会话) = %q, want 空串", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSubmitWiresTurnIDForBothRows 是「单路回合 ⇒ 用户消息与助手回复两条 turn_id
|
||||
// 同值且非空」的**端到端**验收:真存储 + 真编排器,走 Submit → 起路 → 落库 → 结算
|
||||
// 的完整链路,断言提交回报的 TurnID 与落库那两条的 turn_id 完全一致。
|
||||
//
|
||||
// ⏳ 它今天被 Skip 挡住:写入点必须在四个「本子代理硬约束禁止改动」的文件里接线
|
||||
// (session_parallel.go 的 publishLegStart、session_orchestrator.go 的续跑分支、
|
||||
// session_steer.go 的 Steer、session_turn_reply.go 的 persistTurnReply)。接线完成后
|
||||
// 删掉下面这行 Skip,本用例即成为该契约的端到端验收。
|
||||
func TestSubmitWiresTurnIDForBothRows(t *testing.T) {
|
||||
t.Skip("待接线:编排器的四个写入点尚未把 TurnID 交给落库(见交接报告),本用例是接线后的端到端验收")
|
||||
|
||||
ctx := context.Background()
|
||||
_, _, _, st := newHistoryEnv(t)
|
||||
run := newGatedLegRunner()
|
||||
o := NewOrchestrator(OrchestratorOptions{Run: run.Run, Log: testLogger(), History: st})
|
||||
|
||||
sr := o.Submit(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "端到端的问题")}, loop.OwnerEnv())
|
||||
if !sr.Immediate || sr.TurnID == "" {
|
||||
t.Fatalf("Submit = %+v, want 立刻起一路并回报非空 TurnID", sr)
|
||||
}
|
||||
run.waitStarted(1)
|
||||
run.release(0)
|
||||
waitActiveTurns(t, o, history.MainSessionID, 0)
|
||||
|
||||
msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(msgs) != 2 {
|
||||
t.Fatalf("落库 %d 条, want 2(用户消息 + 助手回复): %+v", len(msgs), msgs)
|
||||
}
|
||||
turns := turnsOf(msgs)
|
||||
if turns["端到端的问题"] != sr.TurnID || turns["reply-1"] != sr.TurnID {
|
||||
t.Fatalf("两条 turn_id = %q / %q, want 都是提交回报的 %q",
|
||||
turns["端到端的问题"], turns["reply-1"], sr.TurnID)
|
||||
}
|
||||
}
|
||||
@@ -61,6 +61,11 @@ func modeFrom(ctx context.Context) string {
|
||||
// 与编排器预分配的消息 id:写入因此可被 (session_id, client_msg_id) 的唯一索引
|
||||
// 判重。省略幂等键时 ctx 上没有该值,写入与加入这一列之前逐字相同。
|
||||
//
|
||||
// ctx 上若带着本路的回合标识(design/15 §2.5.9,withTurnID),该标识随**写入**
|
||||
// 落进 turn_id 列 —— 这正是「这条路已经把这条消息吃进去了」要留下的配对信息。
|
||||
// 没有它(排队入队、未接线的路径)时列里留空串,由它实际被处理的那一路随后补写
|
||||
// (见 assignTurnID 与 internal/runtime/history/turn_id.go)。
|
||||
//
|
||||
// 它返回刚落库的那些消息(含存储分配的 id 与**只有引用**的 document 列),
|
||||
// 使调用方能把它们挂到本回合上:抽取结论要等回合跑起来才有,只能在回合结束后
|
||||
// 回写进同一列(design/24 §6.1,见 session_document_status.go)。这就是那条
|
||||
@@ -70,6 +75,7 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string
|
||||
return nil
|
||||
}
|
||||
hint := clientMsgHintFrom(ctx)
|
||||
turnID := turnIDFrom(ctx)
|
||||
persisted := make([]history.Message, 0, len(messages))
|
||||
for _, m := range messages {
|
||||
if m.Role != model.RoleUser {
|
||||
@@ -85,7 +91,7 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string
|
||||
if text == "" && image == "" && documents == "" {
|
||||
continue
|
||||
}
|
||||
msg := history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, Document: documents, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID}
|
||||
msg := history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, Document: documents, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID, TurnID: turnID}
|
||||
// 幂等键只标注**第一条**用户消息:一个回合至多只有一条用户消息,而把
|
||||
// 同一个键盖到多条上会让它们互相冲突(唯一索引把它们判成同一条)。
|
||||
if hint.ClientMsgID != "" {
|
||||
@@ -128,11 +134,18 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string
|
||||
// 它是**裸写入**,判定属于 persistTurnReply(session_turn_reply.go):收尾路径
|
||||
// 经后者进入,由它保证「归本代次、且不是被取消的回合一定有一条可见消息」——
|
||||
// 空文本在这里仍然是空操作,因此本函数绝不自作主张编造内容。
|
||||
func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, deviceID, mode, personaID string) {
|
||||
//
|
||||
// turnID 是本路的回合标识(design/15 §2.5.9),可选:给出即随**写入**落进 turn_id
|
||||
// 列,与它那条用户消息同值,客户端因此能精确配对「回复 ↔ 提问」。省略(未接线的
|
||||
// 调用路径)等价于加入该列之前的行为(列里留空串)。它刻意是可变参数而不是必填
|
||||
// 形参:既有调用点、以及「这一路没有回合标识」的旧路径都不必先传一个空值,而
|
||||
// 已给出的第一个值之外的多余值**被忽略**(一个回合只有一份标识,多给是调用方的
|
||||
// 笔误,绝不据此分叉行为)。
|
||||
func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, deviceID, mode, personaID string, turnID ...string) {
|
||||
if o.history == nil || res == nil || res.Text == "" {
|
||||
return
|
||||
}
|
||||
full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text, Mood: res.Mood, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID})
|
||||
full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text, Mood: res.Mood, DeviceID: deviceID, Mode: modeOrDefault(mode), PersonaID: personaID, TurnID: firstTurnID(turnID)})
|
||||
if err != nil {
|
||||
o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID, "device_id", deviceID)
|
||||
return
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
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))
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user