前提由 FV2 提供(图自 v0.77 起带每路 turnId、每路各落一行)。本片:
- 迁移 0023:flow_graphs 加「按路查询键」(写入时回填图的 TurnID),老行按缺席处理
- GET /api/v1/sessions/{id}/messages 加**可选** toolCalls[],形状严格照 Docs/api-client.md
v0.76 ① 已登记的:{name, status(ok|error|unknown), durationMs, argsSummary, resultSummary}
- 🔴 图取不到 ⇒ **字段整体缺席**(绝不回空数组冒充「没调过工具」)
- 一条批量 SQL(按页内 turnId 集合),禁 N+1;单条 ≤20 个工具、resultSummary ≤500 字符(常量+依据)
- 只放图里已有的、已脱敏内容,不放宽脱敏通道
314 lines
12 KiB
Go
314 lines
12 KiB
Go
package flowgraph
|
||
|
||
// 本文件钉住**按路查图**(切片 CODE-TC;迁移 0023 的 `lane_turn_id`)。它是契约
|
||
// v0.76 ① 的 `toolCalls[]` 一直缺的那块读侧地基:消息行带的是**路标识**
|
||
// (`chat_messages.turn_id`),图行必须能按**同一个值**查到,否则「回看这条回复当时
|
||
// 调了什么工具」只能靠猜时间窗 —— 那是项目口径明令禁止的。
|
||
//
|
||
// 断言的是**库里的事实**(列里的值、查回来的行、索引在不在),不按日志文本断言,
|
||
// 与 flowgraph_test.go / flowgraph_lane_test.go 同一口径。
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"sort"
|
||
"testing"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||
)
|
||
|
||
// laneTurnMigrationID 是 0023 的迁移 id(与文件名逐字一致 —— 回填用例靠它把迁移
|
||
// 记账行撤回,使 `MigrateUp` **真的**重新应用一次 0023)。
|
||
const laneTurnMigrationID = "0023_flow_graphs_lane_turn_id"
|
||
|
||
// laneRoundGraph 造一条路的收尾图:图 id 与真建图器同形(`{会话}-{序号}[#{路}]`),
|
||
// laneTurn 是这一路的权威回合标识(= 消息行 `chat_messages.turn_id` 的同值)。
|
||
func laneRoundGraph(session, graphID, laneTurn string) *trace.FlowGraph {
|
||
return &trace.FlowGraph{
|
||
ID: graphID,
|
||
SessionID: session,
|
||
TurnID: laneTurn,
|
||
Nodes: []trace.FlowNode{
|
||
{ID: "input", Label: "问", Kind: "input", Status: trace.FlowStatusDone},
|
||
{ID: "output", Label: "答", Kind: "output", Status: trace.FlowStatusDone},
|
||
},
|
||
}
|
||
}
|
||
|
||
// TestRecordRound_WritesLaneTurnID_正常 钉住写入侧:落盘时把这一路的回合标识写进
|
||
// `lane_turn_id`(迁移 0023),空标识如实写**空串而不是 NULL**(与 0018 的
|
||
// `chat_messages.turn_id` 同一口径),且二次收尾同轮覆盖时标识跟着更新。
|
||
func TestRecordRound_WritesLaneTurnID_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, conn := newTestStore(t, nil)
|
||
|
||
if err := st.RecordRound(laneRoundGraph("web", "web-1#turn-a", "turn-a")); err != nil {
|
||
t.Fatalf("RecordRound(带路标识): %v", err)
|
||
}
|
||
var (
|
||
lane string
|
||
isNul int
|
||
)
|
||
if err := conn.QueryRowContext(ctx,
|
||
`SELECT lane_turn_id, lane_turn_id IS NULL FROM flow_graphs WHERE session_id = ? AND turn_id = ?`,
|
||
"web", "web-1#turn-a").Scan(&lane, &isNul); err != nil {
|
||
t.Fatalf("读 lane_turn_id: %v", err)
|
||
}
|
||
if lane != "turn-a" {
|
||
t.Fatalf("lane_turn_id = %q, want turn-a", lane)
|
||
}
|
||
if isNul != 0 {
|
||
t.Fatal("lane_turn_id 绝不允许为 NULL(旧的「没写」必须与「空」同值)")
|
||
}
|
||
|
||
// 单路回合(图上没有路标识):如实写空串。
|
||
if err := st.RecordRound(laneRoundGraph("web", "web-2", "")); err != nil {
|
||
t.Fatalf("RecordRound(无路标识): %v", err)
|
||
}
|
||
if err := conn.QueryRowContext(ctx,
|
||
`SELECT lane_turn_id FROM flow_graphs WHERE session_id = ? AND turn_id = ?`, "web", "web-2").Scan(&lane); err != nil {
|
||
t.Fatalf("读单路行的 lane_turn_id: %v", err)
|
||
}
|
||
if lane != "" {
|
||
t.Fatalf("无路标识的行 lane_turn_id = %q, want 空串", lane)
|
||
}
|
||
|
||
// 二次收尾(同一图 id):一行,且标识跟着更新到最新那份。
|
||
if err := st.RecordRound(laneRoundGraph("web", "web-1#turn-a", "turn-a2")); err != nil {
|
||
t.Fatalf("RecordRound(二次收尾): %v", err)
|
||
}
|
||
var rows int
|
||
if err := conn.QueryRowContext(ctx,
|
||
`SELECT count(*) FROM flow_graphs WHERE session_id = ? AND turn_id = ?`, "web", "web-1#turn-a").Scan(&rows); err != nil {
|
||
t.Fatalf("count rows: %v", err)
|
||
}
|
||
if rows != 1 {
|
||
t.Fatalf("同一图 id 的行数 = %d,want 1(唯一键仍必须合并)", rows)
|
||
}
|
||
if err := conn.QueryRowContext(ctx,
|
||
`SELECT lane_turn_id FROM flow_graphs WHERE session_id = ? AND turn_id = ?`, "web", "web-1#turn-a").Scan(&lane); err != nil {
|
||
t.Fatalf("读二次收尾后的 lane_turn_id: %v", err)
|
||
}
|
||
if lane != "turn-a2" {
|
||
t.Fatalf("二次收尾后 lane_turn_id = %q, want turn-a2", lane)
|
||
}
|
||
}
|
||
|
||
// TestGraphsByTurnID_BatchAndMiss_正常 钉住读侧:一次批量查询把页内路标识换成图,
|
||
// **取不到的标识不出现在映射里**(调用方据此让 `toolCalls` 整体缺席,绝不回空数组),
|
||
// 且不跨会话串图。
|
||
func TestGraphsByTurnID_BatchAndMiss_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
for _, g := range []*trace.FlowGraph{
|
||
laneRoundGraph("web", "web-1#turn-a", "turn-a"),
|
||
laneRoundGraph("web", "web-2#turn-b", "turn-b"),
|
||
laneRoundGraph("sess-other", "sess-other-1#turn-c", "turn-c"),
|
||
} {
|
||
if err := st.RecordRound(g); err != nil {
|
||
t.Fatalf("RecordRound(%s): %v", g.ID, err)
|
||
}
|
||
}
|
||
|
||
got, err := st.GraphsByTurnID(ctx, "web", []string{"turn-a", "turn-b", "turn-missing"})
|
||
if err != nil {
|
||
t.Fatalf("GraphsByTurnID: %v", err)
|
||
}
|
||
if len(got) != 2 {
|
||
t.Fatalf("命中 %d 条,want 2(缺席的标识必须**不在**映射里): %v", len(got), keysOf(got))
|
||
}
|
||
if g := got["turn-a"]; g == nil || g.ID != "web-1#turn-a" || g.TurnID != "turn-a" {
|
||
t.Fatalf("turn-a 的图 = %+v,want web-1#turn-a", g)
|
||
}
|
||
if _, ok := got["turn-missing"]; ok {
|
||
t.Fatal("缺席的标识不得出现在映射里(那会让调用方以为查到了图)")
|
||
}
|
||
// 另一个会话的同名标识绝不串过来。
|
||
other, err := st.GraphsByTurnID(ctx, "web", []string{"turn-c"})
|
||
if err != nil {
|
||
t.Fatalf("GraphsByTurnID(别的会话的路): %v", err)
|
||
}
|
||
if len(other) != 0 {
|
||
t.Fatalf("跨会话命中了 %v,want 空(session_id 必须参与过滤)", keysOf(other))
|
||
}
|
||
|
||
// 空集合 / 全空白集合 / 空会话:都不发查询,返回空映射而不是错误。
|
||
for name, ids := range map[string][]string{
|
||
"nil 集合": nil,
|
||
"空集合": {},
|
||
"全空白": {"", " "},
|
||
"空会话": {"turn-a"},
|
||
} {
|
||
session := "web"
|
||
if name == "空会话" {
|
||
session = " "
|
||
}
|
||
empty, err := st.GraphsByTurnID(ctx, session, ids)
|
||
if err != nil {
|
||
t.Fatalf("%s: err = %v, want nil", name, err)
|
||
}
|
||
if len(empty) != 0 {
|
||
t.Fatalf("%s: 命中 %v,want 空", name, keysOf(empty))
|
||
}
|
||
}
|
||
|
||
// 重复标识只算一次,结果仍按原值可查。
|
||
dedup, err := st.GraphsByTurnID(ctx, "web", []string{"turn-a", "turn-a", " turn-a "})
|
||
if err != nil {
|
||
t.Fatalf("GraphsByTurnID(重复标识): %v", err)
|
||
}
|
||
if len(dedup) != 1 || dedup["turn-a"] == nil {
|
||
t.Fatalf("重复标识的结果 = %v,want 恰好 turn-a 一条", keysOf(dedup))
|
||
}
|
||
}
|
||
|
||
// TestGraphsByTurnID_OverChunk_边界 钉住分块上限之外仍能查全:绑定参数集合一旦超过
|
||
// 单次 `IN` 的上限就必须分块,否则整页读接口会因为参数上限报错。这里给 250 个标识
|
||
// (> maxTurnIDParams)而只有 2 个真有图。
|
||
func TestGraphsByTurnID_OverChunk_边界(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st, _ := newTestStore(t, nil)
|
||
|
||
if err := st.RecordRound(laneRoundGraph("web", "web-1#turn-a", "turn-a")); err != nil {
|
||
t.Fatalf("RecordRound: %v", err)
|
||
}
|
||
// 第 250 个标识故意放在最后一块里(分块写错时最容易漏掉末尾)。
|
||
const total = 250
|
||
ids := make([]string, 0, total)
|
||
for i := 0; i < total-1; i++ {
|
||
ids = append(ids, "turn-none-"+itoa(i))
|
||
}
|
||
ids = append(ids, "turn-a")
|
||
|
||
got, err := st.GraphsByTurnID(ctx, "web", ids)
|
||
if err != nil {
|
||
t.Fatalf("GraphsByTurnID(%d 个标识): %v", total, err)
|
||
}
|
||
if len(got) != 1 || got["turn-a"] == nil {
|
||
t.Fatalf("命中 %v,want 恰好 turn-a(最后一块也必须被查到)", keysOf(got))
|
||
}
|
||
}
|
||
|
||
// TestGraphsByTurnID_NilDB_异常 钉住「没有数据库」是明确的错误(ErrNilDB),与
|
||
// Graph / Latest 同一口径 —— 绝不回一个「查不到任何图」的空映射去冒充「没有工具」。
|
||
func TestGraphsByTurnID_NilDB_异常(t *testing.T) {
|
||
t.Parallel()
|
||
var nilStore *SQLStore
|
||
for name, st := range map[string]*SQLStore{"nil 接收者": nilStore, "空存储": {}} {
|
||
if _, err := st.GraphsByTurnID(context.Background(), "web", []string{"turn-a"}); !errors.Is(err, ErrNilDB) {
|
||
t.Fatalf("%s: err = %v, want ErrNilDB", name, err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// TestMigrate0023_BackfillLegacyRows_正常 钉住迁移 0023 对**老行**的处置:
|
||
// - v0.77 起写入、图 JSON 里带 `$.turnId` 的行 ⇒ 就地回填,不必重跑任何轮次;
|
||
// - 图里没有该键的行(v0.77 之前 / 单路回合)⇒ 如实留空串(查不到,`toolCalls` 缺席);
|
||
// - JSON 被写坏的行 ⇒ 保持空串而**不让整次迁移失败**(一条坏行不该让后端起不来);
|
||
// - 重复应用不覆盖已有值(`WHERE lane_turn_id = ”` 的幂等承诺)。
|
||
//
|
||
// 手法:把 0023 的列与记账行先撤回(等价于它的 down),插入老行,再跑一次真的
|
||
// `MigrateUp`。这样验的是**嵌入的那份迁移文件**,而不是测试里抄的一份 SQL。
|
||
func TestMigrate0023_BackfillLegacyRows_正常(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
_, conn := newTestStore(t, nil)
|
||
|
||
for _, stmt := range []string{
|
||
`DROP INDEX IF EXISTS idx_flow_graphs_lane_turn_id`,
|
||
`ALTER TABLE flow_graphs DROP COLUMN lane_turn_id`,
|
||
} {
|
||
if _, err := conn.ExecContext(ctx, stmt); err != nil {
|
||
t.Fatalf("撤回 0023 的结构(%s): %v", stmt, err)
|
||
}
|
||
}
|
||
if _, err := conn.ExecContext(ctx, `DELETE FROM schema_migrations WHERE id = ?`, laneTurnMigrationID); err != nil {
|
||
t.Fatalf("撤回 0023 的记账行: %v", err)
|
||
}
|
||
|
||
legacy := []struct {
|
||
graphID string
|
||
graphRaw string
|
||
wantLane string
|
||
}{
|
||
{"web-1#turn-a", `{"id":"web-1#turn-a","sessionId":"web","turnId":"turn-a"}`, "turn-a"},
|
||
{"web-2", `{"id":"web-2","sessionId":"web"}`, ""},
|
||
{"web-3", `{"id":"web-3","sessionId":"web","turnId":""}`, ""},
|
||
{"web-4", `not json at all`, ""},
|
||
}
|
||
for _, row := range legacy {
|
||
if _, err := conn.ExecContext(ctx,
|
||
`INSERT INTO flow_graphs (session_id, turn_id, graph_json) VALUES (?, ?, ?)`,
|
||
"web", row.graphID, row.graphRaw); err != nil {
|
||
t.Fatalf("插入老行 %s: %v", row.graphID, err)
|
||
}
|
||
}
|
||
|
||
if err := db.MigrateUp(ctx, conn, nil); err != nil {
|
||
t.Fatalf("重新应用 0023: %v", err)
|
||
}
|
||
for _, row := range legacy {
|
||
var lane string
|
||
if err := conn.QueryRowContext(ctx,
|
||
`SELECT lane_turn_id FROM flow_graphs WHERE session_id = ? AND turn_id = ?`,
|
||
"web", row.graphID).Scan(&lane); err != nil {
|
||
t.Fatalf("读回填后的 %s: %v", row.graphID, err)
|
||
}
|
||
if lane != row.wantLane {
|
||
t.Fatalf("%s 的 lane_turn_id = %q, want %q", row.graphID, lane, row.wantLane)
|
||
}
|
||
}
|
||
var idx int
|
||
if err := conn.QueryRowContext(ctx,
|
||
`SELECT count(*) FROM sqlite_master WHERE type = 'index' AND name = 'idx_flow_graphs_lane_turn_id'`).Scan(&idx); err != nil {
|
||
t.Fatalf("查索引: %v", err)
|
||
}
|
||
if idx != 1 {
|
||
t.Fatal("0023 必须建出 idx_flow_graphs_lane_turn_id(按路批量查图靠它)")
|
||
}
|
||
|
||
// 下-上往返(开发库上最常见的动作:回滚一步再前滚):结构恢复,且回填结果与
|
||
// 第一次**逐字相同** —— 回填的唯一依据是图 JSON,因此往返是幂等的,绝不会
|
||
// 因为「跑过第二次」而丢标识或把值改坏。
|
||
for _, stmt := range []string{
|
||
`DROP INDEX IF EXISTS idx_flow_graphs_lane_turn_id`,
|
||
`ALTER TABLE flow_graphs DROP COLUMN lane_turn_id`,
|
||
} {
|
||
if _, err := conn.ExecContext(ctx, stmt); err != nil {
|
||
t.Fatalf("往返撤回(%s): %v", stmt, err)
|
||
}
|
||
}
|
||
if _, err := conn.ExecContext(ctx, `DELETE FROM schema_migrations WHERE id = ?`, laneTurnMigrationID); err != nil {
|
||
t.Fatalf("往返撤回记账行: %v", err)
|
||
}
|
||
if err := db.MigrateUp(ctx, conn, nil); err != nil {
|
||
t.Fatalf("往返后再次应用 0023: %v", err)
|
||
}
|
||
for _, row := range legacy {
|
||
var lane string
|
||
if err := conn.QueryRowContext(ctx,
|
||
`SELECT lane_turn_id FROM flow_graphs WHERE session_id = ? AND turn_id = ?`,
|
||
"web", row.graphID).Scan(&lane); err != nil {
|
||
t.Fatalf("往返后读 %s: %v", row.graphID, err)
|
||
}
|
||
if lane != row.wantLane {
|
||
t.Fatalf("往返后 %s 的 lane_turn_id = %q, want %q(往返必须幂等)", row.graphID, lane, row.wantLane)
|
||
}
|
||
}
|
||
}
|
||
|
||
// keysOf 把映射的键排序后列出(失败信息里要看清「到底命中了哪几个」)。
|
||
func keysOf(m map[string]*trace.FlowGraph) []string {
|
||
out := make([]string, 0, len(m))
|
||
for k := range m {
|
||
out = append(out, k)
|
||
}
|
||
sort.Strings(out)
|
||
return out
|
||
}
|