diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index 19f7a46..fa9432c 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -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) { diff --git a/internal/services/channelbridge/inbound_documents.go b/internal/services/channelbridge/inbound_documents.go index 083d5b3..f863c2e 100644 --- a/internal/services/channelbridge/inbound_documents.go +++ b/internal/services/channelbridge/inbound_documents.go @@ -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 { diff --git a/internal/services/channelbridge/inbound_documents_persist.go b/internal/services/channelbridge/inbound_documents_persist.go new file mode 100644 index 0000000..542bd3f --- /dev/null +++ b/internal/services/channelbridge/inbound_documents_persist.go @@ -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 +} diff --git a/internal/services/channelbridge/inbound_documents_persist_test.go b/internal/services/channelbridge/inbound_documents_persist_test.go new file mode 100644 index 0000000..c88c14c --- /dev/null +++ b/internal/services/channelbridge/inbound_documents_persist_test.go @@ -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) + } +} diff --git a/internal/services/channelbridge/light.go b/internal/services/channelbridge/light.go index 0e8cabd..ed49d53 100644 --- a/internal/services/channelbridge/light.go +++ b/internal/services/channelbridge/light.go @@ -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) } diff --git a/internal/services/channelbridge/persist.go b/internal/services/channelbridge/persist.go index 302c27b..a5d9064 100644 --- a/internal/services/channelbridge/persist.go +++ b/internal/services/channelbridge/persist.go @@ -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 } diff --git a/internal/services/channelbridge/proactive.go b/internal/services/channelbridge/proactive.go index c3c9ee4..c352eba 100644 --- a/internal/services/channelbridge/proactive.go +++ b/internal/services/channelbridge/proactive.go @@ -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