feat(gateway): 文档多值化 + 抽取结论回写（design/24 S3.5）
- 缺口①：「一回合最多 5 个附件、落库只留 1 个」——document 列语义升级为 ≤5 个引用的**数组**（列名/字段名不变、无需迁移，旧单对象行可读回）；落库收集全部引用、转录还原全部部件。 - 缺口②：抽取结论 status/chars/truncated 经 completeTurn 回合后**回写**该回合用户消息的同一列（id 按代次配对，忙时消息挂 pendingMsg.persisted 并在 drain 时以新 gen 重新配对）；历史 DTO 改为 **documents: []**（omitempty ⇒ 0 个附件时省略）。 - 防御：编/解码超 5 都报 ErrTooManyDocuments（报错、绝不裁剪）；坏列走既有不静默路径（转录 WARN + DTO unreadable 最小引用）；>5 上限值被独立用具例钉死。 - 顺手修 S3-B 遗留的函数越界：submitInternal 81 → 79 行（播种块抽成 seedTurnLocked）；session_orchestrator.go 484 行。 - 主会话独立验收：隔离 worktree（HEAD + 本切片 12 文件）build/vet 绿、全树 59 包 0 FAIL、linux+darwin 交叉编译绿、gateway+history 437 PASS；变异独立复现（只留第一个附件 ⇒ 2 条红，含 HTTP 端到端 want 3 条）。 - 遗留（已登记）：chat_documents_test.go 510 行（S3-B 遗留越界，本次未拆）；中断/被取代的回合不回写结论；回写后不补发实时帧（status 需重新拉取/回放可见）。
This commit is contained in:
AskaEth committed
1 parent
20c1edb36a
commit
5b2b385524
12 files changed
+1255
-166
No files matched your search
@@ -0,0 +1,190 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
)
|
||||
|
||||
// 本文件归属**文档抽取结论的回写闭环**（design/24 §4.4/§6.1，S3.5 缺口②）。
|
||||
//
|
||||
// 用户消息在回合**开始前**落库（session_persist.go 的 persistUserMessages），
|
||||
// 那时文档抽取还没发生，document 列里只有引用、没有结论。抽取结论要等回合跑
|
||||
// 起来（loop 的装配期解析）才存在，于是本文件负责在**回合结束后**把它补回同一
|
||||
// 列 —— 两段式是刻意的，也是客户端能看到「昔涟读不到这个文件」的唯一路径。
|
||||
//
|
||||
// 三条纪律：
|
||||
// - 只补结论（status/chars/truncated），身份与大小逐字保留；
|
||||
// - 按**服务端存储名**匹配，匹配不到就原样留着（凭空给它一个结论就是编造）；
|
||||
// - 找不到行、存储未实现该视图、写库失败一律带来源 WARN，绝不静默。
|
||||
//
|
||||
// 它从 session_persist.go 拆出来，使那两个文件都留在「单文件 ≤500 行」的硬约束
|
||||
// 内（与 session_events.go / session_tail.go 同一理由）。
|
||||
|
||||
// documentStatusLogSource 是文档状态回写的日志来源标签，使这条冷路径上的
|
||||
// WARN 能被追溯到具体子功能（AGENTS §10 的「日志必须标来源」）。
|
||||
const documentStatusLogSource = "gateway/session-persist"
|
||||
|
||||
// turnDocuments 配对一次回合同它已落库的用户消息与所属代次（gen）。代次是
|
||||
// 必须的：被打断或被新回合取代的陈旧回合在结算时不该去写别人的那一行。
|
||||
type turnDocuments struct {
|
||||
gen uint64
|
||||
msgs []history.Message
|
||||
}
|
||||
|
||||
// documentStatusWriter 是 HistoryStore 的一个**可选**更窄视图（与
|
||||
// clientMsgIDReader / transcriptReader 同一做法）：按消息 id 更新 document 列。
|
||||
// 未实现该视图的存储替身或后端退回「不回写」并留下带来源的 WARN，不因此让
|
||||
// 回合失败 —— 状态回传是增强，绝不能反过来卡住会话。
|
||||
type documentStatusWriter interface {
|
||||
UpdateMessageDocument(ctx context.Context, id, document string) error
|
||||
}
|
||||
|
||||
// rememberTurnDocuments 记下本回合已落库的用户消息（按代次配对），供回合结束
|
||||
// 后把抽取结论回写进它们的 document 列。空列表是空操作（这条回合没有落库任何
|
||||
// 用户消息，也就没有结论可补）。
|
||||
func (o *Orchestrator) rememberTurnDocuments(sessionID string, gen uint64, msgs []history.Message) {
|
||||
if len(msgs) == 0 {
|
||||
return
|
||||
}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
if st := o.states[sessionID]; st != nil {
|
||||
st.turnDocs = turnDocuments{gen: gen, msgs: msgs}
|
||||
}
|
||||
}
|
||||
|
||||
// attachPendingDocuments 把刚落库的用户消息挂到队列里 seq 所指的那条待处理回合
|
||||
// 上，使它的续跑回合结算时能回写结论（它在入队那一刻就落了库，而那时还没有
|
||||
// 任何结论）。
|
||||
//
|
||||
// 队列里已经找不到那条（被中断清空、已被弹出或已被续跑消费）时什么都不做：
|
||||
// 那一行的结论没有谁能再补，也没有必要为此刷一条 WARN —— 它不是缺陷，是中断
|
||||
// 的正常后果。
|
||||
func (o *Orchestrator) attachPendingDocuments(sessionID string, seq uint64, msgs []history.Message) {
|
||||
if len(msgs) == 0 {
|
||||
return
|
||||
}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
st := o.states[sessionID]
|
||||
if st == nil {
|
||||
return
|
||||
}
|
||||
for i := range st.pending {
|
||||
if st.pending[i].seq == seq {
|
||||
st.pending[i].persisted = msgs
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// takeTurnDocuments 取走与 gen 配对的已落库用户消息并清空记录。第二个返回值
|
||||
// 报告这次代次是否仍是当代：非当代说明这是一次陈旧回合的迟到结算（中断/取代），
|
||||
// 调用方据此静默返回 —— 迟到的结算是正常竞态，不是缺陷。
|
||||
func (o *Orchestrator) takeTurnDocuments(sessionID string, gen uint64) ([]history.Message, bool) {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
st := o.states[sessionID]
|
||||
if st == nil {
|
||||
return nil, true
|
||||
}
|
||||
if st.turnDocs.gen != gen {
|
||||
return nil, false
|
||||
}
|
||||
msgs := st.turnDocs.msgs
|
||||
st.turnDocs = turnDocuments{}
|
||||
return msgs, true
|
||||
}
|
||||
|
||||
// persistDocumentStatus 在回合结束后把装配期解析到的文档结论回写进该回合用户
|
||||
// 消息的 document 列（design/24 §4.4/§6.1）。它由 completeTurn 在助手回复落库
|
||||
// 之后、状态机迁移之前调用，因此结论与该回合的对话内容一起同批入账。
|
||||
//
|
||||
// 找不到对应行（ErrMessageNotFound）、存储未接该视图、列被写坏、写库失败时都
|
||||
// **不静默**：打一条带来源的 WARN 说明「状态未回写」。日志只带会话、消息 id、
|
||||
// 条数与错误，**绝不带正文**，也不带列内容。
|
||||
//
|
||||
// 写入使用全新的后台 context（与 persistAssistantReply 同一理由）：被取消回合
|
||||
// 的 ctx 绝不该中断这次补写。
|
||||
func (o *Orchestrator) persistDocumentStatus(sessionID string, gen uint64, res *loop.Result) {
|
||||
if res == nil || len(res.Documents) == 0 {
|
||||
return
|
||||
}
|
||||
log := logger.WithSource(o.log, documentStatusLogSource)
|
||||
if o.history == nil {
|
||||
log.Warn("document status not persisted", "reason", "no history store", "session", sessionID, "documents", len(res.Documents))
|
||||
return
|
||||
}
|
||||
writer, ok := o.history.(documentStatusWriter)
|
||||
if !ok {
|
||||
log.Warn("document status not persisted", "reason", "history store has no document update path", "session", sessionID, "documents", len(res.Documents))
|
||||
return
|
||||
}
|
||||
msgs, current := o.takeTurnDocuments(sessionID, gen)
|
||||
if !current {
|
||||
return
|
||||
}
|
||||
if len(msgs) == 0 {
|
||||
log.Warn("document status not persisted", "reason", "no persisted user message for this turn", "session", sessionID, "documents", len(res.Documents))
|
||||
return
|
||||
}
|
||||
ctx := context.Background()
|
||||
for _, m := range msgs {
|
||||
if m.Document == "" {
|
||||
continue
|
||||
}
|
||||
merged, changed, err := mergeDocumentStatus(m.Document, res.Documents)
|
||||
if err != nil {
|
||||
log.Warn("document status merge", "err", err, "session", sessionID, "message", m.ID)
|
||||
continue
|
||||
}
|
||||
if !changed {
|
||||
continue
|
||||
}
|
||||
if err := writer.UpdateMessageDocument(ctx, m.ID, merged); err != nil {
|
||||
log.Warn("document status update", "err", err, "session", sessionID, "message", m.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// mergeDocumentStatus 把本回合解析到的结论按存储名并进已落库的引用数组：匹配到
|
||||
// 的补上 status/chars/truncated，没匹配到的原样保留（那意味着本回合没有解析到
|
||||
// 它，凭空给它一个结论就是编造）。第二个返回值报告是否真的补进了新结论 ——
|
||||
// 没变化就不必写库、也不必刷一条「什么都没变」的日志。
|
||||
//
|
||||
// 列被写坏时返回包装了 history 哨兵的错误，由调用方按「不静默」处理。
|
||||
func mergeDocumentStatus(raw string, resolved []history.DocumentRef) (string, bool, error) {
|
||||
refs, err := history.DecodeDocuments(raw)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
conclusion := make(map[string]history.DocumentRef, len(resolved))
|
||||
for _, ref := range resolved {
|
||||
if ref.Name != "" {
|
||||
conclusion[ref.Name] = ref
|
||||
}
|
||||
}
|
||||
changed := false
|
||||
for i := range refs {
|
||||
got, ok := conclusion[refs[i].Name]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if refs[i].Status == got.Status && refs[i].Chars == got.Chars && refs[i].Truncated == got.Truncated {
|
||||
continue
|
||||
}
|
||||
refs[i].Status, refs[i].Chars, refs[i].Truncated = got.Status, got.Chars, got.Truncated
|
||||
changed = true
|
||||
}
|
||||
if !changed {
|
||||
return raw, false, nil
|
||||
}
|
||||
merged, err := history.EncodeDocuments(refs)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
return merged, true, nil
|
||||
}
|
||||
Reference in new issue
Block a user