feat(runtime): 文档摄取①(抽取器 + sidecar + 装配期 resolveDocuments)(design/24 S3-A)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

- 抽取器零新依赖纯函数:UTF-8 清洗、CSV/TSV 行数列名摘要、64K rune 安全截断;PDF/Office/二进制走哨兵错误诚实拒绝(不引解析库)。

- sidecar:<DataDir>/extracted/<存储名>.txt,首行紧凑 JSON 元数据 + 正文;临时文件 + rename 原子落定;只抽一次(删原文件仍出正文);版本不匹配/坏缓存重抽。

- 装配期 resolveDocuments(resolveImages 同位,每步对模型可见视图注入):清单恒在(解析器缺席/报错也在)、截断带续读游标、失败诚实标记 + status=unreadable、回合继续;文档部件永不抵达编码器(负例钉住)。

- 修 S2 移交的两处静默吞掉:checkpoint 比较键纳入文档引用、compactor token 估算计入文档部件(不是免费的)。

- 主会话独立验收:隔离 worktree(HEAD + 本切片)build/vet 绿、全树 59 包 0 FAIL、linux/amd64 + darwin/arm64 交叉编译绿、documents+loop 两包 167 条 PASS;变异独立复现(resolveDocuments 短路 ⇒ 3 条红,含「文档部件仍抵达下游」)。
This commit is contained in:
AskaEth committed 2026-10-04 17:22:05 +08:00
1 parent 834ef3683e
commit a8851dc0a2
12 files changed
+2008 -7

No files matched your search

