fix(channelbridge): IM 入站附件引用写进落库的用户消息(安卓报的 P0)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

根因:摄取结果只挂到「内存里的模型可见消息」(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 条红。
This commit is contained in:
AskaEth committed 2026-10-05 08:10:21 +08:00
1 parent f8f2e9c54c
commit 679cee04f7
7 files changed
+606 -19

No files matched your search

+17 -5
View File
@@ -225,7 +225,8 @@ func inboundTargetFor(sess session.Session, platform string) inboundTarget {
// runInboundTurn 为被放行的消息驱动 agent 轮次:解析
// 路由目标、把消息携带的文件类附件**落库并挂成文档部件**
// (design/24 §3;切片 M1)、构建模型可见的消息列表、在轮次之前持久化入站
// 用户消息(因此即使渠道回复失败它也已落地),运行
// 用户消息**连同它们的 `document` 引用列**(因此即使渠道回复失败它也已落地,
// 客户端也有附件芯片可点),并在轮次之后把抽取结论补回同一列,运行
// 循环并把结果交给 deliverReply。
func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter,
text string, actor adapter.Actor, sess session.Session) {
@@ -252,14 +253,21 @@ func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage,
loopMsgs := b.buildLoopMessages(ctx, text, runEnv, actor, senderName, msg)
attachInboundAttachments(loopMsgs, attachments)
// 附件引用必须同批落进用户消息的 `document` 列(design/24 §2.4):只把部件挂在
// 内存里的模型可见消息上,客户端读到的那条消息就是一片空白 —— 附件芯片只认库
// 里的那一列。这里写的是引用与身份,结论要等回合跑完(下一段)。
document := b.inboundDocumentColumn(msg, attachments)
// 主人私聊消息落在 Web 主会话并广播给 Web 客户端;
// 受限消息落在其隔离会话(适配器日志页),
// 以 `channel_persisted` 事件广播(绝不进入
// 主/`web` 会话或 Web 聊天)。
// 无论有没有文本都要落库:附件就是纯文件消息的内容(design/24 §4.4)。
var userMsg history.Message
if target.private {
b.persistToMain(ctx, history.RoleUser, text, "", target.platform, deviceIDFor(msg), adapterName(a))
userMsg = b.persistToMain(ctx, history.RoleUser, text, "", document, target.platform, deviceIDFor(msg), adapterName(a))
} else {
b.persistIsolated(ctx, history.RoleUser, text, deviceIDFor(msg), sess, msg.Platform, senderName, adapterName(a))
userMsg = b.persistIsolated(ctx, history.RoleUser, text, document, deviceIDFor(msg), sess, msg.Platform, senderName, adapterName(a))
}
res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), target.sessionID, loopMsgs)
@@ -267,6 +275,10 @@ func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage,
b.log.Warn("agent run failed", "platform", msg.Platform,
"session", target.sessionID, "error", err)
}
// 抽取结论(status/chars/truncated/reason)只能在回合跑起来之后才有,这里按
// 服务端存储名把结论补回刚落库的那一行(design/24 §6.1 的两段式,与 Web 侧
// gateway/session_document_status.go 同一口径)。
b.persistTurnDocumentStatus(userMsg, res)
// 出错也交给 deliverReply:回合不能因为「模型那边出了问题」就对用户静默
// (空回复 / 轮次用尽 / 内部报错都落一条诚实的可见回复;被取消的回合除外,
// 见 turnreply.go)。
@@ -297,9 +309,9 @@ func (b *Bridge) deliverReply(ctx context.Context, msg adapter.UnifiedMessage, a
// 广播;受限 → 隔离日志 + channel_persisted 广播),随后
// 送回渠道。
if target.private {
b.persistToMain(ctx, history.RoleAssistant, reply, mood, target.platform, deviceIDFor(msg), adapterName(a))
b.persistToMain(ctx, history.RoleAssistant, reply, mood, "", target.platform, deviceIDFor(msg), adapterName(a))
} else {
b.persistIsolated(ctx, history.RoleAssistant, reply, "", sess, msg.Platform, "", adapterName(a))
b.persistIsolated(ctx, history.RoleAssistant, reply, "", "", sess, msg.Platform, "", adapterName(a))
}
if err := b.sendReply(ctx, msg, a, reply); err != nil {
if errors.Is(err, adapter.ErrUnsupported) {
@@ -5,6 +5,11 @@ package channelbridge
// 上传同一份实现),成功者挂成该消息的文档部件,让装配期 `resolveDocuments`
// 照既有链路完成抽取/注入/清单/结论回写,**不需要新链路**。
//
// 同样的产物还必须是**用户消息落库时的 `document` 列**(documentColumn):只把部件
// 挂在内存里的模型可见消息上,客户端读到的那条消息就是一片空白 —— 附件芯片只认
// 库里的那一列。结论(status/chars/truncated/reason)随后由
// persistTurnDocumentStatus 补进同一列(inbound_documents_persist.go)。
//
// 读不到的附件**绝不静默**(design/24 §4.4):模型会收到一行诚实提示,日志留一条
// 带来源的记录(只记 name/mime/size,**绝不记正文**)。
//
@@ -26,6 +31,7 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)
@@ -81,6 +87,44 @@ func (a inboundAttachments) noticeText() string {
return strings.Join(a.notices, "\n")
}
// documentColumn 把本次摄取产出的文档部件编码成落库用的 `document` 列(design/24
// §2.4 的引用数组)。它与 Web 上传走**同一个字段、同一个结构体**:history.DocumentRef
// 同时就是历史 DTO 里 `documents` 数组的元素形状,因此两边客户端读到的附件芯片
// 逐字一致(名字/MIME/大小取自服务端,客户端给的原始文件名与声明类型都不在其中)。
//
// 与网关侧 documentRefOf(session_persist.go)是同一份映射:那里未导出、且本包
// 不许反向依赖网关,因此这里按同一口径重写一份 —— 两侧任一改动都必须同时改。
//
// 这里只落**引用**,不落结论:chars/truncated/status 是**本回合注入侧**的事实,
// 要等回合跑起来(装配期解析)才存在,由 persistTurnDocumentStatus 在回合结束后
// 补进同一列。没有附件时返回空串(与 EncodeDocuments 对空列表的口径一致:空串
// ⇒ DTO 里该键省略)。
func (a inboundAttachments) documentColumn() (string, error) {
refs := make([]history.DocumentRef, 0, len(a.parts))
for _, c := range a.parts {
refs = append(refs, history.DocumentRef{Name: c.DocumentName, MIME: c.DocumentMIME, Size: c.DocumentSize})
}
return history.EncodeDocuments(refs)
}
// inboundDocumentColumn 取本回合要写进用户消息 `document` 列的引用串。编码失败
// (只可能是坏数据或超限)绝不静默:打一条带来源的 ERROR,并让这条消息按「没有
// 附件」落库 —— 丢一条消息远比丢一个附件引用严重,而那条写不进去的引用也绝不
// 假装存在(与 Web 侧 persistUserMessages 的同一取舍)。
func (b *Bridge) inboundDocumentColumn(msg adapter.UnifiedMessage, attach inboundAttachments) string {
if len(attach.parts) == 0 {
return ""
}
column, err := attach.documentColumn()
if err != nil {
b.log.Error("inbound attachment refs not persisted",
"platform", msg.Platform, "channel", msg.ChannelID,
"documents", len(attach.parts), "error", err)
return ""
}
return column
}
// hasFileMedia 报告消息是否携带文件类媒体(决定「无文本消息」是否仍要跑流水线)。
func hasFileMedia(msg adapter.UnifiedMessage) bool {
for _, m := range msg.Media {
@@ -0,0 +1,109 @@
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
}
@@ -0,0 +1,407 @@
package channelbridge
// 本文件锁定 **IM 入站附件的用户消息落库**(design/24 §2.4 / §6.1)—— 安卓侧报的
// P0:字节取回了、落盘了、抽取了、模型也读到了,但**那条用户消息的 `document` 列
// 是空的**,于是所有客户端(安卓 App、Web 面板)看那条消息都是一片空白
// (`content:""`、`documents` 键缺席)。缺的就是「摄取结果 → 用户消息落库」之间的
// 那一步。
//
// 四条钉住的事实:
// - 带附件的回合:落库的用户消息 `document` 列 = 与 Web 上传**同一个结构体**的
// 引用数组(形状逐字一致),身份取自服务端存储名;
// - 本回合注入侧的结论(status/chars/truncated/reason)在回合结束后补进同一列;
// - 无附件的回合:该列为空串 ⇒ 读侧省略 `documents` 键(不是空数组);
// - 附件读不到:摄取期被拒 ⇒ 没有引用(绝不编造一条指向不存在文件的芯片),
// 诚实提示照旧只进模型可见视图;落库后抽取失败 ⇒ 引用上带 status=unreadable
// 与原因。
import (
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"path/filepath"
"strings"
"sync"
"testing"
"unicode/utf8"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
)
// documentHistory 是记录型历史存储(在 captureHistory 之上),补齐真存储才有、
// 而本用例必须依赖的两件事:
// - **给每条消息分配 id**:真存储总会分配,而「回合后补结论」正是按 id 定位的;
// - **UpdateMessageDocument**:`document` 列的回写面(design/24 §6.1)。
//
// 用替身而不是真 SQLite,是因为本用例只关心桥接器往那一列写了什么,不关心 SQL;
// 那一列本身的读写语义另有 internal/runtime/history 的用例覆盖。
type documentHistory struct {
*captureHistory
mu sync.Mutex
seq int
updates map[string]string
}
func (h *documentHistory) AppendFull(ctx context.Context, msg history.Message) (history.Message, error) {
if msg.ID == "" {
h.mu.Lock()
h.seq++
msg.ID = fmt.Sprintf("msg-%d", h.seq)
h.mu.Unlock()
}
return h.captureHistory.AppendFull(ctx, msg)
}
func (h *documentHistory) UpdateMessageDocument(_ context.Context, id, document string) error {
h.mu.Lock()
defer h.mu.Unlock()
if h.updates == nil {
h.updates = make(map[string]string)
}
h.updates[id] = document
return nil
}
// columnOf 返回 id 所指消息**当前**的 `document` 列:被回写过就是回写值(真存储
// 就是同一列被更新,这里是它的等价物),否则是落库那一刻写下的值。
func (h *documentHistory) columnOf(id string) string {
h.mu.Lock()
updated, ok := h.updates[id]
h.mu.Unlock()
if ok {
return updated
}
for _, m := range h.records() {
if m.ID == id {
return m.Document
}
}
return ""
}
// userRecord 返回最先落库的那条用户消息(一个回合只有一条用户消息)。
func (h *documentHistory) userRecord(t *testing.T) history.Message {
t.Helper()
for _, m := range h.records() {
if m.Role == history.RoleUser {
return m
}
}
t.Fatal("本回合没有落库任何用户消息")
return history.Message{}
}
// updateCount 返回回写发生的次数(0 = 这一列从未被补过结论)。
func (h *documentHistory) updateCount() int {
h.mu.Lock()
defer h.mu.Unlock()
return len(h.updates)
}
// loopStatusChatter 是桩 Chatter:它扮演**装配期解析的出口** —— 从用户消息里读出
// 桥接器挂上去的文档部件,按服务端存储名把本回合注入侧的结论(status/chars/reason)
// 回给桥接器,形状与真循环 `Result.Documents` 逐字一致(loop/document.go 的
// recordDocuments)。status 本身的算法由 loop 包的用例覆盖,这里只验证「结论回到
// 了写库的那一列」这一步。
type loopStatusChatter struct {
status string
chars int
reason string
// messages 记下本次传给循环的消息(模型可见视图),供「诚实提示只进可见视图、
// 绝不进持久化正文」这类断言使用。
messages []model.Message
res *loop.Result
}
func (c *loopStatusChatter) Run(_ context.Context, _ string, msgs []model.Message) (*loop.Result, error) {
c.messages = msgs
c.res = &loop.Result{Text: "看完了"}
for _, m := range msgs {
if m.Role != model.RoleUser {
continue
}
for _, part := range m.Content {
if part.Type != model.ContentTypeDocument {
continue
}
c.res.Documents = append(c.res.Documents, history.DocumentRef{
Name: part.DocumentName, MIME: part.DocumentMIME, Size: part.DocumentSize,
Chars: c.chars, Status: c.status, Reason: c.reason,
})
}
}
return c.res, nil
}
// newPersistEnv 装配最小入站流水线:真 `process` + 给定 Chatter + 记录型 history/
// adapter + 真文档存储(真落盘)。它是本文件所有用例的夹具。
func newPersistEnv(t *testing.T, chat Chatter, cfg documents.StoreConfig) (*Bridge, *documentHistory, string) {
t.Helper()
if cfg.UploadsDir == "" {
cfg.UploadsDir = filepath.Join(t.TempDir(), "uploads", "documents")
}
hist := &documentHistory{captureHistory: &captureHistory{}}
b, err := New(Options{
Loop: chat,
Sessions: &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}},
Owner: testOwnerProfile("telegram", "100"),
History: hist,
DocumentStore: cfg,
})
if err != nil {
t.Fatalf("New: %v", err)
}
b.Add(&fakeAdapter{name: "telegram"})
return b, hist, cfg.UploadsDir
}
// decodeColumn 解码落库的 `document` 列(坏列直接判失败:本用例写下的必须是合法
// 引用数组)。
func decodeColumn(t *testing.T, column string) []history.DocumentRef {
t.Helper()
refs, err := history.DecodeDocuments(column)
if err != nil {
t.Fatalf("解码 document 列 %q: %v", column, err)
}
return refs
}
// assertWebDocumentShape 钉住「元素形状与 Web 侧逐字一致」:`documents` 数组的元素
// 就是 history.DocumentRef 的 JSON 形态(契约 v0.59 的 name/mime/size/chars/
// truncated/status/reason)。这里比对**完整 JSON 串**,因此键名、键序与省略规则
// (omitempty)一起被钉住 —— 两侧客户端不可能读到两种形状。
func assertWebDocumentShape(t *testing.T, ref history.DocumentRef, want string) {
t.Helper()
raw, err := json.Marshal(ref)
if err != nil {
t.Fatalf("marshal document ref: %v", err)
}
if string(raw) != want {
t.Fatalf("documents 元素 = %s, want %s", raw, want)
}
}
// TestInboundAttachmentRefsLandOnPersistedUserMessage 是这条 P0 的主验收用例:一条
// 带附件的主人私聊消息走完真流水线后,**落库的用户消息**必须带着引用数组 —— 与 Web
// 上传同字段、同形状;本回合的抽取结论在回合结束后补进同一列。
func TestInboundAttachmentRefsLandOnPersistedUserMessage(t *testing.T) {
t.Parallel()
// 与主人报的那条一致:10 字节的 .txt。
body := []byte("你好,昔涟\n")
chat := &loopStatusChatter{status: history.DocumentStatusExtracted, chars: 5}
b, hist, uploads := newPersistEnv(t, chat, documents.StoreConfig{})
b.process(context.Background(), fileMessage("看看这个", "1.txt", body, "text/plain"))
names := docDirNames(t, uploads)
if len(names) != 1 {
t.Fatalf("落盘文件数 = %v, want 1", names)
}
name := names[0]
user := hist.userRecord(t)
if user.Content != "看看这个" {
t.Fatalf("用户消息正文 = %q, want 用户文本(附件引用是另一列,绝不拼进正文)", user.Content)
}
if user.Document == "" {
t.Fatal("用户消息的 document 列为空 —— 客户端看到的就是一片空白(本用例要钉住的 P0)")
}
refs := decodeColumn(t, hist.columnOf(user.ID))
if len(refs) != 1 {
t.Fatalf("引用条数 = %d, want 1: %+v", len(refs), refs)
}
got := refs[0]
// 身份一律取自服务端存储:存储名就是客户端去读的那一个(内容寻址),
// MIME 是嗅探所得,大小是落盘字节数 —— 渠道给的原始文件名与声明类型都不在其中。
if got.Name != name || got.MIME != "text/plain" || got.Size != int64(len(body)) {
t.Fatalf("引用身份 = %+v, want {%s text/plain %d}", got, name, len(body))
}
if got.Status != history.DocumentStatusExtracted || got.Chars != 5 || got.Truncated {
t.Fatalf("引用结论 = %+v, want extracted/5/未截断(回合结束后回写)", got)
}
assertWebDocumentShape(t, got,
fmt.Sprintf(`{"name":"%s","mime":"text/plain","size":%d,"chars":5,"status":"extracted"}`, name, len(body)))
}
// TestInboundUnreadableAttachmentKeepsStatusAndReason 钉住「读不到」的两个出口同时
// 成立:落库后抽取失败 ⇒ 引用上带 status=unreadable **且**原因文案还在(契约
// v0.63:reason 与 status 同批下发,客户端不必自己拼一句服务端没说过的话)。
func TestInboundUnreadableAttachmentKeepsStatusAndReason(t *testing.T) {
t.Parallel()
const reason = "无法抽取正文:读取或解析失败"
body := []byte("你好,昔涟\n")
chat := &loopStatusChatter{status: history.DocumentStatusUnreadable, reason: reason}
b, hist, uploads := newPersistEnv(t, chat, documents.StoreConfig{})
b.process(context.Background(), fileMessage("看看这个", "1.txt", body, "text/plain"))
name := docDirNames(t, uploads)[0]
refs := decodeColumn(t, hist.columnOf(hist.userRecord(t).ID))
if len(refs) != 1 {
t.Fatalf("引用条数 = %d, want 1: %+v", len(refs), refs)
}
if refs[0].Status != history.DocumentStatusUnreadable || refs[0].Reason != reason {
t.Fatalf("引用结论 = %+v, want unreadable + 原因 %q", refs[0], reason)
}
if refs[0].Chars != 0 || refs[0].Truncated {
t.Fatalf("读不到却带上了注入统计 = %+v(凭空给它一个数字就是编造)", refs[0])
}
assertWebDocumentShape(t, refs[0],
fmt.Sprintf(`{"name":"%s","mime":"text/plain","size":%d,"status":"unreadable","reason":"%s"}`, name, len(body), reason))
}
// TestInboundRejectedAttachmentKeepsNoticeWithoutRefs 钉住「读不到」的另一条出口:
// 附件在**摄取期**就被共享存储拒掉(不在可读白名单 / 非纯文本 / 超限)⇒ 没有存储名
// 就没有引用,`document` 列必须为空 —— 绝不编造一条指向不存在文件的芯片。诚实提示
// 落在**模型可见视图**里(design/24 §4.3:人类可见的来源标记归客户端渲染层,绝不
// 拼进持久化正文,见本文件对 user.Content 的断言)。
func TestInboundRejectedAttachmentKeepsNoticeWithoutRefs(t *testing.T) {
t.Parallel()
chat := &loopStatusChatter{}
b, hist, uploads := newPersistEnv(t, chat, documents.StoreConfig{})
b.process(context.Background(), fileMessage("跑一下这个", "tool.py", []byte("print('x')\n"), "text/x-python"))
if names := docDirNames(t, uploads); len(names) != 0 {
t.Fatalf("被拒的附件落盘了:%v", names)
}
user := hist.userRecord(t)
if user.Document != "" {
t.Fatalf("被拒附件的 document 列 = %q, want 空串", user.Document)
}
if user.Content != "跑一下这个" {
t.Fatalf("持久化正文 = %q, want 只有用户文本(提示行绝不拼进正文)", user.Content)
}
var visible strings.Builder
for _, m := range chat.messages {
for _, part := range m.Content {
visible.WriteString(part.Text)
visible.WriteString("\n")
}
}
if !strings.Contains(visible.String(), "tool.py") || !strings.Contains(visible.String(), "无法读取") {
t.Fatalf("模型可见视图里没有诚实的拒绝原因:%q", visible.String())
}
}
// TestInboundMessageWithoutAttachmentsOmitsDocumentColumn 钉住边界:无附件的消息
// 落库时那一列是空串(不是空数组)—— 读侧据此**省略** `documents` 键,与引入附件
// 之前逐字相同(网关侧 history_document_test.go 钉住省略本身)。
func TestInboundMessageWithoutAttachmentsOmitsDocumentColumn(t *testing.T) {
t.Parallel()
chat := &loopStatusChatter{}
b, hist, _ := newPersistEnv(t, chat, documents.StoreConfig{})
b.process(context.Background(), directMessage())
user := hist.userRecord(t)
if user.Content != "hello" {
t.Fatalf("用户消息正文 = %q, want hello", user.Content)
}
if user.Document != "" {
t.Fatalf("无附件消息的 document 列 = %q, want 空串", user.Document)
}
if n := hist.updateCount(); n != 0 {
t.Fatalf("无附件却发生了 %d 次 document 回写", n)
}
}
// TestInboundGroupAttachmentRefsLandOnIsolatedUserMessage 钉住受限(群聊/陌生人)
// 会话走的是另一个落库钩子(persistIsolated):第三方消息的附件同样要把引用写进
// 它自己那条日志的用户消息上,且**绝不**镜像进主人主会话。
func TestInboundGroupAttachmentRefsLandOnIsolatedUserMessage(t *testing.T) {
t.Parallel()
uploads := filepath.Join(t.TempDir(), "uploads", "documents")
chat := &loopStatusChatter{status: history.DocumentStatusListed}
hist := &documentHistory{captureHistory: &captureHistory{}}
b, err := New(Options{
Loop: chat,
Sessions: &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false}},
Owner: &user.OwnerProfile{},
History: hist,
DocumentStore: documents.StoreConfig{UploadsDir: uploads},
})
if err != nil {
t.Fatalf("New: %v", err)
}
msg := adapter.UnifiedMessage{
Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup,
GroupID: "-20", GroupName: "Gang", SenderID: "9", SenderName: "张三",
Sender: adapter.Actor{Role: adapter.RoleMember}, Mentions: []adapter.Mention{{UserID: "9"}},
Content: "看下这个", ContentType: adapter.ContentText,
Media: []adapter.Media{{
Kind: adapter.MediaFile, Name: "third.txt", Size: 4, MIMEType: "text/plain", Data: []byte("hi\n"),
}},
}
b.process(context.Background(), msg)
name := docDirNames(t, uploads)[0]
user := hist.userRecord(t)
if user.SessionID != "sess-g" {
t.Fatalf("用户消息落库会话 = %q, want 隔离会话 sess-g", user.SessionID)
}
refs := decodeColumn(t, hist.columnOf(user.ID))
if len(refs) != 1 || refs[0].Name != name {
t.Fatalf("隔离会话的引用 = %+v, want 1 条指向 %s", refs, name)
}
if refs[0].Status != history.DocumentStatusListed {
t.Fatalf("隔离会话的引用结论 = %+v, want listed(回合后回写)", refs[0])
}
}
// TestInboundAttachmentConclusionComesFromRealLoop 是这条链路的端到端验证:真存储 +
// 真抽取器 + **真循环** + 假模型(夹具沿用 inbound_documents_test.go 的 stubModelAdapter)。
// 它钉住「一侧的事实」:模型可见视图里出现清单行与正文,与回给客户端同一列的结论
// (extracted + 注入字符数)必须来自同一次装配 —— 否则「她读到了」与「客户端显示
// 她读到了」就会分叉。
func TestInboundAttachmentConclusionComesFromRealLoop(t *testing.T) {
t.Parallel()
dataDir := t.TempDir()
uploads := filepath.Join(dataDir, "uploads", "documents")
modelAdapter := &stubModelAdapter{}
ag, err := agent.New("测试系统提示词", nil, nil, agent.WithModel(modelAdapter, nil))
if err != nil {
t.Fatalf("构造 agent: %v", err)
}
discard := slog.New(slog.NewTextHandler(io.Discard, nil))
runner := loop.New(loop.Options{
Agent: ag,
DocumentResolver: documents.NewResolver(documents.Config{DataDir: dataDir, UploadsDir: uploads, Log: discard}),
MaxRounds: 4,
Log: discard,
})
body := []byte("你好,昔涟\n")
b, hist, _ := newPersistEnv(t, runner, documents.StoreConfig{DataDir: dataDir, UploadsDir: uploads})
b.process(context.Background(), fileMessage("看看这个", "1.txt", body, "text/plain"))
name := docDirNames(t, uploads)[0]
refs := decodeColumn(t, hist.columnOf(hist.userRecord(t).ID))
if len(refs) != 1 {
t.Fatalf("引用条数 = %d, want 1: %+v", len(refs), refs)
}
got := refs[0]
if got.Name != name || got.MIME != "text/plain" || got.Size != int64(len(body)) {
t.Fatalf("引用身份 = %+v, want {%s text/plain %d}", got, name, len(body))
}
if got.Status != history.DocumentStatusExtracted {
t.Fatalf("引用结论 = %+v, want extracted(真循环的装配期结论)", got)
}
if want := utf8.RuneCountInString(string(body)); got.Chars != want {
t.Fatalf("注入字符数 = %d, want %d(chars 是**本回合注入侧**的字符数)", got.Chars, want)
}
visible := modelAdapter.visibleText()
if !strings.Contains(visible, "[附件 "+name+" · ") || !strings.Contains(visible, string(body)) {
t.Fatalf("模型可见视图里没有该附件的清单行或正文:%q", visible)
}
}
+3 -1
View File
@@ -236,7 +236,9 @@ func (b *Bridge) mentionsTriggerWord(text string) bool {
// 收到该消息的账号适配器,日志因此带上适配器配置标签。
// 它是最小打扰路径中「登记」的那一侧。
func (b *Bridge) registerSilenced(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor, sess session.Session, a adapter.PlatformAdapter) {
b.persistIsolated(ctx, history.RoleUser, msg.Content, deviceIDFor(msg), sess, msg.Platform, senderDisplayName(msg, actor), adapterName(a))
// 被静默的消息不摄取附件(摄取在闸门之后),因此这一列恒为空 ——
// 绝不写一条指向不存在文件的引用。
b.persistIsolated(ctx, history.RoleUser, msg.Content, "", deviceIDFor(msg), sess, msg.Platform, senderDisplayName(msg, actor), adapterName(a))
b.log.Debug("non-owner message silenced, registered only",
"platform", msg.Platform, "channel", msg.ChannelID, "session", sess.SessionID)
}
+25 -12
View File
@@ -60,20 +60,28 @@ func (b *Bridge) notifyPersisted(msg history.Message) {
// Web 来源的消息该字段为空。mood 是该轮角色的心情
// (design/01 §2 情绪状态机),只挂在助手回复上(design/16 §3.3:
// 助手消息承载 mode/mood);用户消息与未跟踪的心情传 ""。
// 未装配历史存储时为空操作,因此桥接器在没有
// 存储的情况下行为不变。
func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, platform, deviceID, adapterID string) {
// document 是这条消息携带的文档引用数组(design/24 §2.4、§6.1),
// 形状即 history.EncodeDocuments 的产物;没有附件时传 ""(那一列因此为
// 空,读侧 `documents` 键随之省略)。IM 入站的附件与 Web 上传共用这一列,
// 客户端两侧读到的附件芯片才是同一个东西。
//
// 它返回刚落库的完整消息(未装配历史存储、或写库失败时是零值),
// 使调用方能在回合结束后按 id 把文档抽取结论补进同一列
// (inbound_documents_persist.go 的两段式)。未装配历史存储时为空操作,
// 因此桥接器在没有存储的情况下行为不变。
func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, document, platform, deviceID, adapterID string) history.Message {
if b.history == nil {
return
return history.Message{}
}
full, err := b.history.AppendFull(ctx, history.Message{
SessionID: history.MainSessionID, Role: role, Content: content, Mood: mood, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, Mode: modeChat,
SessionID: history.MainSessionID, Role: role, Content: content, Mood: mood, Document: document, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, Mode: modeChat,
})
if err != nil {
b.log.Warn("persist main session", "err", err, "role", role, "platform", platform, "device_id", deviceID, "adapter", adapterID)
return
return history.Message{}
}
b.notifyPersisted(full)
return full
}
// persistIsolated 把一条消息写入受限(群聊/陌生人)会话
@@ -84,20 +92,24 @@ func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, platfor
// 该消息仍不会镜像进 Web 主会话(只有主人私聊
// 消息才镜像到 Web);它只交给传输钩子,让
// gateway 广播 `channel_persisted` 事件、IM 适配器日志页无需刷新
// 即可追加。未装配历史存储时为空操作,因此桥接器在
// 没有存储的情况下行为不变。
func (b *Bridge) persistIsolated(ctx context.Context, role, content, deviceID string, sess session.Session, platform, senderName, adapterID string) {
// 即可追加。document 是该消息携带的文档引用数组(design/24 §2.4),
// 与 persistToMain 同一列、同一形状;没有附件时传 ""。
//
// 它返回刚落库的完整消息(未装配历史存储、或写库失败时是零值),
// 供调用方在回合结束后补回文档抽取结论。未装配历史存储时为空操作,
// 因此桥接器在没有存储的情况下行为不变。
func (b *Bridge) persistIsolated(ctx context.Context, role, content, document, deviceID string, sess session.Session, platform, senderName, adapterID string) history.Message {
if b.history == nil {
return
return history.Message{}
}
full, err := b.history.AppendFull(ctx, history.Message{
SessionID: sess.SessionID, Role: role, Content: content, Platform: platform,
SessionID: sess.SessionID, Role: role, Content: content, Document: document, Platform: platform,
DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, Mode: modeChat,
})
if err != nil {
b.log.Warn("persist isolated session", "err", err, "role", role,
"session", sess.SessionID, "platform", platform, "sender_name", senderName, "adapter", adapterID)
return
return history.Message{}
}
// 把隔离写入作为 `channel_persisted` 事件广播,让 IM 适配器
// 日志页实时追加。gateway 按会话区分事件类型:
@@ -105,4 +117,5 @@ func (b *Bridge) persistIsolated(ctx context.Context, role, content, deviceID st
// 消息本身仍只落在隔离的 sess-xxx 会话(绝不在
// 主/`web` 会话,design/16 §4.6 §3)。
b.notifyPersisted(full)
return full
}
+1 -1
View File
@@ -73,7 +73,7 @@ func (b *Bridge) SendProactive(ctx context.Context, msg adapter.UnifiedMessage)
// §4.4 + §4.6.1)。否则模型看不到自己主动说了什么,
// 主人后续回复就会显得不连贯(该消息「丢失」了历史)。
// 它与主人私聊 IM 回复走同一条镜像路径,作用域为 MainSessionID。
b.persistToMain(ctx, history.RoleAssistant, msg.Content, "", target.platform,
b.persistToMain(ctx, history.RoleAssistant, msg.Content, "", "", target.platform,
fmt.Sprintf("%s:%s:%s", target.platform, target.accountID, target.accountID),
target.platform)
return nil