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。
243 lines
8.8 KiB
Go
243 lines
8.8 KiB
Go
package history
|
||
|
||
import (
|
||
"context"
|
||
"testing"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||
)
|
||
|
||
// 本文件锁定聊天消息的 **turn_id** 列（design/15 §2.5.9）：它回答「这一行属于哪
|
||
// 一路回合」，使客户端在重连 / 回放时仍能精确配对「回复 ↔ 提问」。
|
||
//
|
||
// 三条语义被逐条钉住：
|
||
// - 落库时已经知道所属路的消息（新起的一路 + 它的助手回复）随 INSERT 写入，
|
||
// 同一路两条**同值**、两路**不同值**；
|
||
// - 落库时还不知道所属路的消息（忙时入队 / 被 steer 吃进）留空串，由它实际被
|
||
// 处理的那一路**补写**（SetMessagesTurnID），且**先到先得**（已带标识的行不被
|
||
// 改写 —— 一行只属于一路）；
|
||
// - 省略标识与加入本列之前逐字相同：空串默认值、读路径照常工作。
|
||
|
||
func TestAppendFullRoundTripsTurnID(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st := newStore(t)
|
||
|
||
user, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "先问你一句", TurnID: "turn-leg1", CreatedAt: 1000})
|
||
if err != nil {
|
||
t.Fatalf("append user: %v", err)
|
||
}
|
||
reply, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleAssistant, Content: "答你一句", TurnID: "turn-leg1", CreatedAt: 2000})
|
||
if err != nil {
|
||
t.Fatalf("append assistant: %v", err)
|
||
}
|
||
// 第二路（并行）带自己的标识与自己的两条消息。
|
||
if _, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "另一路的问题", TurnID: "turn-leg2", CreatedAt: 3000}); err != nil {
|
||
t.Fatalf("append parallel user: %v", err)
|
||
}
|
||
|
||
if user.TurnID != "turn-leg1" || reply.TurnID != "turn-leg1" {
|
||
t.Fatalf("写入返回值 = %q / %q, want turn-leg1（同一路两条同值）", user.TurnID, reply.TurnID)
|
||
}
|
||
|
||
// 读路径（共用 messageColumns/scanMessage）必须把标识带出来。
|
||
msgs, err := st.List(ctx, MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
got := make(map[string]string, len(msgs))
|
||
for _, m := range msgs {
|
||
got[m.Content] = m.TurnID
|
||
}
|
||
if got["先问你一句"] != "turn-leg1" || got["答你一句"] != "turn-leg1" {
|
||
t.Fatalf("同一路两条 = %q / %q, want 都是 turn-leg1", got["先问你一句"], got["答你一句"])
|
||
}
|
||
if got["另一路的问题"] != "turn-leg2" {
|
||
t.Fatalf("另一路 = %q, want turn-leg2（两路绝不同值）", got["另一路的问题"])
|
||
}
|
||
|
||
// 其余读面同源（它们共用同一份列清单，这里抽验两条最容易漏改的）。
|
||
recent, err := st.Recent(ctx, MainSessionID, 1)
|
||
if err != nil {
|
||
t.Fatalf("recent: %v", err)
|
||
}
|
||
if len(recent) != 1 || recent[0].TurnID != "turn-leg2" {
|
||
t.Fatalf("recent = %+v, want 最新那条带 turn-leg2", recent)
|
||
}
|
||
hits, err := st.Search(ctx, MainSessionID, "先问你一句", 0, 10)
|
||
if err != nil {
|
||
t.Fatalf("search: %v", err)
|
||
}
|
||
if len(hits) != 1 || hits[0].TurnID != "turn-leg1" {
|
||
t.Fatalf("search = %+v, want 命中那条带 turn-leg1", hits)
|
||
}
|
||
found, err := st.FindByClientMsgID(ctx, MainSessionID, "")
|
||
if err == nil {
|
||
t.Fatalf("FindByClientMsgID(空键) = %+v, want ErrMessageNotFound", found)
|
||
}
|
||
}
|
||
|
||
func TestAppendWithoutTurnIDLeavesItEmpty(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st := newStore(t)
|
||
|
||
// 省略标识（老路径 / IM 镜像 / 排队入队）与加入本列之前逐字相同：空串。
|
||
if _, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "没有回合标识", CreatedAt: 1000}); err != nil {
|
||
t.Fatalf("append: %v", err)
|
||
}
|
||
msgs, err := st.List(ctx, MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
if len(msgs) != 1 || msgs[0].TurnID != "" {
|
||
t.Fatalf("msgs = %+v, want 空 turn_id（省略 ⇒ 不属于任何已知回合）", msgs)
|
||
}
|
||
}
|
||
|
||
func TestSetMessagesTurnIDAssignsUnassignedOnly(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st := newStore(t)
|
||
|
||
// 两行在落库时都还没有所属的路（忙时入队 / 被 steer 吃进的那种行）。
|
||
queued, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "排队的一句", CreatedAt: 1000})
|
||
if err != nil {
|
||
t.Fatalf("append queued: %v", err)
|
||
}
|
||
steered, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "被转向吃进的一句", CreatedAt: 1500})
|
||
if err != nil {
|
||
t.Fatalf("append steered: %v", err)
|
||
}
|
||
|
||
// 第二段：由**实际处理它们的那一路**补写。
|
||
if err := st.SetMessagesTurnID(ctx, []string{queued.ID, steered.ID}, "turn-actual"); err != nil {
|
||
t.Fatalf("set turn id: %v", err)
|
||
}
|
||
msgs, err := st.List(ctx, MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
for _, m := range msgs {
|
||
if m.TurnID != "turn-actual" {
|
||
t.Fatalf("消息 %q 的 turn_id = %q, want turn-actual（按实际处理它的那一路写）", m.Content, m.TurnID)
|
||
}
|
||
}
|
||
|
||
// 先到先得：已经带上标识的行绝不被后来者改写（一行只属于一路）。
|
||
if err := st.SetMessagesTurnID(ctx, []string{queued.ID}, "turn-later"); err != nil {
|
||
t.Fatalf("set turn id again: %v", err)
|
||
}
|
||
msgs, err = st.List(ctx, MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list after re-set: %v", err)
|
||
}
|
||
for _, m := range msgs {
|
||
if m.TurnID != "turn-actual" {
|
||
t.Fatalf("消息 %q 的 turn_id 被改写成 %q, want 仍是 turn-actual", m.Content, m.TurnID)
|
||
}
|
||
}
|
||
|
||
// 未知 id 被忽略而不是报错：第二段的常态是「刚落的行」，找不到只可能是它已被删除。
|
||
if err := st.SetMessagesTurnID(ctx, []string{"不存在的 id"}, "turn-actual"); err != nil {
|
||
t.Fatalf("unknown id: %v", err)
|
||
}
|
||
}
|
||
|
||
func TestSetMessagesTurnIDNoopInputs(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st := newStore(t)
|
||
row, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "保持空", CreatedAt: 1000})
|
||
if err != nil {
|
||
t.Fatalf("append: %v", err)
|
||
}
|
||
|
||
// 空 id 列表、空标识、只含空串的列表都是空操作（调用方不必先判断）。
|
||
cases := []struct {
|
||
name string
|
||
ids []string
|
||
turnID string
|
||
}{
|
||
{name: "no ids", ids: nil, turnID: "turn-x"},
|
||
{name: "no turn id", ids: []string{row.ID}, turnID: ""},
|
||
{name: "blank ids", ids: []string{"", ""}, turnID: "turn-x"},
|
||
}
|
||
for _, tc := range cases {
|
||
if err := st.SetMessagesTurnID(ctx, tc.ids, tc.turnID); err != nil {
|
||
t.Fatalf("%s: %v", tc.name, err)
|
||
}
|
||
}
|
||
msgs, err := st.List(ctx, MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
if len(msgs) != 1 || msgs[0].TurnID != "" {
|
||
t.Fatalf("msgs = %+v, want 空 turn_id（空输入不写任何东西）", msgs)
|
||
}
|
||
|
||
// 重复 id 只算一次，且整批落到同一个标识上。
|
||
if err := st.SetMessagesTurnID(ctx, []string{row.ID, row.ID, "", row.ID}, "turn-dup"); err != nil {
|
||
t.Fatalf("duplicate ids: %v", err)
|
||
}
|
||
msgs, err = st.List(ctx, MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list after dup: %v", err)
|
||
}
|
||
if msgs[0].TurnID != "turn-dup" {
|
||
t.Fatalf("turn_id = %q, want turn-dup", msgs[0].TurnID)
|
||
}
|
||
}
|
||
|
||
// TestNewStoreRollsForwardTurnIDColumn 验证早于 0018 建出的 chat_messages 表会被
|
||
// NewStore 幂等前滚出 turn_id 列（与 client_msg_id / document 同一手法）：全新库由
|
||
// 建表 DDL 一次建全，既有库靠补列，两条路都必须能写能读带标识的消息。
|
||
func TestNewStoreRollsForwardTurnIDColumn(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
conn, err := db.Open(ctx, ":memory:", testLogger())
|
||
if err != nil {
|
||
t.Fatalf("open memory db: %v", err)
|
||
}
|
||
t.Cleanup(func() { _ = conn.Close() })
|
||
|
||
// 0018 之前的形状：有 document（0015）但没有 turn_id。
|
||
const legacyTable = `
|
||
CREATE TABLE chat_messages (
|
||
id TEXT PRIMARY KEY,
|
||
session_id TEXT NOT NULL,
|
||
role TEXT NOT NULL,
|
||
content TEXT NOT NULL,
|
||
mode TEXT,
|
||
mood TEXT,
|
||
image TEXT,
|
||
document TEXT NOT NULL DEFAULT '',
|
||
thinking TEXT,
|
||
platform TEXT NOT NULL DEFAULT 'web',
|
||
device_id TEXT NOT NULL DEFAULT '',
|
||
sender_name TEXT NOT NULL DEFAULT '',
|
||
adapter_id TEXT NOT NULL DEFAULT '',
|
||
persona_id TEXT NOT NULL DEFAULT '',
|
||
client_msg_id TEXT NOT NULL DEFAULT '',
|
||
created_at INTEGER NOT NULL
|
||
)`
|
||
if _, err := conn.ExecContext(ctx, legacyTable); err != nil {
|
||
t.Fatalf("create legacy table: %v", err)
|
||
}
|
||
|
||
st, err := NewStore(ctx, conn, testLogger())
|
||
if err != nil {
|
||
t.Fatalf("new store: %v", err)
|
||
}
|
||
if _, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "前滚之后", TurnID: "turn-rolled", CreatedAt: 2000}); err != nil {
|
||
t.Fatalf("append after roll forward: %v", err)
|
||
}
|
||
msgs, err := st.List(ctx, MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
if len(msgs) != 1 || msgs[0].TurnID != "turn-rolled" {
|
||
t.Fatalf("msgs = %+v, want turn-rolled 往返", msgs)
|
||
}
|
||
}
|