Cyrene_v2/internal/services/channelbridge/inbound_documents_persist_test.go

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)
	}
}