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。
227 lines
11 KiB
Go
227 lines
11 KiB
Go
package history
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
)
|
||
|
||
// 本文件归属聊天消息携带的**文档引用**（design/24 §2.4），以及 chat_messages
|
||
// 上「建表之后才引入的列」的前滚聚合（client_msg_id、document 与 turn_id）。
|
||
//
|
||
// document 列只存「引用 + 本回合注入统计」——存储名、嗅探所得 MIME、字节数、注入
|
||
// 字符数、注入是否被截断、抽取状态；**绝不存抽取正文**。G3（base64 原样落库 →
|
||
// 每页历史背着重传）与 H1（每回合重算）两条教训说的都是同一件事：正文一旦进了
|
||
// 历史，每读一页都要为它付一次代价，而且它是提示词注入面（design/11 §10）。
|
||
//
|
||
// 列的内容是**引用数组**（≤maxDocumentsPerMessage 条），因为一个回合可以挂
|
||
// 多个附件（design/24 §6.1 的定稿上限 5）；S2/S3 早期写下的**单对象**行仍能
|
||
// 读回（见 DecodeDocuments）。列名与 Message.Document 字段名保持不变。
|
||
|
||
// 文档在持久化与历史 DTO 里可能处于的抽取状态（design/24 §6.1），由装配期
|
||
// 与抽取器填入。
|
||
const (
|
||
// DocumentStatusExtracted 表示正文已抽取并进了上下文。
|
||
DocumentStatusExtracted = "extracted"
|
||
// DocumentStatusListed 表示只给了模型一份附件清单，正文没有进上下文。
|
||
DocumentStatusListed = "listed"
|
||
// DocumentStatusUnreadable 表示读不到正文（抽取失败、类型不可解析）。
|
||
DocumentStatusUnreadable = "unreadable"
|
||
)
|
||
|
||
// maxDocumentsPerMessage 是一条消息（一个回合）携带的文档引用数上限
|
||
// （design/24 §6.1/§9.4-3 的定稿值 5）。它与网关的 maxChatDocuments **同值
|
||
// 同义**：这里刻意不复用那个常量 —— 网关依赖本包，反向依赖会成环。两个包
|
||
// 各自具名并互相交叉引用，任一侧改动都必须同时改另一侧（网关侧另有
|
||
// TestMaxChatDocumentsMatchesDesign 把值钉在设计定稿上）。
|
||
const maxDocumentsPerMessage = 5
|
||
|
||
// 区分 document 列读写失败的哨兵错误。调用方用 errors.Is 判类，并按「不静默」
|
||
// 处理它：打一条带来源的 WARN 并省掉该条引用，**绝不**把它当成「这条消息没有
|
||
// 附件」—— 一方以为附件传上去了、另一方一无所知，正是 F2 要消灭的误解。
|
||
var (
|
||
// ErrBadDocuments 报告 document 列的内容不是合法的引用数据（乱码、非法
|
||
// JSON、无名引用等）。
|
||
ErrBadDocuments = errors.New("history: malformed document column")
|
||
|
||
// ErrTooManyDocuments 报告一次读写里的引用数超过 maxDocumentsPerMessage。
|
||
ErrTooManyDocuments = errors.New("history: too many document references")
|
||
)
|
||
|
||
// DocumentRef 是一条消息携带的文档引用与抽取统计。它就是 chat_messages.document
|
||
// 列里的 JSON 小对象，**同时**也是历史 DTO 里 documents 数组的元素形状：两边共用
|
||
// 同一个结构体，所以「不回传正文」是结构上保证的，而不是靠约定或代码审查。
|
||
type DocumentRef struct {
|
||
// Name 是服务端存储名（design/24 §3.2：32 位十六进制 + 白名单扩展名）。
|
||
Name string `json:"name"`
|
||
// MIME 是服务端字节嗅探所得，绝不是客户端的声明。
|
||
MIME string `json:"mime"`
|
||
// Size 是该文档的字节数。
|
||
Size int64 `json:"size"`
|
||
// Chars 是**本回合注入**上下文的字符数（被注入预算截断时填截断后的值，
|
||
// 即真正进了模型可见视图的那个数）——不是抽取侧的正文长度。
|
||
// 全文长度只有读侧出口能看到（document-read 的 totalChars）。
|
||
Chars int `json:"chars,omitempty"`
|
||
// Truncated 报告**本回合的注入**是否被预算截断（design/24 §4.3：截断必须
|
||
// 诚实标注，绝不静默）——不是「文件是否被截断」。抽取期自己撞上 sidecar
|
||
// 安全上限（documents.MaxSidecarBytes）是另一回事，由 document-read 的
|
||
// extractTruncated 专指。
|
||
Truncated bool `json:"truncated,omitempty"`
|
||
// Status 是抽取状态，取值见 DocumentStatus* 常量。
|
||
Status string `json:"status,omitempty"`
|
||
// Reason 是 Status == unreadable 时**读不到正文的原因**（design/24 §4.4 的
|
||
// 诚实原因文案），供客户端直接把服务端的说法展示给用户 —— 客户端因此不必
|
||
// 自己拼一句服务端没说过的话（契约 v0.63）。它**只在「读不到」时下发**
|
||
// （extracted / listed 时为空 ⇒ 该键从 DTO 里消失）。
|
||
//
|
||
// 内容纪律：**只有文案**。不含落盘路径、不含任何凭据、不含存储名以外的标识
|
||
// —— 它就是同一份注入给模型的原因短句（装配期与注入共用同一个字符串），原始
|
||
// 错误（可能带路径）只进日志，绝不透出。
|
||
Reason string `json:"reason,omitempty"`
|
||
}
|
||
|
||
// EncodeDocuments 把一条消息携带的文档引用编码成落库用的 JSON 数组。空列表返回
|
||
// ""，使「这条消息没有附件」与「有附件」在库里、在 DTO 里都有确定区别。
|
||
//
|
||
// 超限与无名引用都**报错**而不是就地裁剪：悄悄砍掉几条、或把一条没有名字的引用
|
||
// （它的 URL 只能从名字推出）塞进数组，都会让「客户端传上去的附件」与「库里记着
|
||
// 的附件」对不上 —— 那正是本列要消灭的静默。
|
||
func EncodeDocuments(refs []DocumentRef) (string, error) {
|
||
if len(refs) == 0 {
|
||
return "", nil
|
||
}
|
||
if len(refs) > maxDocumentsPerMessage {
|
||
return "", fmt.Errorf("history: encode %d documents: %w", len(refs), ErrTooManyDocuments)
|
||
}
|
||
for _, ref := range refs {
|
||
if ref.Name == "" {
|
||
return "", fmt.Errorf("history: encode document ref without name: %w", ErrBadDocuments)
|
||
}
|
||
}
|
||
b, err := json.Marshal(refs)
|
||
if err != nil {
|
||
return "", fmt.Errorf("history: encode document refs: %w", err)
|
||
}
|
||
return string(b), nil
|
||
}
|
||
|
||
// DecodeDocuments 解码落库的文档引用。空串（这条消息没有附件）返回 (nil, nil)；
|
||
// 坏内容返回包装了 ErrBadDocuments / ErrTooManyDocuments 的错误，由调用方决定
|
||
// 如何**可见地**报告 —— 解析失败绝不静默当成「没有附件」，那正是 F2 要消灭的
|
||
// 那种「一方以为传上去了，另一方一无所知」。
|
||
//
|
||
// 它兼容 S2/S3 早期写下的**单对象形状**：内容的首字符是 '{' 就按「单元素数组」
|
||
// 处理。那些行里带着一条真实引用，绝不能因为形状老就被当成坏数据丢掉。
|
||
func DecodeDocuments(raw string) ([]DocumentRef, error) {
|
||
trimmed := strings.TrimSpace(raw)
|
||
if trimmed == "" {
|
||
return nil, nil
|
||
}
|
||
refs, err := decodeDocumentData(trimmed)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(refs) > maxDocumentsPerMessage {
|
||
return nil, fmt.Errorf("history: decode %d documents: %w", len(refs), ErrTooManyDocuments)
|
||
}
|
||
return refs, nil
|
||
}
|
||
|
||
// decodeDocumentData 按内容形状解码：'{' 开头是旧单对象形状（当成单元素数组），
|
||
// 其余按数组解析。空数组与字面量 null 都解出空结果，等价于「这条消息没有附件」。
|
||
func decodeDocumentData(raw string) ([]DocumentRef, error) {
|
||
if raw[0] == '{' {
|
||
var ref DocumentRef
|
||
if err := json.Unmarshal([]byte(raw), &ref); err != nil {
|
||
return nil, fmt.Errorf("history: decode document ref: %w", ErrBadDocuments)
|
||
}
|
||
return []DocumentRef{ref}, nil
|
||
}
|
||
var refs []DocumentRef
|
||
if err := json.Unmarshal([]byte(raw), &refs); err != nil {
|
||
return nil, fmt.Errorf("history: decode document refs: %w", ErrBadDocuments)
|
||
}
|
||
return refs, nil
|
||
}
|
||
|
||
// UpdateMessageDocument 只替换 id 所指消息的 document 列（引用数组），其余任何
|
||
// 列都保持已存值。它是「回合结束后补抽取结论」的写入端（design/24 §6.1）。
|
||
//
|
||
// 为什么需要它：用户消息在回合**开始前**就已落库，那时装配期的抽取还没发生，
|
||
// 因此列里只有引用、没有 status/chars/truncated。结论只能在回合结束后补进同一
|
||
// 列 —— 两段式是刻意的，也是客户端能看到「昔涟读不到这个文件」的唯一路径。
|
||
//
|
||
// 未知 id 返回 ErrMessageNotFound（与 UpdateMessage 同一口径），使调用方能区分
|
||
// 「那一行不见了」与「写失败」，并据此打带来源的 WARN，而不是静默当作写完了。
|
||
//
|
||
// 引用计数按**新旧名字集合的差**在同一事务里增减（切片 S7，见 refs.go）：这条
|
||
// 路径的常态是「同一批名字、只补 status/chars/truncated」，那时加减都是空；只有
|
||
// 名字集合真的变了才动计数。旧列读不出来（行不存在·读失败）时先于 UPDATE 返回，
|
||
// 绝不把一次「可能多算」的写留给下一次 GC。
|
||
func (s *Store) UpdateMessageDocument(ctx context.Context, id, document string) error {
|
||
tx, err := s.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return fmt.Errorf("history: update message document %s: %w", id, err)
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
var previous string
|
||
err = tx.QueryRowContext(ctx, `SELECT document FROM chat_messages WHERE id = ?`, id).Scan(&previous)
|
||
if err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return fmt.Errorf("history: update message document %s: %w", id, ErrMessageNotFound)
|
||
}
|
||
return fmt.Errorf("history: read message document %s: %w", id, err)
|
||
}
|
||
res, err := tx.ExecContext(ctx, `UPDATE chat_messages SET document = ? WHERE id = ?`, document, id)
|
||
if err != nil {
|
||
return fmt.Errorf("history: update message document %s: %w", id, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return fmt.Errorf("history: update message document %s rows: %w", id, err)
|
||
}
|
||
if n == 0 {
|
||
return fmt.Errorf("history: update message document %s: %w", id, ErrMessageNotFound)
|
||
}
|
||
if err := s.syncDocumentRefs(ctx, tx, previous, document); err != nil {
|
||
return err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return fmt.Errorf("history: update message document %s: %w", id, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ensureMessageColumns 把 chat_messages 上「建表之后才引入的列」前滚到既有表。
|
||
// 全新库由 createMessagesTable 一次建全，因此这里对它们都是空操作；早于某列
|
||
// 建出的表则在这里补上（与 memory 包的 ensureProjectIDColumn 同一手法）。
|
||
func ensureMessageColumns(ctx context.Context, db *sql.DB) error {
|
||
if err := ensureClientMsgIDColumn(ctx, db); err != nil {
|
||
return err
|
||
}
|
||
if err := ensureDocumentColumn(ctx, db); err != nil {
|
||
return err
|
||
}
|
||
return ensureTurnIDColumn(ctx, db)
|
||
}
|
||
|
||
// ensureDocumentColumn 在 chat_messages 早于本列（0015 迁移）建表时把它前滚：
|
||
// 缺列则补加，已有则忽略重复列错误。它是幂等的，因此 NewStore 可在全新 schema
|
||
// 或既有 schema 上调用。
|
||
func ensureDocumentColumn(ctx context.Context, db *sql.DB) error {
|
||
_, err := db.ExecContext(ctx, `ALTER TABLE chat_messages
|
||
ADD COLUMN document TEXT NOT NULL DEFAULT ''`)
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
// 列已存在时（由 createMessagesTable 建出，或上一次运行补过）SQLite 报
|
||
// "duplicate column name: document"。
|
||
if strings.Contains(err.Error(), "duplicate column name") {
|
||
return nil
|
||
}
|
||
return fmt.Errorf("history: roll forward document column: %w", err)
|
||
}
|