+466
View File
@@ -0,0 +1,466 @@
package documents
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"os"
"path/filepath"
"strings"
"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/model"
)
// 本文件归属抽取 sidecar 与摄取层的装配期接缝(design/24 §4.3、§4.4):
// 抽取结果落 `<DataDir>/extracted/<存储名>.txt`,**只抽一次**,后续回合读
// sidecar 而不再解析原文件(Open WebUI 的 re-index 语义;同时避开 H1 的
// 「每回合重算」)。
//
// sidecar 的形状(本切片的取舍):**单个文件**,第一行是紧凑 JSON 元数据
// (chars/truncated/kind/status/reason/抽取器版本),其后是抽取正文。选单文件
// 而不是「正文 + 同目录 .meta」两个文件,是因为「只抽一次」需要一个**原子**的
// 存在性判据:一个文件 = 一次 rename 就落定,不会留下「正文写了、元数据没写」
// 的半截状态。元数据行永远以 `{` 开头,读侧因此能一眼区分元数据与正文。
//
// 文件名规则是「存储名 + .txt」(`abc.csv` → `abc.csv.txt`),而不是「32 位
// 哈希 + .txt」:抽取结论依赖扩展名(只有 .csv/.tsv 才有行数/列名摘要),
// 同一份字节以 .txt 与 .csv 两个存储名上传时抽取结果并不相同,只按哈希做键
// 会把两者混成一条缓存。
const (
// sidecarDirName 是 sidecar 子目录名:`<DataDir>/extracted`。
sidecarDirName = "extracted"
// sidecarSuffix 是 sidecar 文件名的后缀。
sidecarSuffix = ".txt"
// sidecarHeaderEnd 分隔元数据行与正文。
sidecarHeaderEnd = "\n"
// sidecarDirMode / sidecarFileMode 与 S1 的落盘权限一致(目录 0700、
// 文件 0600):抽取正文与原文同等敏感。
sidecarDirMode = 0o700
sidecarFileMode = 0o600
// maxSidecarBytes 是读 sidecar 时的硬上限(正文预算 64K 字符,UTF-8 最多
// 4 字节/字符,再留元数据余量)。超过即判为损坏并重抽,绝不把一整份畸形
// 文件读进内存。
maxSidecarBytes = MaxExtractChars*4 + 1<<10
// sidecarTempPattern 是原子写入用的临时文件前缀。
sidecarTempPattern = ".sidecar-*"
// storageHashHexLen 是 S1 内容寻址主名的十六进制位数(32 位 = 128 位)。
storageHashHexLen = 32
// storageExtMaxLen 是存储名扩展名长度的上限(最长的是 .markdown)。
storageExtMaxLen = 12
// documentsLogSource 是摄取层的日志来源标签。日志只记 name/size/mime/
// kind/chars/truncated,**绝不记正文**(design/24 §3.2 的日志口径)。
documentsLogSource = "runtime/documents"
)
// 抽取失败时注入模型的诚实原因文案(design/24 §4.4)。它们会被模型看到,
// 因此必须是「人能读懂、不含路径与内部细节」的短句;具体错误只进日志。
const (
reasonNoTextLayer = "无法抽取正文:本期不支持 PDF/Office 解析(无文本层)"
reasonUnsupported = "无法抽取正文:内容不在可抽取的纯文本范围内"
reasonMissing = "无法抽取正文:文档文件不存在或已被清理"
reasonReadFailed = "无法抽取正文:读取文档失败"
reasonNameInvalid = "无法抽取正文:文档引用形态非法"
reasonNoUploadDir = "无法抽取正文:文档上传目录未配置"
)
// ErrInvalidName 报告一个不是 S1 存储名形态的名字被拿去做 sidecar 键或读盘
// 路径。它不该发生(请求边界由网关校验),因此是防御性的哨兵错误。
var ErrInvalidName = errors.New("documents: 非法文档存储名")
// cachedExtract 是一条 sidecar 记录:抽取结论(或失败结论)与它的正文/原因。
type cachedExtract struct {
// Status 是抽取结论,取值见 history.DocumentStatus*。
Status string
// Reason 是 Status == unreadable 时注入模型的诚实原因文案。
Reason string
// Kind 是抽取所得内容的种类(text / csv / tsv)。
Kind string
// Chars 是注入正文的字符数。
Chars int
// Truncated 报告正文是否被预算截断。
Truncated bool
// Text 是 Status == extracted 时的注入正文。
Text string
}
// sidecarMeta 是 sidecar 第一行的元数据。它只承载**统计与结论**,正文在
// 第二行之后(design/24 §2.4:历史与引用里绝不出现正文,但 sidecar 就是
// 抽取正文本身,那是它的全部意义)。
type sidecarMeta struct {
Version string `json:"v"`
Status string `json:"status"`
Reason string `json:"reason,omitempty"`
Kind string `json:"kind,omitempty"`
Chars int `json:"chars,omitempty"`
Truncated bool `json:"truncated,omitempty"`
Name string `json:"name,omitempty"`
Size int64 `json:"size,omitempty"`
}
// sidecarStore 是 `<DataDir>/extracted` 下的抽取缓存。dataDir 为空时它是 nil
// (缓存关闭,每次装配重抽),因此每个方法都容忍 nil 接收者。
type sidecarStore struct {
dir string
log *slog.Logger
}
// newSidecarStore 构造 sidecar 存储;dataDir 为空返回 nil(缓存关闭)。
func newSidecarStore(dataDir string, log *slog.Logger) *sidecarStore {
if strings.TrimSpace(dataDir) == "" {
return nil
}
return &sidecarStore{dir: filepath.Join(dataDir, sidecarDirName), log: log}
}
// pathFor 返回一份文档的 sidecar 路径;名字形态非法时返回 false。它做两道
// 检查:名字必须是单个路径段(无分隔符、无 `..`、非绝对路径),拼出的路径
// 必须仍在 sidecar 目录内(与 S1 读侧的双检同一手法)。
func (s *sidecarStore) pathFor(name string) (string, bool) {
if s == nil || !validStorageName(name) {
return "", false
}
path := filepath.Join(s.dir, name+sidecarSuffix)
if filepath.Dir(path) != filepath.Clean(s.dir) {
return "", false
}
return path, true
}
// load 读一条 sidecar 记录。不存在、读不出、形态损坏、抽取器版本不匹配都
// 返回 false(调用方据此重抽一次),并留下带来源的日志 —— 缓存失效是正常
// 路径,但绝不静默吞掉「文件在那儿却读不出来」这类信号。
func (s *sidecarStore) load(name string) (cachedExtract, bool) {
if s == nil {
return cachedExtract{}, false
}
path, ok := s.pathFor(name)
if !ok {
return cachedExtract{}, false
}
f, err := os.Open(path)
if err != nil {
if !errors.Is(err, os.ErrNotExist) {
s.log.Warn("sidecar open", "name", name, "error", err)
}
return cachedExtract{}, false
}
defer func() { _ = f.Close() }()
data, err := io.ReadAll(io.LimitReader(f, maxSidecarBytes+1))
if err != nil || len(data) > maxSidecarBytes {
s.log.Warn("sidecar unreadable", "name", name, "bytes", len(data), "error", err)
return cachedExtract{}, false
}
header, body, ok := strings.Cut(string(data), sidecarHeaderEnd)
if !ok || !strings.HasPrefix(header, "{") {
s.log.Warn("sidecar malformed", "name", name)
return cachedExtract{}, false
}
meta := sidecarMeta{}
if err := json.Unmarshal([]byte(header), &meta); err != nil {
s.log.Warn("sidecar meta decode", "name", name, "error", err)
return cachedExtract{}, false
}
if meta.Version != ExtractorVersion {
// 抽取器升级:旧缓存按失效处理并重抽一次(记版本号的意义所在)。
s.log.Info("sidecar stale", "name", name, "version", meta.Version)
return cachedExtract{}, false
}
return cachedExtract{
Status: meta.Status,
Reason: meta.Reason,
Kind: meta.Kind,
Chars: meta.Chars,
Truncated: meta.Truncated,
Text: body,
}, true
}
// save 把一条结论原子地写进 sidecar。它**只写一次**:已存在且版本匹配的记录
// 一律不覆盖(连 mtime 都不动),因此「后续回合不再碰原文件」在文件系统上是
// 可观测的。写入用「临时文件 + rename」,绝不留下半截 sidecar。
func (s *sidecarStore) save(name string, rec cachedExtract, size int64) error {
if s == nil {
return nil
}
path, ok := s.pathFor(name)
if !ok {
return fmt.Errorf("documents: sidecar %q: %w", name, ErrInvalidName)
}
if _, exists := s.load(name); exists {
return nil
}
if err := os.MkdirAll(s.dir, sidecarDirMode); err != nil {
return fmt.Errorf("documents: sidecar dir: %w", err)
}
header, err := json.Marshal(sidecarMeta{
Version: ExtractorVersion,
Status: rec.Status,
Reason: rec.Reason,
Kind: rec.Kind,
Chars: rec.Chars,
Truncated: rec.Truncated,
Name: name,
Size: size,
})
if err != nil {
return fmt.Errorf("documents: sidecar meta encode: %w", err)
}
tmp, err := os.CreateTemp(s.dir, sidecarTempPattern)
if err != nil {
return fmt.Errorf("documents: sidecar temp: %w", err)
}
tmpName := tmp.Name()
if _, err := tmp.WriteString(string(header) + sidecarHeaderEnd + rec.Text); err != nil {
_ = tmp.Close()
_ = os.Remove(tmpName)
return fmt.Errorf("documents: sidecar write: %w", err)
}
if err := tmp.Chmod(sidecarFileMode); err != nil {
_ = tmp.Close()
_ = os.Remove(tmpName)
return fmt.Errorf("documents: sidecar chmod: %w", err)
}
if err := tmp.Close(); err != nil {
_ = os.Remove(tmpName)
return fmt.Errorf("documents: sidecar close: %w", err)
}
if err := os.Rename(tmpName, path); err != nil {
_ = os.Remove(tmpName)
return fmt.Errorf("documents: sidecar commit: %w", err)
}
return nil
}
// validStorageName 报告 name 是不是 S1 的合法文档存储名形态:32 位小写十六
// 进制主名 + 一个小写字母数字扩展名。它**不**重复 S1 的扩展名白名单(那是
// 存储侧的裁决),只保证这个名字不会把路径带出目录、也不会把 sidecar 写到
// 别处 —— 名字可能来自数据库里的历史引用,因此这里不假设它已被校验过。
func validStorageName(name string) bool {
if name == "" || name != filepath.Base(name) || strings.ContainsAny(name, `/\`) {
return false
}
stem, ext, ok := strings.Cut(name, ".")
if !ok || len(stem) != storageHashHexLen || ext == "" || len(ext) > storageExtMaxLen {
return false
}
for _, c := range stem {
if !isLowerHex(c) {
return false
}
}
for _, c := range ext {
if !isLowerAlphaNum(c) {
return false
}
}
return true
}
// isLowerHex 报告 c 是不是小写十六进制数字。
func isLowerHex(c rune) bool {
return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f')
}
// isLowerAlphaNum 报告 c 是不是小写字母或数字。
func isLowerAlphaNum(c rune) bool {
return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'z')
}
// Config 是摄取层的接线参数。路径口径由组合根给出,本包不猜数据目录布局。
type Config struct {
// DataDir 是数据目录;sidecar 落在它的 extracted/ 子目录下。
DataDir string
// UploadsDir 是 S1 的文档落盘目录(`<DataDir>/uploads/documents`)。它与
// gateway/documents.go 的 documentSubDir 同源:改其一时两处必须同步,
// 否则装配期会去读一个不存在的目录(症状是「附件全都读不到」)。
UploadsDir string
// Log 是基础 logger;为 nil 时替换为标记来源 runtime/documents 的丢弃 logger。
Log *slog.Logger
}
// Resolver 是摄取层的装配期实现。它**结构上**满足 loop.DocumentResolver
// (不依赖 loop 包,与 vision.MessageResolver 满足 loop.ImageResolver 的方式
// 相同),因此循环也不必依赖本包。
type Resolver struct {
uploadsDir string
store *sidecarStore
log *slog.Logger
}
// NewResolver 构造摄取层解析器。路径缺失不会让构造失败:缓存目录为空时退化为
// 「每次重抽」,上传目录为空时每份文档都得到诚实的 unreadable 结论(而不是
// 假装读到),两种情况都留一条带来源的 WARN。
func NewResolver(cfg Config) *Resolver {
log := sourcedLog(cfg.Log)
r := &Resolver{
uploadsDir: strings.TrimSpace(cfg.UploadsDir),
store: newSidecarStore(cfg.DataDir, log),
log: log,
}
if r.store == nil {
log.Warn("document sidecar disabled: data dir not configured")
}
if r.uploadsDir == "" {
log.Warn("document uploads dir not configured: extractions will be unreadable")
}
return r
}
// sourcedLog 返回标记模块来源 "runtime/documents" 的 logger,把 nil logger
// 替换为丢弃 logger,使摄取层总能安全地记录日志。
func sourcedLog(l *slog.Logger) *slog.Logger {
if l == nil {
l = slog.New(slog.NewTextHandler(io.Discard, nil))
}
return logger.WithSource(l, documentsLogSource)
}
// Resolve 解析 msg 里出现的所有文档部件,返回两份以**服务端存储名**为键的
// 映射:结论(形状即 history.DocumentRef —— 与落库/DTO 同一个结构体)与要注入
// 模型可见上下文的正文。
//
// 与 loop.ImageResolver「就地改写一条消息」同族,但回的是两份映射而不是一条
// 消息:清单行、截断标记与「清单恒在」必须由循环**统一**渲染(未接线解析器
// 或解析失败时也要成立),而 status/chars/truncated 要经循环的 Result 回给
// 客户端(design/24 §4.4/§6.1)。用中立类型(string + history.DocumentRef)
// 让摄取层不必依赖 loop 包。
//
// 单份文档的任何失败(不存在、读不出、类型不支持)都**不是** error:它是
// unreadable 结论 + 诚实的原因文案(§4.4「失败必须可见」)。error 只表示整条
// 消息都解析不了(例如回合已被取消),由循环降级为诚实的兜底渲染。
func (r *Resolver) Resolve(ctx context.Context, msg model.Message) (map[string]history.DocumentRef, map[string]string, error) {
if err := ctx.Err(); err != nil {
return nil, nil, fmt.Errorf("documents: resolve: %w", err)
}
parts := documentPartsOf(msg)
if len(parts) == 0 {
return nil, nil, nil
}
refs := make(map[string]history.DocumentRef, len(parts))
bodies := make(map[string]string, len(parts))
for _, part := range parts {
if _, done := refs[part.DocumentName]; done {
// 同一条消息里对同一份文档的重复引用只解析一次。
continue
}
ref, body := r.resolveOne(part)
refs[part.DocumentName] = ref
bodies[part.DocumentName] = body
}
return refs, bodies, nil
}
// documentPartsOf 返回 msg 里的文档部件(按出现顺序)。
func documentPartsOf(msg model.Message) []model.Content {
out := make([]model.Content, 0, len(msg.Content))
for _, part := range msg.Content {
if part.Type == model.ContentTypeDocument {
out = append(out, part)
}
}
return out
}
// resolveOne 解析一份文档部件:先读 sidecar(design/24 §4.3「只抽一次」),
// 未命中才读原文件并抽取。任何一种失败都折成 unreadable 结论 + 诚实原因,
// 绝不会让整条消息失败。
func (r *Resolver) resolveOne(part model.Content) (history.DocumentRef, string) {
ref := history.DocumentRef{Name: part.DocumentName, MIME: part.DocumentMIME, Size: part.DocumentSize}
if !validStorageName(part.DocumentName) {
// 形态非法(请求边界本已由网关校验)→ 绝不按它读盘,但也不静默。
r.log.Warn("document name rejected", "name", part.DocumentName)
ref.Status = history.DocumentStatusUnreadable
return ref, reasonNameInvalid
}
if rec, ok := r.store.load(part.DocumentName); ok {
r.log.Debug("document sidecar hit", "name", part.DocumentName, "status", rec.Status)
return documentRefFrom(rec, ref), recordBody(rec)
}
rec := r.extract(part)
if err := r.store.save(part.DocumentName, rec, part.DocumentSize); err != nil {
// 写 sidecar 失败只意味着下回合要重抽,绝不影响本回合的注入。
r.log.Warn("sidecar save", "name", part.DocumentName, "error", err)
}
return documentRefFrom(rec, ref), recordBody(rec)
}
// extract 读原文件并抽取,把任何失败折成 unreadable 记录(含诚实原因)。
// 失败同样会被写进 sidecar:20 MiB 的 PDF 不该每回合重读一遍。
func (r *Resolver) extract(part model.Content) cachedExtract {
if r.uploadsDir == "" {
r.log.Warn("document extraction skipped: uploads dir not configured", "name", part.DocumentName)
return unreadable(reasonNoUploadDir)
}
path := filepath.Join(r.uploadsDir, part.DocumentName)
data, err := os.ReadFile(path)
if err != nil {
r.log.Warn("document read", "name", part.DocumentName, "size", part.DocumentSize, "error", err)
if errors.Is(err, os.ErrNotExist) {
return unreadable(reasonMissing)
}
return unreadable(reasonReadFailed)
}
res, err := Extract(part.DocumentName, part.DocumentMIME, data)
if err != nil {
// 抽取器只报哨兵错误;日志记具体错误与名字/大小,**绝不记正文**。
r.log.Info("document extraction refused", "name", part.DocumentName, "size", len(data), "error", err)
return unreadable(reasonForExtractError(err))
}
r.log.Info("document extracted", "name", part.DocumentName, "size", len(data),
"kind", res.Kind, "chars", res.Chars, "truncated", res.Truncated)
return cachedExtract{
Status: history.DocumentStatusExtracted,
Kind: res.Kind,
Chars: res.Chars,
Truncated: res.Truncated,
Text: res.Text,
}
}
// documentRefFrom 把 sidecar 记录折进文档引用:身份(存储名、MIME、字节数)
// 来自部件本身,status/chars/truncated 来自抽取结论。空状态归为 listed
// (「清单已给出、正文未注入」是事实,而不是「已抽取」)。
func documentRefFrom(rec cachedExtract, ref history.DocumentRef) history.DocumentRef {
ref.Status = rec.Status
ref.Chars = rec.Chars
ref.Truncated = rec.Truncated
if ref.Status == "" {
ref.Status = history.DocumentStatusListed
ref.Chars, ref.Truncated = 0, false
}
return ref
}
// recordBody 返回要注入的正文:抽到正文就是正文,读不到就是诚实的原因。
func recordBody(rec cachedExtract) string {
if rec.Status == history.DocumentStatusUnreadable {
return rec.Reason
}
return rec.Text
}
// unreadable 构造一条「读不到正文」的记录。
func unreadable(reason string) cachedExtract {
return cachedExtract{Status: history.DocumentStatusUnreadable, Reason: reason}
}
// reasonForExtractError 把抽取器的哨兵错误映射为注入模型的诚实原因。
func reasonForExtractError(err error) string {
if errors.Is(err, ErrNoTextLayer) {
return reasonNoTextLayer
}
return reasonUnsupported
}