From 2a43035f5d0636b95bb6a983b6db9b7afb065d86 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 19:25:00 +0800 Subject: [PATCH] =?UTF-8?q?refactor+feat(documents,channelbridge):=20?= =?UTF-8?q?=E5=85=B1=E4=BA=AB=E5=AD=98=E5=82=A8=E6=A0=B8=E5=BF=83=20+=20IM?= =?UTF-8?q?=20=E5=85=A5=E7=AB=99=E9=99=84=E4=BB=B6=E6=91=84=E5=8F=96?= =?UTF-8?q?=EF=BC=88=E4=B8=BB=E4=BA=BA=202.3=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 抽共享存储核心：internal/runtime/documents/store.go（22 项白名单 + 字节嗅探唯一裁决 + 内容寻址 sha256[:32]+ext + 去重 + 限额 + 路径双检）+ store_meta.go（documents 表唯一写入实现）。 - gateway 改为调用 store，**对外行为零变化**（机器证明：既有 documents_test/chat_documents_test/chat_documents_bounds_test 三个文件 0 行改动且全绿；状态码/响应字段/错误码/审计动作名/限额文档口径全部逐字保留；maxChatDocuments 改为引用 documents.MaxPerMessage）。 - IM 入站附件摄取（channelbridge）：文件类媒体 → store.Save → 挂 model.DocumentMessage(RoleUser) 部件 ⇒ 后续 resolveDocuments 自动抽取/注入/清单；挂载点在「会话解析 + 黑白名单 + 最小打扰」三重闸门之后（被拦消息绝不在盘上留文件）；**不支持的格式不静默**（诚实提示行 + 带来源日志，只记 name/mime/size，名字做展示净化防伪造行）；图片/语音保持既有链路。 - 主会话独立验收：隔离 worktree（M1 的 12 文件）build/vet 绿、全树 59 包 0 FAIL、双平台交叉编译绿、12 条 Inbound 用例 PASS；变异独立复现（关闭 IM 摄取 ⇒ 5 条红，含「落盘文件数 = [] want 1」）。 - ⚠️ 真机边界（需裁决）：现网四个适配器上报的文件媒体**只带 URL**（telegram FileID / qq-obv11 file|url / weixin·kook CDN URL），Media.Data 恒空 ⇒ 本切片只摄**内联字节**，URL 一律「诚实提示，不代取远程」（避免第三份 SSRF 实现）。真机要让「她读到 IM 文件」，需各适配器入站时下载并内联（凭据在各自渠道里，推荐）或新增共享受守卫抓取器。 - 另登记：IM 附件暂未写 chat_messages.document 列（二手回合看不到引用，建议并入 S7）；channelbridge 既有测试夹具在 -race 下有数据竞争（stubSessions.GetOrCreate，与本次无关）。 --- cmd/cyrene/wire_channels.go | 17 + cmd/cyrene/wire_channels_documents_test.go | 47 +++ internal/runtime/documents/store.go | 349 ++++++++++++++++ internal/runtime/documents/store_meta.go | 62 +++ internal/runtime/documents/store_test.go | 310 ++++++++++++++ .../services/channelbridge/channelbridge.go | 11 + internal/services/channelbridge/inbound.go | 54 ++- .../channelbridge/inbound_documents.go | 290 +++++++++++++ .../inbound_documents_honesty_test.go | 122 ++++++ .../channelbridge/inbound_documents_test.go | 395 ++++++++++++++++++ internal/services/gateway/chat_documents.go | 5 +- internal/services/gateway/documents.go | 239 ++++------- 12 files changed, 1743 insertions(+), 158 deletions(-) create mode 100644 cmd/cyrene/wire_channels_documents_test.go create mode 100644 internal/runtime/documents/store.go create mode 100644 internal/runtime/documents/store_meta.go create mode 100644 internal/runtime/documents/store_test.go create mode 100644 internal/services/channelbridge/inbound_documents.go create mode 100644 internal/services/channelbridge/inbound_documents_honesty_test.go create mode 100644 internal/services/channelbridge/inbound_documents_test.go diff --git a/cmd/cyrene/wire_channels.go b/cmd/cyrene/wire_channels.go index 322d454..0cb56be 100644 --- a/cmd/cyrene/wire_channels.go +++ b/cmd/cyrene/wire_channels.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "path/filepath" "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" @@ -11,6 +12,7 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/config" "git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge" "git.yeij.top/AskaEth/Cyrene_v2/internal/user" @@ -77,6 +79,9 @@ func (a *app) wireChannels() error { } } + // 入站附件的目录口径与文档摄取层（wire_runtime.go 的 docResolver）同源， + // 两处必须指向同一棵目录树，否则附件会落在一个地方、抽取去另一个地方找。 + docDataDir := resolveDataDir(a.cfg) bridge, berr := channelbridge.New(channelbridge.Options{ Log: logger.WithSource(a.log, "services/channelbridge"), Loop: a.loop, @@ -85,6 +90,18 @@ func (a *app) wireChannels() error { History: a.history, QuietStartHour: a.cfg.Background.QuietStartHour, QuietEndHour: a.cfg.Background.QuietEndHour, + // 入站附件摄取（design/24 §3；切片 M1）：IM 收到的受支持格式文件落进与 + // Web 上传**同一个**存储核心（目录口径与 wire_runtime.go 的文档摄取层 + // 同源：`/uploads/documents`），元数据也写进同一张 `documents` + // 表——`documents` 表的写入路径因此全仓只有一条（S7 的引用计数与 GC 有 + // 单一事实来源）。未配置数据目录时写入面为 nil，桥接器走「未接线」的 + // 诚实路径（模型知道有附件读不到），绝不假装读到了。 + DocumentStore: documents.StoreConfig{ + DataDir: docDataDir, + UploadsDir: filepath.Join(docDataDir, "uploads", "documents"), + Meta: documents.MetadataWriterFor(a.dbPath, a.log), + Log: a.log, + }, // 可选的复审模型在接近单适配器上限时剔除低价值排队消息 // （design/12 §6 策略四）。适配器为 nil 时，闸门回落到 FIFO 驱逐。 Review: reviewFor(a.modelAdapter, logger.WithSource(a.log, "services/channelbridge/review")), diff --git a/cmd/cyrene/wire_channels_documents_test.go b/cmd/cyrene/wire_channels_documents_test.go new file mode 100644 index 0000000..e4d2a4b --- /dev/null +++ b/cmd/cyrene/wire_channels_documents_test.go @@ -0,0 +1,47 @@ +package main + +import ( + "os" + "strings" + "testing" +) + +// TestWireChannels_WiresInboundDocumentStore 是一道「接线存在性」防回归检查 +// （切片 M1；与 `TestWireRuntime_UsesTheCompositionRootSeams` 同一手法）。 +// +// 组合根本身无法在单元测试里实例化（开库、迁移、绑端口，而后端还在运行）， +// 因此「IM 入站附件确实被接上了共享文档存储」只能在源码层钉住：列出的字符串 +// 少任何一个，桥接器都会静默退回「未接线」的诚实路径 —— 附件不再落库、S7 的 GC +// 也看不到 IM 来源的文件，而这一切**不会**让任何既有用例变红。所以必须显式钉住。 +// +// 最后一对断言是**成对契约**：入站附件落盘的目录与文档摄取层读原文件的目录 +// 必须逐字同源（`/uploads/documents`），否则附件会落在一个地方、 +// 装配期去另一个地方找 —— 症状是「附件全都读不到」，且两边各自都自洽。 +func TestWireChannels_WiresInboundDocumentStore(t *testing.T) { + t.Parallel() + + const dirSeam = `UploadsDir: filepath.Join(docDataDir, "uploads", "documents")` + src, err := os.ReadFile("wire_channels.go") + if err != nil { + t.Fatalf("read wire_channels.go: %v", err) + } + for _, want := range []string{ + "DocumentStore: documents.StoreConfig{", + dirSeam, + "Meta: documents.MetadataWriterFor(a.dbPath, a.log)", + "docDataDir := resolveDataDir(a.cfg)", + } { + if !strings.Contains(string(src), want) { + t.Fatalf("wire_channels.go 里找不到 %q —— IM 入站附件没有被接上共享文档存储", want) + } + } + + reader, err := os.ReadFile("wire_runtime.go") + if err != nil { + t.Fatalf("read wire_runtime.go: %v", err) + } + if !strings.Contains(string(reader), dirSeam) { + t.Fatalf("文档摄取层的目录与 IM 入站附件的目录不同源（wire_runtime.go 里找不到 %q）—— "+ + "附件会落在一个地方、装配期去另一个地方找", dirSeam) + } +} diff --git a/internal/runtime/documents/store.go b/internal/runtime/documents/store.go new file mode 100644 index 0000000..a7c3ffe --- /dev/null +++ b/internal/runtime/documents/store.go @@ -0,0 +1,349 @@ +package documents + +// 本文件归属**文档存储核心**（design/24 §3.2；切片 M1）：把 S1 原本长在网关 +// `internal/services/gateway/documents.go` 里的存储规则（字节嗅探唯一裁决 + +// 扩展名白名单 + 内容寻址 + 去重 + 限额 + 元数据）抽成本包的**一份**实现， +// 供两条来源路径共用： +// +// - Web 上传（`POST /api/v1/documents`，网关调用）； +// - IM 入站附件（channelbridge 把消息里的文件类媒体交给这里，切片 M1）。 +// +// 为什么必须先抽出来：规则里有安全裁决（嗅探否决、白名单、路径双检），在桥接器 +// 里再写一套必然漂移；而 `/uploads/documents` 与 `documents` 元数据表 +// 也必须是**唯一写入面**，S7 的引用计数与 GC 才有单一事实来源。 +// +// 落盘规则与 S1 逐字同源：客户端原始文件名**永不参与路径**，落盘名是 +// `sha256(bytes)[:32] + ext`，同字节同扩展名必然同名、天然去重；目录 0700、 +// 文件 0600（继承临时文件）。读侧（`GET /api/v1/documents/{name}`）与装配期 +// 摄取（Resolver）不改，仍旧只认这个形态的名字。 + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "log/slog" + "net/http" + "os" + "path/filepath" + "strings" +) + +const ( + // DefaultMaxBytes 是文档单文件字节的**默认上限**（20 MiB，即 + // `CHAT_DOCUMENT_MAX_BYTES` 的默认值）。它只在这里定义一次：网关的 + // `maxDocumentBytes`、配置的 default 标签与 IM 入站附件都从它派生，绝不 + // 在别处再写一个数（design/24 §3.3：文档上限与图片上限刻意解耦）。 + DefaultMaxBytes = 20 << 20 + + // MaxPerMessage 是**单条消息**（一个回合）可携带的附件数上限 + // （design/24 §9.4-3 的定稿值 5）。网关的单回合引用上限与 IM 入站附件 + // 的上限同源于它。 + MaxPerMessage = 5 + + // HashHexLen 是内容寻址落盘名使用的小写十六进制主名长度（32 位十六进制 + // = 128 位）。读侧的名字形态校验与 sidecar 的键同名同宽。 + HashHexLen = 32 + + // SniffOKMIME 是唯一放行的嗅探结果**基类型**。白名单里每一种文档（文本、 + // 代码、CSV、JSON、YAML…）经 `http.DetectContentType` 嗅探后都落在 + // `text/plain`；HTML/XML/SVG/压缩包/可执行文件/判不定者各有自己的签名， + // 因此「不是 text/plain」即「拒绝」。 + SniffOKMIME = "text/plain" + + // tempPattern 是原子落盘用的临时文件前缀（点前缀，绝不与内容寻址名混淆）。 + tempPattern = ".document-*" + + // dirMode 是文档目录权限；文件权限继承 `os.CreateTemp` 的 0600。 + dirMode = 0o700 + + // uploadsSubDir / documentSubDir 是 `` 之下的默认布局： + // `/uploads/documents`。与网关的 `documentSubDir`、组合根给摄取层 + // 的 UploadsDir 同源，改一处必须同步（否则装配期会去读一个不存在的目录）。 + uploadsSubDir = "uploads" + documentSubDir = "documents" +) + +// 存储侧的哨兵错误。调用方一律用 `errors.Is` 分类，**绝不按错误文本匹配**： +// 网关把它们映射为既有的 415 / 413 / 400 契约错误码，IM 侧映射为诚实提示。 +var ( + // ErrExtNotAllowed 报告原始文件名的扩展名不在白名单内（deny-by-default）。 + ErrExtNotAllowed = errors.New("documents: 扩展名不在白名单内") + // ErrContentNotText 报告字节嗅探否决了这份内容（基类型不是 text/plain）。 + ErrContentNotText = errors.New("documents: 内容不是纯文本类（字节嗅探否决）") + // ErrTooLarge 报告内容超过生效的单文件字节上限。 + ErrTooLarge = errors.New("documents: 内容超过字节上限") + // ErrEmpty 报告内容为空（空文件没有落盘意义，也不该产生元数据行）。 + ErrEmpty = errors.New("documents: 内容为空") + // ErrNoDir 报告文档落盘目录未配置（组合根/网关都没给出 UploadsDir 与 DataDir）。 + ErrNoDir = errors.New("documents: 文档落盘目录未配置") + // ErrBadPath 报告拼出的落盘路径逃出了文档目录（内容寻址名由本包签发， + // 因此这是防御性的第二道门）。 + ErrBadPath = errors.New("documents: 落盘路径逃出文档目录") + // ErrMetadata 报告元数据写入失败。它与「文件已经落盘」是两个事实：调用方 + // 据它把 500 文案分开（S1 的「记录文档元数据失败」），并知道盘上那份文件 + // 需要 S7 的 GC 容忍或清理（design/24 §9.2-5）。 + ErrMetadata = errors.New("documents: 元数据写入失败") +) + +// Ref 是一次成功落盘的结果：服务端签发的引用 + 落盘路径。前五个字段是调用方 +// 对外呈现的事实（网关的响应字段、IM 侧的展示身份），Path 只给服务端自己用 +// （日志/清理），**绝不进模型上下文、绝不回给客户端**。 +type Ref struct { + // Name 是内容寻址的存储名（`sha256(bytes)[:32] + ext`）。 + Name string + // MIME 是字节嗅探所得的基类型（去参数、小写）。 + MIME string + // Size 是落盘字节数。 + Size int64 + // SHA256 是完整内容的十六进制摘要（落盘名只取前 HashHexLen 位）。 + SHA256 string + // Deduped 报告这次写入命中了盘上已有的同一份内容（同字节同扩展名）。 + Deduped bool + // Path 是落盘绝对/相对路径（按构造方给的 UploadsDir 决定形态）。 + Path string +} + +// Record 是写入 `documents` 表（迁移 0014）的一行元数据。列与 S1 逐字一致。 +type Record struct { + // Name 是内容寻址的存储名（表上的唯一键）。 + Name string + // SHA256 是完整内容的十六进制摘要。 + SHA256 string + // MIME 是嗅探所得的基类型。 + MIME string + // Size 是字节数。 + Size int64 + // Uploader 是上传者身份：网关给请求 claims 的 sub，IM 给渠道来源标签。 + // 身份**只由构造方给出**，store 绝不从文件名/内容里猜（项目红线）。 + Uploader string +} + +// MetadataWriter 是元数据写入的**可选注入面**（design/24 §9.2-5 的技术债： +// 网关当前每次上传 `db.Open` 开新连接，本切片不把它改成连接注入）。 +// 未注入时 Save 照旧成功，只跳过写表并留一条 DEBUG 痕迹。 +type MetadataWriter interface { + // WriteDocument 写入一行文档元数据。同一存储名重复上传由实现方 + // `ON CONFLICT(name) DO NOTHING`（首次的 uploader/created_at 是权威）。 + WriteDocument(ctx context.Context, rec Record) error +} + +// StoreConfig 是文档存储核心的接线参数。路径口径由调用方给出，本包不猜数据 +// 目录布局（与摄取层 Config 同一理由）。 +type StoreConfig struct { + // DataDir 是数据目录。UploadsDir 留空时文档落进 + // `/uploads/documents`（网关与组合根用的是同一布局）。 + DataDir string + // UploadsDir 是文档落盘的**完整目录**。非空时优先于 DataDir 推导。 + UploadsDir string + // MaxBytes 是单文件字节上限；≤0 时取 DefaultMaxBytes。口径是 + // 「≤0 = 用默认」而不是「禁用上限」，与网关的 documentLimitBytes 一致。 + MaxBytes int + // Uploader 是写进元数据的上传者身份（见 Record.Uploader）。 + Uploader string + // Meta 是元数据写入的可选注入面；nil 时跳过写表并留痕。 + Meta MetadataWriter + // Log 是基础 logger；nil 时替换为标记来源 runtime/documents 的丢弃 logger。 + Log *slog.Logger +} + +// Store 是文档落盘的唯一实现。它可安全并发使用（无可变状态，落盘的竞态由 +// 「临时文件 + rename」与「已存在即去重」处理）。 +type Store struct { + dir string + maxBytes int + uploader string + meta MetadataWriter + log *slog.Logger +} + +// NewStore 构造文档存储。UploadsDir 与 DataDir 都为空 ⇒ 每次 Save 都返回 +// ErrNoDir（宁可明确失败，也不猜一个目录出来）。 +func NewStore(cfg StoreConfig) *Store { + dir := strings.TrimSpace(cfg.UploadsDir) + if dir == "" { + if root := strings.TrimSpace(cfg.DataDir); root != "" { + dir = filepath.Join(root, uploadsSubDir, documentSubDir) + } + } + return &Store{ + dir: dir, + maxBytes: cfg.MaxBytes, + uploader: strings.TrimSpace(cfg.Uploader), + meta: cfg.Meta, + log: sourcedLog(cfg.Log), + } +} + +// Dir 返回生效的文档落盘目录（未配置时为空串）。它给测试与日志用，不参与落盘。 +func (s *Store) Dir() string { return s.dir } + +// Limit 返回本次生效的单文件字节上限（≤0 归一为 DefaultMaxBytes）。 +func (s *Store) Limit() int { + if s.maxBytes <= 0 { + return DefaultMaxBytes + } + return s.maxBytes +} + +// Save 把一份原始文件落进文档存储，返回服务端签发的引用。规则与 S1 逐字同源， +// 判定顺序也刻意保持（扩展名 → 空 → 上限 → 嗅探），因为「同一次上传里两条规则 +// 都不过时给哪个错误」是对外可见的行为： +// +// 1. **扩展名白名单**（deny-by-default）：原始文件名的扩展名不在放行集合内一律 +// 拒绝（`ErrExtNotAllowed`）—— 嗅探成 `text/plain` 的 `.js`/`.py` 也在此列； +// 2. **空内容**拒绝（`ErrEmpty`）； +// 3. **字节上限**拒绝（`ErrTooLarge`，上限见 Limit）； +// 4. **字节嗅探唯一裁决**：`http.DetectContentType` 的基类型必须是 `text/plain`， +// 否则拒绝（`ErrContentNotText`）—— 客户端声明的 MIME 与文件名只作参考、 +// 可被否决； +// 5. 内容寻址落盘 + 路径双检 + 去重（同字节同扩展名 ⇒ 同名 ⇒ `Deduped=true`）， +// 随后（可选）写一行 `documents` 元数据。 +// +// 元数据写入失败时返回的错误同时满足 `errors.Is(err, ErrMetadata)`，且**已落盘的 +// 引用照旧返回**：文件确实在盘上，调用方据此如实上报（S1 的行为）。 +func (s *Store) Save(ctx context.Context, origName string, data []byte) (Ref, error) { + sniffed := http.DetectContentType(data) + ref, err := s.prepare(origName, sniffed, data) + if err != nil { + return Ref{}, err + } + committed, err := s.commit(ref, data) + if err != nil { + return Ref{}, err + } + if err := s.record(ctx, committed); err != nil { + return committed, err + } + return committed, nil +} + +// prepare 执行落盘之前的全部裁决并算出内容寻址引用（不触盘、不建目录）。 +// sniffed 由调用方传入，使「嗅探发生在任何磁盘动作之前」这一顺序在代码里可见。 +func (s *Store) prepare(origName, sniffed string, data []byte) (Ref, error) { + ext := ExtOf(origName) + if !ExtAllowed(ext) { + return Ref{}, fmt.Errorf("documents: save: %w", ErrExtNotAllowed) + } + if len(data) == 0 { + return Ref{}, fmt.Errorf("documents: save: %w", ErrEmpty) + } + if limit := s.Limit(); len(data) > limit { + return Ref{}, fmt.Errorf("documents: save: 上限 %d 字节，本次 %d 字节: %w", limit, len(data), ErrTooLarge) + } + if !SniffAllowed(sniffed) { + return Ref{}, fmt.Errorf("documents: save: 嗅探 %q 非纯文本: %w", baseMIME(sniffed), ErrContentNotText) + } + sum := sha256.Sum256(data) + return Ref{ + Name: hashName(sum[:], ext), + MIME: baseMIME(sniffed), + Size: int64(len(data)), + SHA256: hex.EncodeToString(sum[:]), + }, nil +} + +// commit 把字节落进 `/`：先写点前缀临时文件再 rename（部分写入绝不会 +// 以可读的名字露面），目标名已存在时丢弃临时文件并报告 deduped —— 盘上只留一份。 +func (s *Store) commit(ref Ref, data []byte) (Ref, error) { + if s.dir == "" { + return Ref{}, fmt.Errorf("documents: commit: %w", ErrNoDir) + } + path, err := targetPath(s.dir, ref.Name) + if err != nil { + return Ref{}, err + } + if info, serr := os.Stat(path); serr == nil && !info.IsDir() { + ref.Path, ref.Deduped = path, true + return ref, nil + } + if err := os.MkdirAll(s.dir, dirMode); err != nil { + return Ref{}, fmt.Errorf("documents: 建目录: %w", err) + } + tmp, err := os.CreateTemp(s.dir, tempPattern) + if err != nil { + return Ref{}, fmt.Errorf("documents: 建临时文件: %w", err) + } + tmpName := tmp.Name() + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + _ = os.Remove(tmpName) + return Ref{}, fmt.Errorf("documents: 写临时文件: %w", err) + } + if err := tmp.Close(); err != nil { + _ = os.Remove(tmpName) + return Ref{}, fmt.Errorf("documents: 关临时文件: %w", err) + } + if err := os.Rename(tmpName, path); err != nil { + _ = os.Remove(tmpName) + return Ref{}, fmt.Errorf("documents: 提交落盘: %w", err) + } + ref.Path = path + return ref, nil +} + +// record 写入一行 `documents` 元数据（可选注入面）。未注入时跳过并留痕： +// 文件已经落盘，元数据留待接线（S1 在 `dbPath == ""` 时的行为）。 +func (s *Store) record(ctx context.Context, ref Ref) error { + if s.meta == nil { + s.log.Debug("document metadata skipped: writer not wired", + "name", ref.Name, "size", ref.Size, "deduped", ref.Deduped) + return nil + } + rec := Record{ + Name: ref.Name, SHA256: ref.SHA256, MIME: ref.MIME, + Size: ref.Size, Uploader: s.uploader, + } + if err := s.meta.WriteDocument(ctx, rec); err != nil { + return fmt.Errorf("documents: 元数据 %q: %w: %w", ref.Name, ErrMetadata, err) + } + return nil +} + +// targetPath 按落盘名拼出路径并做 S1 的**路径双检**：拼出的路径必须仍在文档 +// 目录内。名字由本包签发（内容寻址），因此这是防御性的第二道门 —— 与读侧 +// `filepath.Dir(path) != filepath.Clean(dir)` 是同一手法。 +func targetPath(dir, name string) (string, error) { + path := filepath.Join(dir, name) + if filepath.Dir(path) != filepath.Clean(dir) { + return "", fmt.Errorf("documents: 落盘名 %q: %w", name, ErrBadPath) + } + return path, nil +} + +// ExtOf 从原始文件名取出小写扩展名（含点）。它只作**参考**：结果必须同时通过 +// 扩展名白名单与字节嗅探，客户端因此无法靠改扩展名换类型。 +func ExtOf(filename string) string { + return strings.ToLower(filepath.Ext(filename)) +} + +// ExtAllowed 报告扩展名是否在放行集合内。deny-by-default：集合之外的一切 +// （`.html`/`.svg`/`.xml`/`.zip`/`.docx`/`.xlsx`/`.exe`/`.dll`/`.bat`/`.cmd`/ +// `.ps1`/`.js`/`.mjs`/`.sh`/`.py`…）一律拒绝。注意 `.js`/`.py` 即便嗅探成 +// `text/plain` 也必须拒绝 —— 判定以扩展名与嗅探的**合取**为准。 +// +// 这 22 项是 design/24 §9.2-2 的定稿集合，全仓只此一份（网关的名字形态校验 +// 与 IM 入站附件都经它判定）。 +func ExtAllowed(ext string) bool { + switch ext { + case ".txt", ".md", ".markdown", ".csv", ".tsv", ".json", ".jsonl", + ".yaml", ".yml", ".log", ".ini", ".cfg", ".toml", + ".go", ".ts", ".tsx", ".java", ".c", ".h", ".cpp", ".rs", ".sql": + return true + default: + return false + } +} + +// SniffAllowed 报告嗅探所得类型是否落在放行的文本类内。它是「类型裁决的唯一 +// 裁决者」在代码里的落点：客户端声明的 `Content-Type` 从不参与决定。 +func SniffAllowed(sniffed string) bool { + return baseMIME(sniffed) == SniffOKMIME +} + +// hashName 返回内容寻址的落盘名：sha256 十六进制的前 HashHexLen 位 + ext。 +// 同一份字节配同一个扩展名必然得到同一个名字，去重因此是「同名」的自然结果。 +func hashName(sum []byte, ext string) string { + return hex.EncodeToString(sum)[:HashHexLen] + ext +} diff --git a/internal/runtime/documents/store_meta.go b/internal/runtime/documents/store_meta.go new file mode 100644 index 0000000..9152330 --- /dev/null +++ b/internal/runtime/documents/store_meta.go @@ -0,0 +1,62 @@ +package documents + +// 本文件归属 `documents` 元数据表（迁移 0014）的**唯一写入实现**（design/24 +// §3.2；切片 M1）。S1 把这条 SQL 写在网关里；抽到共享存储核心之后，Web 上传与 +// IM 入站附件必须共用同一份写入逻辑，否则「哪条来源写了元数据、哪条没写」会 +// 分叉，S7 的引用计数与 GC 也就失去单一事实来源。 +// +// 沿用 S1 的两条既有事实（刻意不改，改动属另一片）： +// - 每次写入 `db.Open` 开一条新连接（design/24 §9.2-5 已登记的技术债）； +// - 同一存储名重复上传 `ON CONFLICT(name) DO NOTHING` —— 首次的 +// uploader/created_at 是权威，去重不刷新它们。 + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// ErrMetadataNoPath 报告元数据写入器没有数据库路径（未接库的构建/测试）。 +// 调用方据此不构造写入器（Store.Meta 留 nil），而不是拿到一个写不进去的替身。 +var ErrMetadataNoPath = errors.New("documents: 元数据写入器缺少数据库路径") + +// SQLMetadataWriter 把元数据行写进 SQLite 的 `documents` 表。它实现 +// MetadataWriter，供网关（Web 上传）与组合根（IM 入站附件）共用。 +type SQLMetadataWriter struct { + path string + log *slog.Logger +} + +// MetadataWriterFor 按数据库路径给出元数据写入面；路径为空（或全空白）时返回 +// **真正的 nil 接口**，调用方可以直接把它放进 `StoreConfig.Meta` 而不会踩 +// 「接口里装着一个 nil 具体值」的陷阱（那会让「未注入」判定失效，把一次成功的 +// 落盘变成 500）。未接库的构建/测试因此照旧可用：Store 按「未注入」跳过并留痕。 +func MetadataWriterFor(path string, log *slog.Logger) MetadataWriter { + if strings.TrimSpace(path) == "" { + return nil + } + return &SQLMetadataWriter{path: path, log: log} +} + +// WriteDocument 写入一行文档元数据。参数化 SQL、无拼接，错误按项目规范 wrap。 +func (w *SQLMetadataWriter) WriteDocument(ctx context.Context, rec Record) error { + if w == nil || w.path == "" { + return fmt.Errorf("documents: 写元数据: %w", ErrMetadataNoPath) + } + conn, err := db.Open(ctx, w.path, w.log) + if err != nil { + return fmt.Errorf("documents: 打开元数据库: %w", err) + } + defer func() { _ = conn.Close() }() + const query = `INSERT INTO documents (name, sha256, mime, size, uploader) +VALUES (?, ?, ?, ?, ?) +ON CONFLICT(name) DO NOTHING` + if _, err := conn.ExecContext(ctx, query, rec.Name, rec.SHA256, rec.MIME, rec.Size, rec.Uploader); err != nil { + return fmt.Errorf("documents: 插入元数据行: %w", err) + } + return nil +} diff --git a/internal/runtime/documents/store_test.go b/internal/runtime/documents/store_test.go new file mode 100644 index 0000000..f8db790 --- /dev/null +++ b/internal/runtime/documents/store_test.go @@ -0,0 +1,310 @@ +package documents + +// 本文件锁定**文档存储核心**（design/24 §3.2；切片 M1）的规则：白名单放行/拒绝、 +// 字节嗅探否决客户端声明、内容寻址去重、限额、路径穿越、元数据落表。规则与 S1 +// 逐字同源（`internal/services/gateway/documents.go` 的既有断言是另一侧的机器证明： +// 网关改成调用本实现之后，那批断言必须原样全绿）。 +// +// 断言一律走哨兵错误 + errors.Is，**不按错误文本子串断言** —— 文案会变，契约不变。 + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "os" + "path/filepath" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// whitelistExtensions 是 design/24 §9.2-2 定稿的 22 项白名单。**独立列一遍**， +// 使「白名单被悄悄改宽/改窄」这件事有用例能发现（值本身必须钉住，而不只是 +// 「白名单里的项能被放行」这种自洽测试）。 +var whitelistExtensions = []string{ + ".txt", ".md", ".markdown", ".csv", ".tsv", ".json", ".jsonl", + ".yaml", ".yml", ".log", ".ini", ".cfg", ".toml", + ".go", ".ts", ".tsx", ".java", ".c", ".h", ".cpp", ".rs", ".sql", +} + +// deniedExtensions 是必须被拒的扩展名（含「嗅探成 text/plain 也不放行」的脚本类）。 +var deniedExtensions = []string{ + ".html", ".svg", ".xml", ".zip", ".docx", ".xlsx", ".exe", ".dll", + ".bat", ".cmd", ".ps1", ".js", ".mjs", ".sh", ".py", ".pdf", "", +} + +// newTestStore 构造一个落在临时目录的 Store（默认上限）。 +func newTestStore(t *testing.T, cfg StoreConfig) *Store { + t.Helper() + if cfg.UploadsDir == "" && cfg.DataDir == "" { + cfg.UploadsDir = filepath.Join(t.TempDir(), uploadsSubDir, documentSubDir) + } + return NewStore(cfg) +} + +// expectedName 复算一份内容在给定扩展名下的期望落盘名（内容寻址口径的独立实现）。 +func expectedName(t *testing.T, ext string, data []byte) (string, string) { + t.Helper() + sum := sha256.Sum256(data) + hexSum := hex.EncodeToString(sum[:]) + return hexSum[:HashHexLen] + ext, hexSum +} + +// TestStoreSaveAcceptsWhitelistedExtensions 覆盖验收第 1 条：22 项白名单逐项放行， +// 落盘名是内容寻址名、MIME 是嗅探基类型、内容逐字节一致。 +func TestStoreSaveAcceptsWhitelistedExtensions(t *testing.T) { + t.Parallel() + if len(whitelistExtensions) != 22 { + t.Fatalf("白名单项数 = %d, want 22（design/24 §9.2-2 定稿）", len(whitelistExtensions)) + } + store := newTestStore(t, StoreConfig{}) + ctx := context.Background() + for _, ext := range whitelistExtensions { + if !ExtAllowed(ext) { + t.Fatalf("ExtAllowed(%q) = false, want true（白名单缺项或拼写变了）", ext) + } + data := []byte("昔涟的测试文本 " + ext + "\n") + wantName, wantSHA := expectedName(t, ext, data) + ref, err := store.Save(ctx, "orig"+ext, data) + if err != nil { + t.Fatalf("%s: Save: %v", ext, err) + } + if ref.Name != wantName || ref.SHA256 != wantSHA { + t.Fatalf("%s: name/sha = (%q,%q), want (%q,%q)", ext, ref.Name, ref.SHA256, wantName, wantSHA) + } + if ref.MIME != SniffOKMIME || ref.Size != int64(len(data)) || ref.Deduped { + t.Fatalf("%s: mime/size/deduped = (%q,%d,%v), want (%q,%d,false)", + ext, ref.MIME, ref.Size, ref.Deduped, SniffOKMIME, len(data)) + } + stored, err := os.ReadFile(ref.Path) + if err != nil { + t.Fatalf("%s: 盘上读不到 %s: %v", ext, ref.Path, err) + } + if !bytes.Equal(stored, data) { + t.Fatalf("%s: 落盘内容与输入不一致", ext) + } + } +} + +// TestStoreSaveRejectsDisallowedExtensions 覆盖验收第 1 条的拒绝侧：deny-by-default， +// 含 `.js`/`.py`/`.sh` 这类「内容确是纯文本但扩展名不在白名单」的样本。 +func TestStoreSaveRejectsDisallowedExtensions(t *testing.T) { + t.Parallel() + if ExtAllowed("") { + t.Fatal("ExtAllowed(\"\") = true, want false（无扩展名必须拒绝）") + } + store := newTestStore(t, StoreConfig{}) + ctx := context.Background() + for _, ext := range deniedExtensions { + if ExtAllowed(ext) { + t.Fatalf("ExtAllowed(%q) = true, want false", ext) + } + if _, err := store.Save(ctx, "sample"+ext, []byte("const a = 1;\n")); !errors.Is(err, ErrExtNotAllowed) { + t.Fatalf("%s: err = %v, want ErrExtNotAllowed", ext, err) + } + } + if entries, err := os.ReadDir(store.Dir()); err == nil && len(entries) != 0 { + t.Fatalf("被拒的扩展名仍在盘上留下 %d 个文件", len(entries)) + } +} + +// TestStoreSaveSniffVetoesContent 覆盖验收第 1 条的嗅探侧：文件名与内容不符 +// （PNG / HTML / zip 字节顶着 `.txt`）时，字节嗅探是唯一裁决者，一律拒绝。 +func TestStoreSaveSniffVetoesContent(t *testing.T) { + t.Parallel() + store := newTestStore(t, StoreConfig{}) + ctx := context.Background() + png := append([]byte("\x89PNG\r\n\x1a\n"), bytes.Repeat([]byte{0x00}, 64)...) + zip := append([]byte("PK\x03\x04"), bytes.Repeat([]byte{0x00}, 64)...) + cases := []struct { + label string + data []byte + }{ + {"PNG", png}, + {"HTML", []byte("hi")}, + {"zip", zip}, + {"NUL 字节", append([]byte("text\x00more"), bytes.Repeat([]byte{0x01}, 32)...)}, + } + for _, tc := range cases { + if _, err := store.Save(ctx, "lie.txt", tc.data); !errors.Is(err, ErrContentNotText) { + t.Fatalf("%s: err = %v, want ErrContentNotText", tc.label, err) + } + } + if _, err := os.Stat(store.Dir()); !os.IsNotExist(err) { + t.Fatalf("被嗅探否决的写入仍创建了文档目录（err=%v）", err) + } +} + +// TestStoreSaveDeduplicatesByContent 覆盖验收第 1 条的去重侧：同字节同扩展名 +// ⇒ 同名 + Deduped=true + 盘上只留一份；换客户端文件名不影响落盘名。 +func TestStoreSaveDeduplicatesByContent(t *testing.T) { + t.Parallel() + store := newTestStore(t, StoreConfig{}) + ctx := context.Background() + data := []byte("同一份内容的两次上传\n") + + first, err := store.Save(ctx, "first.txt", data) + if err != nil { + t.Fatalf("首次 Save: %v", err) + } + if first.Deduped { + t.Fatal("首次 Save 报告 Deduped = true") + } + second, err := store.Save(ctx, "renamed.txt", data) + if err != nil { + t.Fatalf("重复 Save: %v", err) + } + if !second.Deduped || second.Name != first.Name || second.Path != first.Path { + t.Fatalf("重复 Save = (%q,%v,%q), want 与首次同名同路径且 Deduped", second.Name, second.Deduped, second.Path) + } + entries, err := os.ReadDir(store.Dir()) + if err != nil { + t.Fatalf("读文档目录: %v", err) + } + if len(entries) != 1 { + t.Fatalf("盘上文件数 = %d, want 1（同字节同扩展名只留一份）", len(entries)) + } +} + +// TestStoreSaveEnforcesLimit 覆盖验收第 1 条的限额侧：≤0 归一到 DefaultMaxBytes、 +// 恰好等于上限放行、多一个字节拒绝。 +func TestStoreSaveEnforcesLimit(t *testing.T) { + t.Parallel() + const limit = 1024 + if got := NewStore(StoreConfig{}).Limit(); got != DefaultMaxBytes { + t.Fatalf("未配置上限时 Limit() = %d, want %d", got, DefaultMaxBytes) + } + store := newTestStore(t, StoreConfig{MaxBytes: limit}) + ctx := context.Background() + if _, err := store.Save(ctx, "exact.txt", bytes.Repeat([]byte("a"), limit)); err != nil { + t.Fatalf("恰好等于上限被拒: %v", err) + } + if _, err := store.Save(ctx, "over.txt", bytes.Repeat([]byte("b"), limit+1)); !errors.Is(err, ErrTooLarge) { + t.Fatalf("超限一个字节: err = %v, want ErrTooLarge", err) + } + if _, err := store.Save(ctx, "empty.txt", nil); !errors.Is(err, ErrEmpty) { + t.Fatalf("空内容: err = %v, want ErrEmpty", err) + } + if _, err := NewStore(StoreConfig{}).Save(ctx, "no.txt", []byte("x")); !errors.Is(err, ErrNoDir) { + t.Fatalf("未配置目录: err = %v, want ErrNoDir", err) + } +} + +// TestStoreSaveNeverUsesOriginalName 覆盖验收第 1 条的路径侧：客户端原始文件名 +// （含 `..`、分隔符、绝对路径）**永不参与落盘路径**，落盘名恒为内容寻址名。 +func TestStoreSaveNeverUsesOriginalName(t *testing.T) { + t.Parallel() + root := t.TempDir() + dir := filepath.Join(root, uploadsSubDir, documentSubDir) + store := newTestStore(t, StoreConfig{UploadsDir: dir}) + ctx := context.Background() + names := []string{ + "../../escape.txt", + `..\..\escape.txt`, + "/etc/passwd.txt", + "nested/dir/name.txt", + "..txt", + } + for _, origName := range names { + ref, err := store.Save(ctx, origName, []byte("路径穿越样本 "+origName+"\n")) + if err != nil { + t.Fatalf("%q: Save: %v", origName, err) + } + if filepath.Dir(ref.Path) != filepath.Clean(dir) { + t.Fatalf("%q: 落盘路径 %q 逃出了文档目录 %q", origName, ref.Path, dir) + } + if ref.Name != filepath.Base(ref.Name) || strings.ContainsAny(ref.Name, `/\`) { + t.Fatalf("%q: 落盘名 %q 不是单个路径段", origName, ref.Name) + } + if ext := ExtOf(origName); !strings.HasSuffix(ref.Name, ext) { + t.Fatalf("%q: 落盘名 %q 丢了扩展名 %q", origName, ref.Name, ext) + } + } + // 整个数据目录里只应出现文档目录与文档目录下的文件，别处没有逃逸产物。 + roots, err := os.ReadDir(root) + if err != nil { + t.Fatalf("读数据目录: %v", err) + } + for _, e := range roots { + if e.Name() != uploadsSubDir { + t.Fatalf("数据目录下出现了意外条目 %q（路径穿越没有收敛在文档目录内）", e.Name()) + } + } +} + +// TestStoreSaveWritesMetadataRow 覆盖验收第 1 条的元数据侧：注入写入器时 +// `documents` 表确有对应行（name/sha256/mime/size/uploader），重复上传 +// `ON CONFLICT DO NOTHING` 不新增第二行；未注入时跳过但不影响落盘。 +func TestStoreSaveWritesMetadataRow(t *testing.T) { + t.Parallel() + dataDir := t.TempDir() + dbPath := filepath.Join(dataDir, "store.db") + conn, err := db.Open(context.Background(), dbPath, nil) + if err != nil { + t.Fatalf("打开临时库: %v", err) + } + if err := db.MigrateUp(context.Background(), conn, nil); err != nil { + _ = conn.Close() + t.Fatalf("临时库跑迁移: %v", err) + } + if err := conn.Close(); err != nil { + t.Fatalf("关闭临时库: %v", err) + } + if MetadataWriterFor("", nil) != nil { + t.Fatal("空路径应得到 nil 接口（调用方据此不接线）") + } + if MetadataWriterFor(" ", nil) != nil { + t.Fatal("空白路径应得到 nil 接口") + } + + store := newTestStore(t, StoreConfig{ + DataDir: dataDir, + Uploader: "acct", + Meta: MetadataWriterFor(dbPath, nil), + }) + ctx := context.Background() + data := []byte("元数据行断言\n") + ref, err := store.Save(ctx, "meta.txt", data) + if err != nil { + t.Fatalf("Save: %v", err) + } + if _, err := store.Save(ctx, "meta-again.txt", data); err != nil { + t.Fatalf("重复 Save: %v", err) + } + + check, err := db.Open(ctx, dbPath, nil) + if err != nil { + t.Fatalf("复开临时库: %v", err) + } + defer func() { _ = check.Close() }() + var name, sha, mimeType, uploader string + var size int64 + if err := check.QueryRowContext(ctx, + `SELECT name, sha256, mime, size, uploader FROM documents WHERE name = ?`, ref.Name). + Scan(&name, &sha, &mimeType, &size, &uploader); err != nil { + t.Fatalf("documents 表里没有对应元数据行: %v", err) + } + if name != ref.Name || sha != ref.SHA256 || mimeType != SniffOKMIME || size != int64(len(data)) || uploader != "acct" { + t.Fatalf("元数据行 = (%q,%q,%q,%d,%q), want 与 Ref 一致且 uploader=acct", name, sha, mimeType, size, uploader) + } + var rows int + if err := check.QueryRowContext(ctx, `SELECT COUNT(*) FROM documents`).Scan(&rows); err != nil { + t.Fatalf("数元数据行: %v", err) + } + if rows != 1 { + t.Fatalf("documents 行数 = %d, want 1（同内容重复上传不新增行）", rows) + } + + // 未注入写入器：落盘照旧成功，只跳过写表。 + plain := newTestStore(t, StoreConfig{UploadsDir: filepath.Join(t.TempDir(), documentSubDir)}) + ref2, err := plain.Save(ctx, "nowriter.md", []byte("# 无元数据\n")) + if err != nil { + t.Fatalf("未注入写入器时 Save: %v", err) + } + if _, err := os.Stat(ref2.Path); err != nil { + t.Fatalf("未注入写入器时文件没落盘: %v", err) + } +} diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 628c615..0bd5056 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -23,6 +23,7 @@ import ( "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/logger" + "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" @@ -148,6 +149,12 @@ type Options struct { // 被移入某个角色的作用域（design/22 §7-4）。为 nil 时所有渠道都使用全局 // 默认角色。 ChannelPersona func(platform, accountID string) string + // DocumentStore 是 IM **入站附件**的落盘配置（design/24 §3；切片 M1）： + // 目录、单文件上限与 `documents` 元数据写入面。它由组合根给出（桥接器不猜 + // 目录布局），并与 Web 上传共用同一个存储核心。UploadsDir 与 DataDir 都为 + // 空即视为**未接线**：带文件的消息照旧能跑（模型会收到一行「读不到」的诚实 + // 提示），只是不落库、不抽取 —— 绝不假装读到了。 + DocumentStore documents.StoreConfig } // Bridge 掌管渠道适配器生命周期与入站消息流水线。它 @@ -195,6 +202,9 @@ type Bridge struct { // 都由 mu 保护。 channelPersona func(platform, accountID string) string adapterPersona map[string]string + // documentsCfg 是入站附件落盘配置（design/24 §3；切片 M1）。零值 = 未接线： + // 带文件的消息照旧能跑，模型会收到一行「读不到」的诚实提示（绝不静默）。 + documentsCfg documents.StoreConfig // proactiveCursor 是主动推送「无信号」兜底时的轮转游标 // （channelbridge/proactive.go），使默认渠道不会总是 // 白名单中的第一个主人渠道。由 mu 保护。 @@ -267,6 +277,7 @@ func New(opts Options) (*Bridge, error) { factory: opts.Factory, channelPersona: opts.ChannelPersona, adapterPersona: make(map[string]string), + documentsCfg: opts.DocumentStore, cancel: func() {}, sem: make(chan struct{}, maxConcurrent), }, nil diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index 17b4f55..44625fe 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -134,7 +134,10 @@ func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) { // （项目红线）。 func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) { text := strings.TrimSpace(msg.Content) - if text == "" { + // 无文本且无文件类附件 ⇒ 没有可处理的内容（与既有行为一致）。带附件的 + // 「纯文件消息」必须继续往下走：附件就是它的内容（design/24 §4.4 的 + // 「不许静默丢」）。 + if text == "" && !hasFileMedia(msg) { b.log.Debug("inbound message has no chat text, skipped", "platform", msg.Platform, "channel", msg.ChannelID) return @@ -220,13 +223,19 @@ func inboundTargetFor(sess session.Session, platform string) inboundTarget { } // runInboundTurn 为被放行的消息驱动 agent 轮次：解析 -// 路由目标、构建模型可见的消息列表、在轮次之前持久化入站 +// 路由目标、把消息携带的文件类附件**落库并挂成文档部件** +// （design/24 §3；切片 M1）、构建模型可见的消息列表、在轮次之前持久化入站 // 用户消息（因此即使渠道回复失败它也已落地），运行 // 循环并把结果交给 deliverReply。 func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, text string, actor adapter.Actor, sess session.Session) { target := inboundTargetFor(sess, msg.Platform) + // 入站附件摄取留在**闸门之后**（本函数只在消息被放行后调用）：被黑白名单或 + // 最小打扰拦下的消息绝不在盘上留下文件。成功者挂成文档部件，装配期 + // resolveDocuments 照既有链路抽取/注入；读不到的走诚实提示。 + attachments := b.ingestInboundDocuments(ctx, msg) + // 第三方边界只在受限（群聊/陌生人）上下文中生效： // 记忆只读（MemWritable=false），并携带发送者名， // 使 agent 能看到是谁在说话（design/16 §4.6 §1-2）。 @@ -241,6 +250,7 @@ func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage, runEnv.PersonaID = b.resolvePersona(ctx, sess, target, adapterName(a)) senderName := senderDisplayName(msg, actor) loopMsgs := b.buildLoopMessages(ctx, text, runEnv, actor, senderName, msg) + attachInboundAttachments(loopMsgs, attachments) // 主人私聊消息落在 Web 主会话并广播给 Web 客户端； // 受限消息落在其隔离会话（适配器日志页）， @@ -307,6 +317,30 @@ func deviceIDFor(msg adapter.UnifiedMessage) string { return msg.Platform + ":" + msg.ChannelID } +// attachInboundAttachments 把入站附件摄取的结果挂到消息列表里的**用户回合**上 +// （design/24 §3/§4.4；切片 M1）：文档部件与「读不到」的诚实提示行都只进 +// `RoleUser`，且挂在那条已构建好的用户消息上 —— 来源系统行、第三方边界、 +// SenderName 等既有装配因此逐字不变（本切片不重排消息，只追加部件）。 +// +// 一个回合只有一条用户消息（buildLoopMessages 的既有形状），因此找到第一条即 +// 挂载并返回；没有附件产物时是空操作。文本部件为空且既有文本为空时**不塞空文本 +// 部件**（provider 会拒）：附件部件本身就是这条消息的内容。 +func attachInboundAttachments(msgs []model.Message, attach inboundAttachments) { + if len(attach.parts) == 0 && len(attach.notices) == 0 { + return + } + for i := range msgs { + if msgs[i].Role != model.RoleUser { + continue + } + msgs[i].Content = append(msgs[i].Content, attach.parts...) + if notice := attach.noticeText(); notice != "" { + msgs[i].Content = append(msgs[i].Content, model.Content{Type: model.ContentTypeText, Text: notice}) + } + return + } +} + // buildLoopMessages 为一次入站轮次构建模型可见的消息列表。它 // 用 "[来源: · ]" 前缀装饰用户消息，使 // 模型能看到消息来自哪个平台、哪个会话（design/16 @@ -318,8 +352,12 @@ func deviceIDFor(msg adapter.UnifiedMessage) string { // SessionEnv/actor 结构化决定，绝不由 LLM 推断。原始消息文本 // 保持不动（构建带前缀的副本），因此持久化内容与 // 来源平台投递的内容一致。 +// +// 入站附件摄取的产物（已落库的文档部件与「读不到」的诚实提示行）由 +// attachInboundAttachments 挂到**同一条 RoleUser 消息**上 —— 用户内容只进用户 +// 角色消息是项目红线。 func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.SessionEnv, actor adapter.Actor, senderName string, src adapter.UnifiedMessage) []model.Message { - msg := model.TextMessage(model.RoleUser, text) + msg := userTurnMessage(text) thirdParty := env.Restricted() && !actor.IsOwner() && senderName != "" if thirdParty { // 说话者由结构化元数据承载，不织入用户文本 @@ -349,6 +387,16 @@ func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.Sess return []model.Message{model.TextMessage(model.RoleSystem, sys), msg} } +// userTurnMessage 构建入站回合的用户消息基底：文本为空时**不塞空文本部件** +// （provider 会拒空文本块），让附件部件或诚实提示行本身成为这条消息的内容 —— +// 与网关 `baseUserMessage` 对纯附件回合的口径一致（design/24 §4.2）。 +func userTurnMessage(text string) model.Message { + if text == "" { + return model.Message{Role: model.RoleUser} + } + return model.TextMessage(model.RoleUser, text) +} + // thirdPartyBoundaryPrompt 是加在受限（群聊/陌生人）上下文 // 最前面的系统消息，让模型知道身份边界。只有当会话 // 在结构上受限时才注入（design/16 §4.6 §1）； diff --git a/internal/services/channelbridge/inbound_documents.go b/internal/services/channelbridge/inbound_documents.go new file mode 100644 index 0000000..01d8de5 --- /dev/null +++ b/internal/services/channelbridge/inbound_documents.go @@ -0,0 +1,290 @@ +package channelbridge + +// 本文件归属 **IM 入站附件摄取**（design/24 §3 / §4.4；切片 M1）：把入站消息携带的 +// 文件类媒体交给共享文档存储（`internal/runtime/documents` 的 Store —— 与 Web +// 上传同一份实现），成功者挂成该消息的文档部件，让装配期 `resolveDocuments` +// 照既有链路完成抽取/注入/清单/结论回写，**不需要新链路**。 +// +// 读不到的附件**绝不静默**（design/24 §4.4）：模型会收到一行诚实提示，日志留一条 +// 带来源的记录（只记 name/mime/size，**绝不记正文**）。 +// +// 四条边界： +// - **图片/语音不走这里**（design/24 §4.2 末：图片走既有视觉链路），本切片只补 +// 文件类； +// - 入站附件是**用户内容** ⇒ 文档部件只挂在 `model.RoleUser` 消息上； +// - 摄取发生在**闸门之后**（会话解析、黑白名单、最小打扰判定都通过之后）， +// 被拒的消息因此绝不会在盘上留下文件； +// - 输出仍只走 chat 文本：IM 侧不新增任何文件出站能力。 + +import ( + "context" + "errors" + "fmt" + "strings" + "unicode" + "unicode/utf8" + + "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/model" +) + +// 入站附件的具名常量与提示文案（design/24 §4.4）。提示行的形状与循环的清单行 +// 一致（`[附件 <名> · …]`），但**只进模型可见视图**，不写进持久化正文 —— +// design/24 §4.3 要求人类可见的来源标记归客户端渲染层，绝不拼进正文。 +const ( + // inboundAttachmentNameMaxRunes 是提示行里原始文件名的展示上限。文件名来自 + // 渠道（发送方可任意构造），因此限长之外还要**剔控制字符**：换行会让一条 + // 附件名在模型上下文里伪造出额外的行。 + inboundAttachmentNameMaxRunes = 64 + + // inboundAttachmentMIMEMaxRunes 是提示行里媒体类型的展示上限。 + inboundAttachmentMIMEMaxRunes = 48 + + // inboundAttachmentUnknownName 是文件名缺失（或净化后为空）时的占位。 + inboundAttachmentUnknownName = "未命名文件" + + // inboundAttachmentUnknownMIME 是渠道未给 MIME 时的展示值（与循环清单行的 + // 「未知类型」同一口径：宁可写出来，也不留一个空洞让模型误读）。 + inboundAttachmentUnknownMIME = "未知类型" + + // inboundUploaderPrefix 是写进 `documents` 元数据的上传者身份前缀，用来与 + // Web 账号身份分开（渠道来源 vs 登录账号）。 + inboundUploaderPrefix = "im:" + + // 诚实提示行（design/24 §4.4）。参考 `%s=展示名`、`%s=展示 MIME`、`%d=字节数`。 + // 文案按**拒绝原因**分类：把「超限」说成「格式不支持」本身就是一种不诚实。 + inboundNoticeExtDenied = "[附件 %s · %s · %d 字节 · 无法读取：格式不在可读白名单内（只支持纯文本类文档）]" + inboundNoticeNotText = "[附件 %s · %s · %d 字节 · 无法读取：内容不是纯文本（类型由字节嗅探裁决）]" + inboundNoticeTooLarge = "[附件 %s · %s · %d 字节 · 无法读取：文件超过服务端体积上限]" + inboundNoticeEmpty = "[附件 %s · %s · %d 字节 · 无法读取：文件为空]" + inboundNoticeStoreError = "[附件 %s · %s · %d 字节 · 无法读取：服务端保存该附件失败]" + inboundNoticeNoBytes = "[附件 %s · %s · %d 字节 · 无法读取：渠道只给了远程引用，服务端未代取文件字节]" + inboundNoticeUnwired = "[附件 %s · %s · %d 字节 · 无法读取：服务端未接线文档存储]" + inboundNoticeTooMany = "[另有附件未处理：单条消息最多 %d 个文件]" +) + +// inboundAttachments 是一次入站附件摄取的产物。 +type inboundAttachments struct { + // parts 是已落库的文档部件（挂在同一条 RoleUser 消息上）。 + parts []model.Content + // notices 是「有附件读不到」的诚实提示行，顺序与消息里的文件媒体顺序一致。 + notices []string +} + +// noticeText 把提示行拼成一段文本；没有提示时返回空串（调用方据此不塞空文本部件）。 +func (a inboundAttachments) noticeText() string { + return strings.Join(a.notices, "\n") +} + +// hasFileMedia 报告消息是否携带文件类媒体（决定「无文本消息」是否仍要跑流水线）。 +func hasFileMedia(msg adapter.UnifiedMessage) bool { + for _, m := range msg.Media { + if isFileMedia(msg, m) { + return true + } + } + return false +} + +// isFileMedia 报告一条媒体是不是**文件类**。部分适配器只给内容类型而不给媒体 +// Kind（`ContentType == file`，附件就在 Media 里），那种消息也按文件处理，否则 +// 附件会被静默忽略 —— 静默忽略正是本切片要消灭的行为。 +func isFileMedia(msg adapter.UnifiedMessage, m adapter.Media) bool { + if m.Kind == adapter.MediaFile { + return true + } + return m.Kind == "" && msg.ContentType == adapter.ContentFile +} + +// fileMedia 返回消息里全部文件类媒体（保序）。 +func fileMedia(msg adapter.UnifiedMessage) []adapter.Media { + out := make([]adapter.Media, 0, len(msg.Media)) + for _, m := range msg.Media { + if isFileMedia(msg, m) { + out = append(out, m) + } + } + return out +} + +// ingestInboundDocuments 把一条入站消息的文件类媒体落进共享文档存储，返回要挂到 +// 该消息上的文档部件与「读不到」的诚实提示。调用点必须在**放行之后**：被黑白名单 +// 或最小打扰拦下的消息不会在盘上留下任何文件。 +func (b *Bridge) ingestInboundDocuments(ctx context.Context, msg adapter.UnifiedMessage) inboundAttachments { + files := fileMedia(msg) + if len(files) == 0 { + return inboundAttachments{} + } + store := b.inboundDocumentStore(msg) + if store == nil { + // 未接线：绝不假装读到了。每个附件都得到一行诚实提示 + 一条带来源的日志。 + b.log.Warn("inbound attachment ingest skipped: document store not wired", + "platform", msg.Platform, "channel", msg.ChannelID, "files", len(files)) + out := inboundAttachments{notices: make([]string, 0, len(files))} + for _, m := range files { + out.notices = append(out.notices, noticeLine(m, inboundNoticeUnwired)) + } + return out + } + return b.ingestFiles(ctx, store, msg, files) +} + +// ingestFiles 逐个落盘并分类：成功挂文档部件，被拒/无字节则记一条带来源的日志 + +// 一行诚实提示。条数上限与 Web 侧同源（documents.MaxPerMessage），超出的附件 +// **不静默**：整条提示一次，且日志记下实际条数。 +func (b *Bridge) ingestFiles(ctx context.Context, store *documents.Store, msg adapter.UnifiedMessage, files []adapter.Media) inboundAttachments { + out := inboundAttachments{ + parts: make([]model.Content, 0, len(files)), + notices: make([]string, 0, len(files)), + } + for i, media := range files { + if i >= documents.MaxPerMessage { + b.log.Warn("inbound attachments beyond per-message limit ignored", + "platform", msg.Platform, "channel", msg.ChannelID, + "files", len(files), "limit", documents.MaxPerMessage) + out.notices = append(out.notices, fmt.Sprintf(inboundNoticeTooMany, documents.MaxPerMessage)) + break + } + if len(media.Data) == 0 { + // 渠道只给了远程引用（CDN URL / 平台文件 id）：本切片**不代取** —— + // 服务端代取任意 http(s) 正是 F3 的教训（design/24 §5），需要专门 + // 的裁决与地址守卫，不在这里顺手打开。绝不静默。 + b.log.Warn("inbound attachment has no inlined bytes", + "platform", msg.Platform, "channel", msg.ChannelID, + "name", media.Name, "mime", media.MIMEType, "size", media.Size) + out.notices = append(out.notices, noticeLine(media, inboundNoticeNoBytes)) + continue + } + ref, err := store.Save(ctx, media.Name, media.Data) + if err != nil { + // 白名单/嗅探/限额的裁决全在 store（与 Web 上传同源）。这里只如实 + // 记日志（name/mime/size，**绝不记正文**）并让模型看到「读不到」， + // 且提示文案按**拒绝原因**分类（把「超限」说成「格式不支持」也是一种 + // 不诚实）。 + b.log.Warn("inbound attachment rejected", + "platform", msg.Platform, "channel", msg.ChannelID, + "name", media.Name, "mime", media.MIMEType, "size", len(media.Data), "error", err) + out.notices = append(out.notices, noticeLine(media, rejectedNoticeFor(err))) + continue + } + b.log.Info("inbound attachment stored", + "platform", msg.Platform, "channel", msg.ChannelID, + "name", ref.Name, "mime", ref.MIME, "size", ref.Size, + "sha256", ref.SHA256, "deduped", ref.Deduped) + out.parts = append(out.parts, documentPart(ref)) + } + return out +} + +// inboundDocumentStore 按消息的渠道身份构造文档存储：目录/上限/元数据写入面来自 +// 组合根注入的配置，上传者身份按消息给出。未接线（UploadsDir 与 DataDir 都为空） +// 时返回 nil —— 调用方据此走「读不到」的诚实路径，而不是假装读到了。 +func (b *Bridge) inboundDocumentStore(msg adapter.UnifiedMessage) *documents.Store { + cfg := b.documentsCfg + if cfg.UploadsDir == "" && cfg.DataDir == "" { + return nil + } + cfg.Uploader = inboundUploader(msg) + return documents.NewStore(cfg) +} + +// inboundUploader 给出写进 `documents` 元数据的上传者身份（平台 + 本地账号 + +// 平台 uid）。身份**只来自渠道元数据**，绝不从文件名或内容里推断（项目红线）。 +func inboundUploader(msg adapter.UnifiedMessage) string { + return inboundUploaderPrefix + deviceIDFor(msg) +} + +// documentPart 把一条已落库的附件折成模型部件。身份一律取自**服务端存储名** +// （原始文件名永不进上下文）；URL 刻意留空：IM 侧没有文件出站能力、装配期也 +// 用不到读侧 URL，而落盘路径（Ref.Path）更是绝不能进模型上下文。 +func documentPart(ref documents.Ref) model.Content { + msg := model.DocumentMessage(model.RoleUser, "", model.Document{ + Name: ref.Name, + MIME: ref.MIME, + Size: ref.Size, + }) + return msg.Content[0] +} + +// rejectedNoticeFor 按 store 的哨兵错误给出对应的诚实提示模板（design/24 §4.4）： +// 分类只依据哨兵链，**错误原文只进日志、绝不进模型上下文**。 +func rejectedNoticeFor(err error) string { + switch { + case errors.Is(err, documents.ErrExtNotAllowed): + return inboundNoticeExtDenied + case errors.Is(err, documents.ErrContentNotText): + return inboundNoticeNotText + case errors.Is(err, documents.ErrTooLarge): + return inboundNoticeTooLarge + case errors.Is(err, documents.ErrEmpty): + return inboundNoticeEmpty + default: + return inboundNoticeStoreError + } +} + +// noticeLine 渲染一行诚实提示（design/24 §4.4）。名字与媒体类型都来自渠道， +// 因此先做展示净化：剔控制字符、限长、缺失给占位。 +func noticeLine(media adapter.Media, template string) string { + return fmt.Sprintf(template, displayName(media.Name), displayMIME(media.MIMEType), attachmentSize(media)) +} + +// attachmentSize 取附件的展示字节数：优先用渠道上报的大小，缺失或为负时退回 +// 实际拿到的内联字节数 —— 提示行里的数字宁可是「已知的最小值」，也不能是负数。 +func attachmentSize(media adapter.Media) int64 { + if media.Size > 0 { + return media.Size + } + return int64(len(media.Data)) +} + +// displayName 把渠道给的文件名整理成可安全放进模型上下文的展示名：去控制字符、 +// 折叠空白、限长；净化后为空时给占位名。 +func displayName(name string) string { + cleaned := strings.Map(dropAttachmentControl, name) + cleaned = strings.Join(strings.Fields(cleaned), " ") + if cleaned == "" { + return inboundAttachmentUnknownName + } + return truncateRunes(cleaned, inboundAttachmentNameMaxRunes) +} + +// displayMIME 只保留 MIME 里合法的字符（字母/数字与 `/+;=-.` 与空白），其余一律 +// 丢弃 —— 它同样是渠道给的字符串，绝不原样进上下文；净化后为空时给「未知类型」。 +func displayMIME(mime string) string { + cleaned := strings.Map(func(r rune) rune { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9': + return r + case r == '/', r == '+', r == ';', r == '=', r == '-', r == '.', r == ' ': + return r + default: + return -1 + } + }, mime) + cleaned = strings.TrimSpace(cleaned) + if cleaned == "" { + return inboundAttachmentUnknownMIME + } + return truncateRunes(cleaned, inboundAttachmentMIMEMaxRunes) +} + +// dropAttachmentControl 把换行/制表折叠为空格，其余控制字符直接丢弃。 +func dropAttachmentControl(r rune) rune { + if r == '\n' || r == '\r' || r == '\t' { + return ' ' + } + if unicode.IsControl(r) { + return -1 + } + return r +} + +// truncateRunes 按 rune 截断并加省略号（绝不把多字节字符切成两半）。 +func truncateRunes(s string, max int) string { + if max <= 0 || utf8.RuneCountInString(s) <= max { + return s + } + return string([]rune(s)[:max]) + "…" +} diff --git a/internal/services/channelbridge/inbound_documents_honesty_test.go b/internal/services/channelbridge/inbound_documents_honesty_test.go new file mode 100644 index 0000000..a69943b --- /dev/null +++ b/internal/services/channelbridge/inbound_documents_honesty_test.go @@ -0,0 +1,122 @@ +package channelbridge + +// 本文件承载 IM 入站附件摄取的**诚实性与边界**用例（design/24 §4.4；切片 M1）， +// 与 inbound_documents_test.go 是同一批夹具的搬家（夹具留在那边，同包可见）： +// +// - 拒绝原因必须**分类正确**：超限说超限，绝不笼统说成「格式不支持」； +// - 渠道只给远程引用（目前适配器的现实形态）时不代取、不静默； +// - 图片类媒体保持现状（不走文档摄取）； +// - 组合根未接线时照旧跑回合，但模型知道附件读不到。 + +import ( + "context" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// TestInboundOversizeFileIsHonestAboutSize 钉住诚实口径的**分类**：体积超限必须 +// 说「超过体积上限」，绝不能笼统说成「格式不支持」——把原因说错也是不诚实。 +func TestInboundOversizeFileIsHonestAboutSize(t *testing.T) { + t.Parallel() + env := newIngestEnv(t, documents.StoreConfig{MaxBytes: 8}, true) + msg := fileMessage("这个大吗", "big.txt", []byte("这是明显超过八个字节的正文\n"), "text/plain") + + env.bridge.process(context.Background(), msg) + + if names := docDirNames(t, env.uploads); len(names) != 0 { + t.Fatalf("超限附件被落库了：%v", names) + } + visible := env.model.visibleText() + if !strings.Contains(visible, "体积上限") { + t.Fatalf("超限附件的提示没有点明体积上限：%q", visible) + } + if strings.Contains(visible, "格式不在可读白名单内") { + t.Fatalf("超限附件被说成了「格式不支持」（原因说错）：%q", visible) + } + if log := env.logs.String(); !strings.Contains(log, "inbound attachment rejected") { + t.Fatalf("缺少带来源的拒绝日志：%q", log) + } +} + +// TestInboundRemoteOnlyMediaIsHonestWithoutFetch 钉住「渠道只给远程引用」这一现实形态 +// （适配器目前上报 URL/平台文件 id，而不是内联字节）：本切片**不代取**，但绝不静默。 +func TestInboundRemoteOnlyMediaIsHonestWithoutFetch(t *testing.T) { + t.Parallel() + env := newIngestEnv(t, documents.StoreConfig{}, true) + msg := directMessage() + msg.Content = "这个文件里是什么" + msg.Media = []adapter.Media{{ + Kind: adapter.MediaFile, URL: "https://cdn.example.com/secret-token=abc/report.pdf", + Name: "report.pdf", Size: 1024, MIMEType: "application/pdf", + }} + + env.bridge.process(context.Background(), msg) + + if names := docDirNames(t, env.uploads); len(names) != 0 { + t.Fatalf("远程引用的附件被落库了：%v", names) + } + visible := env.model.visibleText() + if !strings.Contains(visible, "report.pdf") || !strings.Contains(visible, "远程引用") { + t.Fatalf("模型可见视图里没有诚实提示行：%q", visible) + } + log := env.logs.String() + if !strings.Contains(log, "no inlined bytes") { + t.Fatalf("缺少「无内联字节」的带来源日志：%q", log) + } + // 远程引用可能带凭据，绝不进日志。 + if strings.Contains(log, "secret-token=abc") { + t.Fatalf("日志里出现了远程引用原串（可能含凭据）：%q", log) + } +} + +// TestInboundImageMediaKeepsExistingPath 钉住边界：图片类媒体**不**走文档摄取 +// （design/24 §4.2 末：图片走既有视觉链路），因此不落库、不挂文档部件、不产生提示行。 +func TestInboundImageMediaKeepsExistingPath(t *testing.T) { + t.Parallel() + env := newIngestEnv(t, documents.StoreConfig{}, true) + msg := directMessage() + msg.Content = "看这张图" + msg.Media = []adapter.Media{{ + Kind: adapter.MediaImage, Name: "photo.png", MIMEType: "image/png", + Data: []byte("\x89PNG\r\n\x1a\n"), Size: 8, + }} + + env.bridge.process(context.Background(), msg) + + if names := docDirNames(t, env.uploads); len(names) != 0 { + t.Fatalf("图片类媒体被文档摄取落库了：%v", names) + } + visible := env.model.visibleText() + if strings.Contains(visible, "[附件 ") || strings.Contains(visible, "无法读取") { + t.Fatalf("图片类媒体产生了文档清单/提示行：%q", visible) + } + if !strings.Contains(visible, "看这张图") { + t.Fatalf("用户文本丢了：%q", visible) + } + if log := env.logs.String(); strings.Contains(log, "attachment") { + t.Fatalf("图片类媒体产生了附件日志：%q", log) + } +} + +// TestInboundFileMediaWithoutStoreIsHonest 钉住「未接线」路径：组合根没给落盘配置时， +// 带文件的消息照旧能跑（模型收到诚实提示），只是不落库 —— 绝不假装读到了。 +func TestInboundFileMediaWithoutStoreIsHonest(t *testing.T) { + t.Parallel() + env := newIngestEnv(t, documents.StoreConfig{}, false) + msg := fileMessage("这个文件呢", "notes.txt", []byte("正文\n"), "text/plain") + + env.bridge.process(context.Background(), msg) + + if env.model.calls() == 0 { + t.Fatal("未接线时回合没有跑到模型") + } + if !strings.Contains(env.model.visibleText(), "未接线文档存储") { + t.Fatalf("未接线时没有诚实提示：%q", env.model.visibleText()) + } + if log := env.logs.String(); !strings.Contains(log, "document store not wired") { + t.Fatalf("未接线时没有带来源的日志：%q", log) + } +} diff --git a/internal/services/channelbridge/inbound_documents_test.go b/internal/services/channelbridge/inbound_documents_test.go new file mode 100644 index 0000000..4b96cde --- /dev/null +++ b/internal/services/channelbridge/inbound_documents_test.go @@ -0,0 +1,395 @@ +package channelbridge + +// 本文件锁定 **IM 入站附件摄取**（design/24 §3 / §4.4；切片 M1）： +// +// - 文件类媒体 → 共享文档存储落库 → 装配期抽取 → **模型可见视图里出现清单行与 +// 正文**（真 store + 真抽取器 + 真循环 + 假模型，端到端）； +// - 不支持的格式**不落库**，但模型收到一行诚实提示、日志留一条带来源的记录， +// 且日志**绝不记正文**； +// - 渠道只给远程引用（无内联字节）时同样诚实，不静默； +// - 摄取在**闸门之后**：被拉黑/被隔离的消息绝不在盘上留下文件，其附件也只进 +// 隔离会话，不镜像到主人主会话； +// - 只有附件的消息（无文本）不再被判成「没有内容」而丢弃； +// - 图片类媒体保持现状（不走摄取，也不落库）。 + +import ( + "bytes" + "context" + "log/slog" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "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/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// stubModelAdapter 是最小的 model.Adapter：记下每次补全请求的**模型可见视图** +// （即装配期 resolveDocuments 替换之后的形态），并回一条固定文本。 +type stubModelAdapter struct { + mu sync.Mutex + seen [][]model.Message +} + +func (a *stubModelAdapter) Name() string { return "stub" } + +func (a *stubModelAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) { + a.mu.Lock() + a.seen = append(a.seen, req.Messages) + a.mu.Unlock() + return model.Response{Model: "stub", Content: "读完了", StopReason: "stop"}, nil +} + +func (a *stubModelAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) { + return a.Chat(ctx, req, model.Options{}) +} + +// visibleText 返回模型最后看到的全部消息文本（含系统与用户回合）。 +func (a *stubModelAdapter) visibleText() string { + a.mu.Lock() + defer a.mu.Unlock() + if len(a.seen) == 0 { + return "" + } + var b strings.Builder + for _, m := range a.seen[len(a.seen)-1] { + for _, c := range m.Content { + b.WriteString(c.Text) + b.WriteString("\n") + } + } + return b.String() +} + +// calls 返回模型被调用的次数（0 = 循环根本没跑）。 +func (a *stubModelAdapter) calls() int { + a.mu.Lock() + defer a.mu.Unlock() + return len(a.seen) +} + +// safeBuffer 是并发安全的日志缓冲（桥接器的入站路径可能在 goroutine 里写日志）。 +type safeBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *safeBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *safeBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// ingestEnv 是入站附件端到端用例的装配：**真 store + 真抽取器 + 真循环 + 假模型**， +// 因此模型看到的清单行与正文就是生产路径装配出来的那一份。 +type ingestEnv struct { + bridge *Bridge + model *stubModelAdapter + adapter *fakeAdapter + sessions *stubSessions + uploads string + logs *safeBuffer +} + +// newIngestEnv 装配上述端到端环境。cfg 给出入站附件的落盘配置（未给 UploadsDir 时 +// 落到临时目录）；cfg 为零值时即「未接线」用例（跳过 SetDocumentStore）。 +func newIngestEnv(t *testing.T, cfg documents.StoreConfig, wireStore bool) *ingestEnv { + t.Helper() + dataDir := t.TempDir() + uploads := cfg.UploadsDir + if uploads == "" { + uploads = filepath.Join(dataDir, "uploads", "documents") + } + cfg.UploadsDir = uploads + logs := &safeBuffer{} + log := slog.New(slog.NewTextHandler(logs, nil)) + + modelAdapter := &stubModelAdapter{} + ag, err := agent.New("测试系统提示词", nil, nil, agent.WithModel(modelAdapter, nil)) + if err != nil { + t.Fatalf("构造 agent: %v", err) + } + resolver := documents.NewResolver(documents.Config{DataDir: dataDir, UploadsDir: uploads, Log: log}) + runner := loop.New(loop.Options{ + Agent: ag, DocumentResolver: resolver, MaxRounds: 4, Log: log, + }) + sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}} + opts := Options{ + Log: log, Loop: runner, Sessions: sess, Owner: testOwnerProfile("telegram", "100"), + } + if wireStore { + opts.DocumentStore = cfg + } + b, err := New(opts) + if err != nil { + t.Fatalf("New: %v", err) + } + platform := &fakeAdapter{name: "telegram"} + b.Add(platform) + return &ingestEnv{ + bridge: b, model: modelAdapter, adapter: platform, + sessions: sess, uploads: uploads, logs: logs, + } +} + +// fileMessage 构造一条主人私聊消息，附带一个文件类媒体。 +func fileMessage(text, name string, data []byte, mime string) adapter.UnifiedMessage { + msg := directMessage() + msg.Content = text + msg.ContentType = adapter.ContentText + if text == "" { + msg.ContentType = adapter.ContentFile + } + msg.Media = []adapter.Media{{ + Kind: adapter.MediaFile, Name: name, Size: int64(len(data)), MIMEType: mime, Data: data, + }} + return msg +} + +// docDirNames 返回文档落盘目录里的条目名（目录不存在时返回 nil）。 +func docDirNames(t *testing.T, dir string) []string { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + if os.IsNotExist(err) { + return nil + } + t.Fatalf("读文档目录: %v", err) + } + out := make([]string, 0, len(entries)) + for _, e := range entries { + out = append(out, e.Name()) + } + return out +} + +// TestInboundFileMediaIngestedAndVisibleToModel 覆盖验收第 3 条：一条带文件媒体的 +// 入站消息 → 落库 → 抽取 → **模型可见视图里出现该附件的清单行与正文**。 +func TestInboundFileMediaIngestedAndVisibleToModel(t *testing.T) { + t.Parallel() + env := newIngestEnv(t, documents.StoreConfig{}, true) + body := "第一行：昔涟的测试附件\n第二行：正文必须真的进上下文\n" + msg := fileMessage("看看这个", "notes.txt", []byte(body), "text/plain") + + env.bridge.process(context.Background(), msg) + + names := docDirNames(t, env.uploads) + if len(names) != 1 { + t.Fatalf("落盘文件数 = %v, want 1（附件必须落库）", names) + } + stored, err := os.ReadFile(filepath.Join(env.uploads, names[0])) + if err != nil { + t.Fatalf("读落盘文件: %v", err) + } + if string(stored) != body { + t.Fatalf("落盘内容 = %q, want %q", stored, body) + } + if !strings.HasSuffix(names[0], ".txt") || len(names[0]) != documents.HashHexLen+len(".txt") { + t.Fatalf("落盘名 = %q, want <32 位十六进制>.txt（内容寻址）", names[0]) + } + + visible := env.model.visibleText() + if !strings.Contains(visible, "[附件 "+names[0]+" · ") { + t.Fatalf("模型可见视图里没有附件清单行：%q", visible) + } + if !strings.Contains(visible, "第二行：正文必须真的进上下文") { + t.Fatalf("模型可见视图里没有附件正文：%q", visible) + } + if !strings.Contains(visible, "看看这个") { + t.Fatalf("模型可见视图里丢了用户文本：%q", visible) + } + if log := env.logs.String(); !strings.Contains(log, "inbound attachment stored") || + !strings.Contains(log, "source=services/channelbridge") { + t.Fatalf("落库日志缺少来源或内容：%q", log) + } +} + +// TestInboundUnsupportedFileIsHonestAndNotStored 覆盖验收第 4 条：不支持的格式 +// 不落库，但模型知道「有附件读不到」，且日志带来源、只记 name/mime/size（绝不记正文）。 +func TestInboundUnsupportedFileIsHonestAndNotStored(t *testing.T) { + t.Parallel() + cases := []struct { + label string + name string + mime string + data []byte + }{ + {"扩展名不在白名单", "tool.py", "text/x-python", []byte("print('秘密内容')\n")}, + {"内容被嗅探否决", "lie.txt", "text/plain", append([]byte("\x89PNG\r\n\x1a\n"), bytes.Repeat([]byte{0x00}, 32)...)}, + } + for _, tc := range cases { + env := newIngestEnv(t, documents.StoreConfig{}, true) + env.bridge.process(context.Background(), fileMessage("跑一下", tc.name, tc.data, tc.mime)) + + if names := docDirNames(t, env.uploads); len(names) != 0 { + t.Fatalf("%s: 被拒的附件仍落库 %v", tc.label, names) + } + visible := env.model.visibleText() + if !strings.Contains(visible, tc.name) || !strings.Contains(visible, "无法读取") { + t.Fatalf("%s: 模型可见视图里没有诚实提示行：%q", tc.label, visible) + } + if env.model.calls() == 0 { + t.Fatalf("%s: 回合没有跑到模型", tc.label) + } + log := env.logs.String() + if !strings.Contains(log, "inbound attachment rejected") || !strings.Contains(log, tc.name) { + t.Fatalf("%s: 缺少带来源的拒绝日志：%q", tc.label, log) + } + if strings.Contains(log, "秘密内容") { + t.Fatalf("%s: 日志里出现了附件正文（红线：只记 name/mime/size）", tc.label) + } + } +} + +// TestInboundFileMediaBlacklistedNotIngested 覆盖验收第 5 条的一半：被黑白名单拉黑的 +// 群/陌生人消息，附件**不落库**、也不驱动循环（摄取在闸门之后）。 +func TestInboundFileMediaBlacklistedNotIngested(t *testing.T) { + t.Parallel() + uploads := filepath.Join(t.TempDir(), "uploads", "documents") + chat := &stubChatter{res: &loop.Result{Text: "不该回复"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}} + var rejected []string + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: &captureHistory{}, + DocumentStore: documents.StoreConfig{UploadsDir: uploads}, + OnRejected: func(_ adapter.UnifiedMessage, reason string) { rejected = append(rejected, reason) }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + ad := &fakeAdapter{name: "telegram", cfg: map[string]any{ + "bot_token": "configured", "denyGroups": []string{"-100"}, + }} + b.Add(ad) + + msg := adapter.UnifiedMessage{ + Platform: "telegram", ChannelID: "-100", ChannelType: adapter.ChannelGroup, + GroupID: "-100", SenderID: "9", SenderName: "陌生人", + Sender: adapter.Actor{Role: adapter.RoleMember}, Content: "给你个文件", + ContentType: adapter.ContentText, + Media: []adapter.Media{{ + Kind: adapter.MediaFile, Name: "stranger.txt", Size: 6, MIMEType: "text/plain", Data: []byte("你好\n"), + }}, + } + b.process(context.Background(), msg) + + if names := docDirNames(t, uploads); len(names) != 0 { + t.Fatalf("被拉黑的消息把附件落库了：%v", names) + } + if chat.sessionID != "" || chat.messages != nil { + t.Fatalf("被拉黑的消息仍驱动了循环：session=%q", chat.sessionID) + } + if len(rejected) != 1 || rejected[0] != "group_denied" { + t.Fatalf("拒绝原因 = %v, want [group_denied]", rejected) + } +} + +// TestInboundGroupFileStaysInIsolatedSession 覆盖验收第 5 条的另一半：第三方（群聊） +// 消息的附件只随该消息进入**隔离会话**，绝不镜像到主人主会话（沿用既有策略测试写法）。 +func TestInboundGroupFileStaysInIsolatedSession(t *testing.T) { + t.Parallel() + uploads := filepath.Join(t.TempDir(), "uploads", "documents") + chat := &stubChatter{res: &loop.Result{Text: "收到"}} + sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}} + hist := &captureHistory{} + b, err := New(Options{ + Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist, + DocumentStore: documents.StoreConfig{UploadsDir: uploads}, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + // 被点名的群消息 → 回复（design/12 §6 策略二），因此会真的走完流水线。 + 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) + + if names := docDirNames(t, uploads); len(names) != 1 { + t.Fatalf("群聊附件落盘文件数 = %v, want 1（第三方附件同样走共享存储）", names) + } + if chat.sessionID != "sess-g" { + t.Fatalf("循环会话 = %q, want 隔离会话 sess-g", chat.sessionID) + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("隔离会话落库条数 = %d, want 2", len(recs)) + } + for i, rec := range recs { + if rec.SessionID != "sess-g" { + t.Fatalf("record[%d] SessionID = %q, want 隔离会话（绝不镜像到主人主会话）", i, rec.SessionID) + } + } + // 用户消息在循环里带着文档部件（图片/文档只进用户回合）。 + var docParts int + for _, m := range chat.messages { + if m.Role == model.RoleSystem { + continue + } + for _, c := range m.Content { + if c.Type == model.ContentTypeDocument { + docParts++ + } + } + } + if docParts != 1 { + t.Fatalf("用户消息里的文档部件数 = %d, want 1", docParts) + } +} + +// TestInboundFileOnlyMessageReachesLoop 钉住「只有附件、没有文本」的消息不再被判成 +// 没有内容而丢弃，且那条消息**不塞空文本部件**（provider 会拒空文本块）。 +func TestInboundFileOnlyMessageReachesLoop(t *testing.T) { + t.Parallel() + env := newIngestEnv(t, documents.StoreConfig{}, true) + msg := fileMessage("", "only.md", []byte("# 只有附件\n正文\n"), "text/markdown") + + env.bridge.process(context.Background(), msg) + + if env.model.calls() == 0 { + t.Fatal("纯附件消息没有驱动循环") + } + names := docDirNames(t, env.uploads) + if len(names) != 1 { + t.Fatalf("落盘文件数 = %v, want 1", names) + } + visible := env.model.visibleText() + if !strings.Contains(visible, "[附件 "+names[0]+" · ") || !strings.Contains(visible, "# 只有附件") { + t.Fatalf("模型可见视图里没有清单行或正文：%q", visible) + } + env.model.mu.Lock() + defer env.model.mu.Unlock() + if len(env.model.seen) == 0 { + t.Fatal("没有记录到模型请求") + } + for _, m := range env.model.seen[0] { + if m.Role != model.RoleUser { + continue + } + for _, c := range m.Content { + if c.Type == model.ContentTypeText && strings.TrimSpace(c.Text) == "" { + t.Fatalf("纯附件消息里塞了空文本部件：%+v", m.Content) + } + } + } +} diff --git a/internal/services/gateway/chat_documents.go b/internal/services/gateway/chat_documents.go index b70eeb5..24a6b42 100644 --- a/internal/services/gateway/chat_documents.go +++ b/internal/services/gateway/chat_documents.go @@ -22,11 +22,14 @@ import ( "strings" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) // maxChatDocuments 是**单回合**附件数上限（design/24 §6.1、§9.4-3 的定稿值 5）。 -const maxChatDocuments = 5 +// 取值取自共享存储核心（切片 M1）：Web 上传的单回合上限与 IM 入站附件的单条 +// 上限因此同源，不可能一处改宽、另一处照旧。 +const maxChatDocuments = documents.MaxPerMessage // tooManyDocumentsCode 是附件数超限的机器可读错误码（design/24 §6.2 表格）。 // 它与 image 侧的 image_too_large 同一风格：客户端只看 code 就能分开处理。 diff --git a/internal/services/gateway/documents.go b/internal/services/gateway/documents.go index 073c36b..b5b6349 100644 --- a/internal/services/gateway/documents.go +++ b/internal/services/gateway/documents.go @@ -16,9 +16,6 @@ package gateway // 元数据表 `documents` 只写不回收（引用计数与 GC 属切片 S7）。 import ( - "context" - "crypto/sha256" - "encoding/hex" "errors" "fmt" "io" @@ -31,17 +28,19 @@ import ( "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" - "git.yeij.top/AskaEth/Cyrene_v2/internal/db" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" ) const ( - // maxDocumentBytes 是 `POST /api/v1/documents` 的**单文件字节**默认上限：20 MiB。 + // maxDocumentBytes 是 `POST /api/v1/documents` 的**单文件字节**默认上限： + // 它就是共享存储核心的 `documents.DefaultMaxBytes`（20 MiB）—— 全仓只定义 + // 一次，网关与 IM 入站附件因此不可能各写一个数（design/24 §3.3）。 // 它与图片上限（`maxUploadBytes` / `CHAT_IMAGE_MAX_BYTES`）**刻意解耦**： // 图片上限被 vision 解码层钳住，文档上限由本项目的解析与上下文预算决定； // 共用同一个数会让「改文档上限」静默改掉图片行为（design/24 §3.3）。 - maxDocumentBytes = 20 << 20 + maxDocumentBytes = documents.DefaultMaxBytes // documentMultipartSlack 是 multipart 信封（边界、部件头）在文件上限之上 // 预留的字节余量；文件本身的校验按确切上限做。 @@ -53,7 +52,8 @@ const ( // documentHashHexLen 是内容寻址落盘名使用的 sha256 十六进制前缀长度 // （32 位十六进制 = 128 位），与 `validUploadName` 的形态校验同宽。 - documentHashHexLen = 32 + // 它取自共享存储核心，与落盘/读侧的名字形态同源（切片 M1）。 + documentHashHexLen = documents.HashHexLen // documentFormField 是上传 multipart 里的文件字段名（design/24 §6.1）。 documentFormField = "file" @@ -75,8 +75,8 @@ const ( // documentSniffOKMIME 是唯一放行的嗅探结果基类型。白名单里的每一种文档 // （文本、代码、CSV、JSON、YAML…）经 `http.DetectContentType` 嗅探后都落在 // `text/plain`；HTML/XML/SVG/压缩包/可执行文件/判不定者各有自己的签名， - // 因此「不是 text/plain」即「拒绝」。 - documentSniffOKMIME = "text/plain" + // 因此「不是 text/plain」即「拒绝」。取值取自共享存储核心（同一裁决）。 + documentSniffOKMIME = documents.SniffOKMIME // documentFallbackMIME 是读侧嗅探不出放行类型时下发的媒体类型。配合 // attachment + nosniff，它保证浏览器永远不会按类型渲染文档。 @@ -91,10 +91,6 @@ const ( documentUploadFailedCode = "document_upload_failed" ) -// ErrDocumentTooLarge 报告文档体量超过生效上限。`storeDocument` 用它把 -// 「超限（413）」与「存储失败（500）」分开，调用方用 `errors.Is` 判定。 -var ErrDocumentTooLarge = errors.New("gateway: document too large") - // documentResponse 是上传成功的载荷。URL 是**相对**同源路径（design/24 §6.1）： // 读侧要求 Bearer，因此该 URL 不是「贴进 `` 就能取」的能力 URL，绝对化 // 没有意义；客户端真正要用的是 `chat.documents` 里的 `name`。 @@ -107,23 +103,6 @@ type documentResponse struct { Deduped bool `json:"deduped"` } -// documentRecord 是写入 `documents` 表的一行元数据（S1 只写）。 -type documentRecord struct { - Name string - SHA256 string - MIME string - Size int64 - Uploader string -} - -// documentStored 是一次成功落盘的结果。 -type documentStored struct { - name string - sha256 string - size int64 - deduped bool -} - // documentLimitBytes 解析本次生效的单文件字节上限：配置给了正数就用它， // 否则（未配置的零值、0、负数）回落到默认 20 MiB —— 语义是「≤0 = 用默认」 // 而不是「禁用上限」，与图片侧 `chatImageLimitBytes` 的口径一致。 @@ -138,23 +117,18 @@ func documentLimitBytes(configured int) int { // documentExtOf 从客户端原始文件名取出小写扩展名（含点）。它只作参考： // 结果必须同时通过扩展名白名单与字节嗅探，客户端因此无法靠改扩展名换类型。 +// 取值逻辑与共享存储核心同源（切片 M1），读侧不再各写一套。 func documentExtOf(filename string) string { - return strings.ToLower(filepath.Ext(filename)) + return documents.ExtOf(filename) } // documentExtAllowed 报告扩展名是否在放行集合内。deny-by-default：集合之外的 // 一切（`.html`/`.svg`/`.xml`/`.zip`/`.docx`/`.xlsx`/`.exe`/`.dll`/`.bat`/ // `.cmd`/`.ps1`/`.js`/`.mjs`/`.sh`/`.py`…）一律拒绝。注意 `.js`/`.py` 即便 // 嗅探成 `text/plain` 也必须拒绝 —— 判定以扩展名与嗅探的**合取**为准。 +// 白名单本体只存在于 `documents.ExtAllowed`（22 项，design/24 §9.2-2 定稿）。 func documentExtAllowed(ext string) bool { - switch ext { - case ".txt", ".md", ".markdown", ".csv", ".tsv", ".json", ".jsonl", - ".yaml", ".yml", ".log", ".ini", ".cfg", ".toml", - ".go", ".ts", ".tsx", ".java", ".c", ".h", ".cpp", ".rs", ".sql": - return true - default: - return false - } + return documents.ExtAllowed(ext) } // documentBaseMIME 把嗅探结果归一为不含参数的基类型（`text/plain; charset=utf-8` @@ -169,14 +143,9 @@ func documentBaseMIME(sniffed string) string { // documentSniffAllowed 报告嗅探所得类型是否落在放行的文本类内。它是「类型裁决 // 的唯一裁决者」在代码里的落点：客户端声明的 `Content-Type` 从不参与决定。 +// 裁决本体只存在于 `documents.SniffAllowed`（写侧与读侧同一份）。 func documentSniffAllowed(sniffed string) bool { - return documentBaseMIME(sniffed) == documentSniffOKMIME -} - -// documentHashName 返回内容寻址的落盘名：sha256 十六进制的前 32 位 + ext。 -// 同一份字节配同一个扩展名必然得到同一个名字，去重因此是「同名」的自然结果。 -func documentHashName(sum []byte, ext string) string { - return hex.EncodeToString(sum)[:documentHashHexLen] + ext + return documents.SniffAllowed(sniffed) } // validDocumentName 报告 name 是否恰好是 32 位小写十六进制主名加白名单文档 @@ -238,10 +207,45 @@ func (s *Server) readDocumentUpload(w http.ResponseWriter, r *http.Request, limi return file, header, true } +// documentStore 按本次生效的目录与上限构造一个文档存储（切片 M1 抽出的共享 +// 核心，见 `internal/runtime/documents`）：网关只是它的一个调用方 —— IM 入站 +// 附件走的是**同一个** Store，两条来源因此不可能漂移。上限来自 +// `CHAT_DOCUMENT_MAX_BYTES`（经 documentLimitBytes 归一），并且**每个请求** +// 按当时的配置取值，面板改配置后下一个请求即生效。 +// +// 元数据写入面由 `MetadataWriterFor` 给出：未接库（`s.dbPath` 为空）时它返回 +// 真正的 nil 接口，Store 因此按「未注入」跳过写表并留痕（S1 的行为）。 +func (s *Server) documentStore(dir string, limit int, uploader string) *documents.Store { + return documents.NewStore(documents.StoreConfig{ + UploadsDir: dir, + MaxBytes: limit, + Uploader: uploader, + Meta: documents.MetadataWriterFor(s.dbPath, s.log), + Log: s.log, + }) +} + +// readDocumentBytes 读出本次上传的文件字节，最多 `limit+1` 字节：多读一个字节 +// 就足以判定超限，因此既不把超限内容放大进内存，也不把它留在盘上。S1 的流式 +// 落盘在抽成共享存储后改为「先读进内存再交给 store」——上限本身（默认 20 MiB） +// 没有放宽，而文档抽取路径本来就整份读进内存（design/24 §9.6 已登记的同一代价）。 +func readDocumentBytes(file io.Reader, limit int) ([]byte, error) { + data, err := io.ReadAll(io.LimitReader(file, int64(limit)+1)) + if err != nil { + return nil, fmt.Errorf("documents: 读上传内容: %w", err) + } + return data, nil +} + // handleDocumentUpload 接受带单个 `file` 部件的 multipart/form-data 请求， // 用字节嗅探裁决类型，按内容寻址存到 `/uploads/documents`，写一行 // 元数据并审计 `document.upload`。首次上传回 201 + `deduped:false`；命中同一 // 内容时回 200 + `deduped:true` 且 `name` 与首次逐字一致。 +// +// 类型/体积的裁决**全部**由 `documents.Store.Save` 给出（切片 M1 起与 IM 入站 +// 附件同源）；这里只把它的哨兵错误映射为 S1 既有的契约错误码，判定顺序 +// （扩展名 → 空 → 上限 → 嗅探）由 store 保持，因此线上答案与抽共享核心之前 +// 逐字一致。 func (s *Server) handleDocumentUpload(w http.ResponseWriter, r *http.Request) { limit := documentLimitBytes(s.cfg.Server.ChatDocumentMaxBytes) file, header, ok := s.readDocumentUpload(w, r, limit) @@ -250,139 +254,66 @@ func (s *Server) handleDocumentUpload(w http.ResponseWriter, r *http.Request) { } defer func() { _ = file.Close() }() - ext := documentExtOf(header.Filename) - if !documentExtAllowed(ext) { + // 扩展名先于读体判定：明显不该收的扩展名不必先把请求体读进内存（S1 的 + // 顺位也是先看扩展名）。判定用的是**共享存储核心的同一份白名单**， + // 与 store 内部的裁决不可能分叉；store 仍会再判一次（防御性）。 + if !documents.ExtAllowed(documents.ExtOf(header.Filename)) { s.failDocumentUpload(w, r, documentUnsupportedError()) return } - if header.Size <= 0 { - s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "文件为空")) - return - } - if header.Size > int64(limit) { - s.failDocumentUpload(w, r, documentTooLargeError(limit)) - return - } - sniffed, err := sniffImageMIME(file) + data, err := readDocumentBytes(file, limit) if err != nil { - s.documentsLog().Error("document sniff", "err", err) + s.documentsLog().Error("document read", "err", err) s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "读取上传内容失败")) return } - if !documentSniffAllowed(sniffed) { - s.failDocumentUpload(w, r, documentUnsupportedError()) - return - } dir, err := s.documentsDir() if err != nil { s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "文档存储目录未配置")) return } - stored, err := s.storeDocument(dir, ext, file, int64(limit)) - if errors.Is(err, ErrDocumentTooLarge) { - s.failDocumentUpload(w, r, documentTooLargeError(limit)) - return - } + ref, err := s.documentStore(dir, limit, documentUploader(r)).Save(r.Context(), header.Filename, data) if err != nil { - s.documentsLog().Error("document store", "err", err) - s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "保存文档失败")) - return - } - rec := documentRecord{ - Name: stored.name, - SHA256: stored.sha256, - MIME: documentBaseMIME(sniffed), - Size: stored.size, - Uploader: documentUploader(r), - } - if err := s.insertDocumentMetadata(r.Context(), rec); err != nil { - s.documentsLog().Error("document metadata", "err", err, "name", rec.Name) - s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "记录文档元数据失败")) + s.failDocumentSave(w, r, limit, ref, err) return } s.documentsLog().Info("document stored", - "name", rec.Name, "size", rec.Size, "mime", rec.MIME, - "sha256", rec.SHA256, "deduped", stored.deduped) + "name", ref.Name, "size", ref.Size, "mime", ref.MIME, + "sha256", ref.SHA256, "deduped", ref.Deduped) s.auditClaims(r, documentUploadAction, "ok") status := http.StatusCreated - if stored.deduped { + if ref.Deduped { status = http.StatusOK } writeJSON(w, status, documentResponse{ - Name: rec.Name, - URL: documentFilePrefix + rec.Name, - MIME: rec.MIME, - Size: rec.Size, - SHA256: rec.SHA256, - Deduped: stored.deduped, + Name: ref.Name, + URL: documentFilePrefix + ref.Name, + MIME: ref.MIME, + Size: ref.Size, + SHA256: ref.SHA256, + Deduped: ref.Deduped, }) } -// storeDocument 把 f 按内容寻址写入 dir：先流式写点前缀临时文件、同时计算 -// sha256，再把临时文件重命名到 ``（部分写入因此绝不会以可读的 -// 名字露面）。目标名已存在（同字节同扩展名）时丢弃临时文件并报告 deduped， -// 盘上只留一份。目录 0700、文件继承临时的 0600。 -func (s *Server) storeDocument(dir, ext string, f io.Reader, limit int64) (documentStored, error) { - if err := os.MkdirAll(dir, 0o700); err != nil { - return documentStored{}, fmt.Errorf("create document dir: %w", err) +// failDocumentSave 把 store 的哨兵错误映射为 S1 既有的线上错误并审计。映射表 +// 逐条对应（415 类型拒绝 / 413 超限 / 400 空文件 / 500 存储或元数据失败）， +// 与抽共享核心之前的对外行为一字不差：拒绝是**可预期的输入问题**，因此不额外 +// 记错误日志（与 S1 相同）；存储与元数据失败才留日志。 +func (s *Server) failDocumentSave(w http.ResponseWriter, r *http.Request, limit int, ref documents.Ref, err error) { + switch { + case errors.Is(err, documents.ErrTooLarge): + s.failDocumentUpload(w, r, documentTooLargeError(limit)) + case errors.Is(err, documents.ErrExtNotAllowed), errors.Is(err, documents.ErrContentNotText): + s.failDocumentUpload(w, r, documentUnsupportedError()) + case errors.Is(err, documents.ErrEmpty): + s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "文件为空")) + case errors.Is(err, documents.ErrMetadata): + s.documentsLog().Error("document metadata", "err", err, "name", ref.Name) + s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "记录文档元数据失败")) + default: + s.documentsLog().Error("document store", "err", err) + s.failDocumentUpload(w, r, apperr.Wrap(apperr.ErrInternal, documentUploadFailedCode, "保存文档失败")) } - tmp, err := os.CreateTemp(dir, ".document-*") - if err != nil { - return documentStored{}, fmt.Errorf("create document temp: %w", err) - } - tmpName := tmp.Name() - hasher := sha256.New() - written, err := io.Copy(io.MultiWriter(tmp, hasher), io.LimitReader(f, limit+1)) - if err != nil { - _ = tmp.Close() - _ = os.Remove(tmpName) - return documentStored{}, fmt.Errorf("write document: %w", err) - } - if err := tmp.Close(); err != nil { - _ = os.Remove(tmpName) - return documentStored{}, fmt.Errorf("close document: %w", err) - } - // 文件级上限的第二次把关（第一道是 multipart 部件大小）：多读一个字节即可 - // 判定超限，因此既不放大内存也不把超限文件留在盘上。 - if written > limit { - _ = os.Remove(tmpName) - return documentStored{}, ErrDocumentTooLarge - } - digest := hasher.Sum(nil) - sum := hex.EncodeToString(digest) - name := documentHashName(digest, ext) - if info, serr := os.Stat(filepath.Join(dir, name)); serr == nil && !info.IsDir() { - _ = os.Remove(tmpName) - return documentStored{name: name, sha256: sum, size: written, deduped: true}, nil - } - if err := os.Rename(tmpName, filepath.Join(dir, name)); err != nil { - _ = os.Remove(tmpName) - return documentStored{}, fmt.Errorf("commit document: %w", err) - } - return documentStored{name: name, sha256: sum, size: written, deduped: false}, nil -} - -// insertDocumentMetadata 把一行文档元数据写进 `documents` 表（迁移 0014）。 -// 同一内容寻址名重复上传时 `DO NOTHING`：首次的 uploader/created_at 是权威， -// 去重不刷新它们。`s.dbPath` 未注入（未接数据库的构建/测试）时跳过 —— 上传 -// 本身仍然成功，元数据留待 S7 的引用计数与 GC 统一接线。 -func (s *Server) insertDocumentMetadata(ctx context.Context, rec documentRecord) error { - path := strings.TrimSpace(s.dbPath) - if path == "" { - return nil - } - conn, err := db.Open(ctx, path, s.log) - if err != nil { - return fmt.Errorf("open db for document metadata: %w", err) - } - defer func() { _ = conn.Close() }() - const query = `INSERT INTO documents (name, sha256, mime, size, uploader) -VALUES (?, ?, ?, ?, ?) -ON CONFLICT(name) DO NOTHING` - if _, err := conn.ExecContext(ctx, query, rec.Name, rec.SHA256, rec.MIME, rec.Size, rec.Uploader); err != nil { - return fmt.Errorf("insert document metadata: %w", err) - } - return nil } // documentUploader 从请求上下文取上传者 id（由 auth.Authorize 填充），取不到时