feat(gateway): 面 1 补齐 —— 消息读取面 toolCalls[](工具回看)

前提由 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 字符(常量+依据)
- 只放图里已有的、已脱敏内容,不放宽脱敏通道
This commit is contained in:
AskaEth committed 2026-10-06 09:03:55 +08:00
1 parent 99a644d545
commit c907a34510
10 files changed
+1429 -16

No files matched your search

+313
View File
@@ -0,0 +1,313 @@
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
}