AskaEth
679cee04f7
根因：摄取结果只挂到「内存里的模型可见消息」（inbound.go 的 attachInboundAttachments）， 落库时只传了 text、history.Message 没有 Document 字段 ⇒ 该列恒空 ⇒ 读侧 omitempty 导致 documents 键整个缺席 ⇒ 任何客户端只看到一条空消息。Web 侧不这样，是因为 gateway/session_persist.go 有 messageTextImageDocument 把 ContentTypeDocument 部件编码进该列 —— IM 路径缺的就是同一步。 修法：inboundAttachments.documentColumn() 复用 history.EncodeDocuments 同一口径写出该列； persistToMain/persistIsolated 增 document 参数并返回刚落库的消息；新增两段式第二段 persistTurnDocumentStatus 按 id 回写抽取结论（只补结论、按存储名匹配、匹配不到原样留）， 与 gateway/session_document_status.go 逐字对齐。 契约零扩张：无新端点/帧/字段；0 附件 ⇒ 列空 ⇒ 键省略；形状即 web 的同一结构体。 测试 6 条（含元素 JSON 逐字断言、unreadable 带 reason、被拒不编造芯片、群聊隔离分支、 真循环端到端）；变异两条（还原挂载 / 关闭结论回写）⇒ 各 4 条红。
110 lines
5.1 KiB
Go
110 lines
5.1 KiB
Go
package channelbridge
|
|
|
|
// 本文件归属 IM 回合的**文档抽取结论回写**（design/24 §6.1 的两段式）：入站附件的
|
|
// 用户消息在回合**开始前**就落了库，那时装配期的抽取还没发生，`document` 列里只有
|
|
// 引用。结论（status/chars/truncated/reason）只能在回合跑起来之后补进同一列 —— 与
|
|
// Web 侧 gateway/session_document_status.go 逐项同构，两端客户端因此读到同一份事实。
|
|
//
|
|
// 为什么另写一份而不是共用：那份实现在网关包内且未导出，而本包既不许反向依赖网关，
|
|
// 也不该把「网关的会话编排」搬进渠道桥。合并口径与三条纪律逐字对齐（见 mergeDocumentStatus）：
|
|
// - 只补结论，身份（存储名/MIME/字节数）逐字保留；
|
|
// - 按**服务端存储名**匹配，匹配不到就原样留着（凭空给它一个结论就是编造）；
|
|
// - 找不到行、存储未实现该视图、写库失败一律带来源 WARN，绝不静默。
|
|
|
|
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"
|
|
)
|
|
|
|
// documentStatusLogSource 是文档状态回写的日志来源标签，使这条冷路径上的 WARN
|
|
// 能被追溯到具体子功能（AGENTS §10 的「日志必须标来源」）。
|
|
const documentStatusLogSource = "services/channelbridge/inbound-documents"
|
|
|
|
// documentStatusWriter 是 HistoryStore 的一个**可选**更窄视图（与网关侧同名接口
|
|
// 同一做法）：按消息 id 更新 `document` 列。未实现该视图的替身或后端退回「不回写」
|
|
// 并留下带来源的 WARN，绝不因此让回合失败 —— 状态回传是增强，不能反过来卡住会话。
|
|
type documentStatusWriter interface {
|
|
UpdateMessageDocument(ctx context.Context, id, document string) error
|
|
}
|
|
|
|
// persistTurnDocumentStatus 在回合结束后把装配期解析到的文档结论补进该回合用户
|
|
// 消息的 `document` 列（design/24 §4.4/§6.1）。它由 runInboundTurn 在循环返回后
|
|
// 调用，因此「引用 + 结论」最终是同一列里的一份事实，客户端不必自己拼。
|
|
//
|
|
// 三个提前返回都是**正常**情形，不刷日志：这条回合没有附件引用（msg.Document 为空
|
|
// ⇒ 根本没落过列）、没有落库（msg.ID 为空 ⇒ 存储未接或写库失败）、本回合没有解析
|
|
// 到任何文档结论（res 为 nil 或 Documents 为空 ⇒ 循环报错或没有文档部件）。
|
|
//
|
|
// 写入使用全新的后台 context（与网关 persistDocumentStatus 同一理由）：被取消回合
|
|
// 的 ctx 绝不该让这次补写半途而废。
|
|
func (b *Bridge) persistTurnDocumentStatus(msg history.Message, res *loop.Result) {
|
|
if msg.ID == "" || msg.Document == "" || res == nil || len(res.Documents) == 0 {
|
|
return
|
|
}
|
|
log := logger.WithSource(b.log, documentStatusLogSource)
|
|
writer, ok := b.history.(documentStatusWriter)
|
|
if !ok {
|
|
log.Warn("document status not persisted",
|
|
"reason", "history store has no document update path", "message", msg.ID, "documents", len(res.Documents))
|
|
return
|
|
}
|
|
merged, changed, err := mergeDocumentStatus(msg.Document, res.Documents)
|
|
if err != nil {
|
|
log.Warn("document status merge", "err", err, "message", msg.ID)
|
|
return
|
|
}
|
|
if !changed {
|
|
return
|
|
}
|
|
if err := writer.UpdateMessageDocument(context.Background(), msg.ID, merged); err != nil {
|
|
log.Warn("document status update", "err", err, "message", msg.ID)
|
|
}
|
|
}
|
|
|
|
// mergeDocumentStatus 把本回合解析到的结论按存储名并进已落库的引用数组：匹配到的
|
|
// 补上 status/chars/truncated/reason，没匹配到的原样保留（那意味着本回合没有解析到
|
|
// 它，凭空给它一个结论就是编造）。第二个返回值报告是否真的补进了新结论 —— 没变化
|
|
// 就不必写库、也不必刷一条「什么都没变」的日志。
|
|
//
|
|
// reason 与 status 同批搬运（契约 v0.63）：抽取结论这三项 + 原因文案是同一个回合的
|
|
// 同一份事实，分两次写会让客户端看到「unreadable 但没有理由」的中间态。
|
|
//
|
|
// 列被写坏时返回包装了 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 && refs[i].Reason == got.Reason {
|
|
continue
|
|
}
|
|
refs[i].Status, refs[i].Chars, refs[i].Truncated, refs[i].Reason =
|
|
got.Status, got.Chars, got.Truncated, got.Reason
|
|
changed = true
|
|
}
|
|
if !changed {
|
|
return raw, false, nil
|
|
}
|
|
merged, err := history.EncodeDocuments(refs)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
return merged, true, nil
|
|
}
|