diff --git a/internal/runtime/embedding/api.go b/internal/runtime/embedding/api.go index 92cda5f..9986e7b 100644 --- a/internal/runtime/embedding/api.go +++ b/internal/runtime/embedding/api.go @@ -7,10 +7,10 @@ import ( "errors" "fmt" "io" - "log/slog" "net/http" "strings" "sync" + "sync/atomic" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" @@ -38,23 +38,24 @@ const ( // DefaultRequestTimeout 是**单次** /embeddings 调用的墙钟上限，独立于 HTTP 客户端 // 那一层的 180s 总超时（httpRequestTimeout）。 // -// 依据（是实测数据，不是拍脑袋的调优）： -// - 上层「入库前审查」的单条总预算是 20s（memory.defaultReviewTimeout），embedding -// 只是它内部的第一步。2026-10-07 实测（本机 llama.cpp + Qwen3-Embedding-0.6B， -// 服务空闲）：单条文本约 60ms；**64 条一批约 2.1~2.9s**；审查最坏情况 -// （ScanLimit=200 + 新条目 1）要 4 批 ⇒ 约 8~10s，已经占掉 20s 预算的一半， -// 留给后面的 T1 灰区判定（defaultT1Timeout=8s）的余量很薄。一次卡死的 HTTP -// 就能把这点余量连同整条审查一起吃掉（实测日志里 -// `memory/review: context deadline exceeded` 正是这个形态）； -// - 所以每次调用必须有**自己的**上限：卡住的那一批到点就失败并降级，不能把整条 -// 审查的 20s 都赌在一次 HTTP 上； -// - 5s 的取值：相对实测的 64 条一批 2.9s 留约 1.7 倍余量，能吸收并发争用带来的 -// 抖动，同时远小于 20s 的审查预算。若日后批次变大或服务变慢，先调这里而不是 -// 调审查总预算； -// - 真慢就**快速失败**：调用方按**既有语义**降级（审查降级到词项余弦并如实留痕 -// degraded），**绝不静默丢记忆** —— 本常量只砍等待时间，不改任何判定规则； -// - 已知代价：llama.cpp 首次加载模型可能超过 5s，那一次调用会失败并降级，模型 -// 常驻后的下一次调用即恢复正常。降级本就是既有行为，因此可接受。 +// 它**不是**单独取值：批次上限（DefaultBatchLimit / DefaultBatchChars）与它由 +// budget.go 里**同一个不等式**推出 —— 单次请求的耗时上界不得超过超时的一半。 +// 改这一行就等于改了批次上限，反之亦然；两侧永远是配套的。 +// +// 为什么每次调用必须有自己的上限：上层「入库前审查」的单条总预算是 20s +// （memory.defaultReviewTimeout），embedding 只是它内部的第一步，后面还有 T1 灰区 +// 判定（8s）。一次卡死的 HTTP 不能把整条审查的 20s 都赌掉。真慢就**快速失败**， +// 调用方按**既有语义**降级（审查降级到词项余弦并如实留痕 degraded）， +// **绝不静默丢记忆** —— 本常量只砍等待时间，不改任何判定规则。 +// +// 5s 保持不变的依据：本次故障（2026-10-13）的成因不是这个数太小，而是批次太大 —— +// 64 条的实测耗时 7.5~8.5s 已经越过它，客户端放弃、服务端还在算，把整台本机 +// embedding 服务顶住，连带单条请求（写向量索引）也一起超时。所以修法是**把批次 +// 收进预算内**，而不是把预算放大到迁就批次（后者会让真挂死的请求也拖住整轮）。 +// +// 冷启动另有 warmupTimeout（见 budget.go），但它**只放宽本进程的第一个请求**、 +// 且一次性花掉：失败不会让它续期，因此「服务一直挂死」这件事**不会**变成 +// 「每次调用都慢」—— 第二个请求起就回到本常量。 const DefaultRequestTimeout = 5 * time.Second // openAICompatibleRequest is the JSON body sent to an OpenAI-compatible @@ -92,7 +93,11 @@ var ErrNoAPIURL = errors.New("embedding: api url is empty") type APIEmbedder struct { cfg config.EmbeddingConfig client *http.Client - log *slog.Logger + // warmupLeft 是**一次性**冷启动放宽的开关：为真时下一个请求用 warmupTimeout， + // 取走即置假（CompareAndSwap ⇒ 并发下也只有一个请求能拿到）。 + // 它**不是**「未成功就一直放宽」—— 失败也照样花掉这一次，因此服务挂死时 + // 只会慢一次，之后每个请求都按 DefaultRequestTimeout 快速失败（见 budget.go）。 + warmupLeft atomic.Bool mu sync.RWMutex dim int @@ -120,23 +125,20 @@ func WithRoundTripper(rt http.RoundTripper) APIOption { } } -// WithLogger sets the logger used for embed failures. A nil logger is ignored -// and the default discard logger is kept. -func WithLogger(l *slog.Logger) APIOption { - return func(e *APIEmbedder) { - if l != nil { - e.log = l - } - } -} - // NewAPI builds an APIEmbedder from cfg using the default HTTP client. +// +// 本包**不记日志**（没有任何 logger 缝）：失败一律沿错误返回，由四个调用方 +// （审查召回 / 整理召回 / 向量索引写入 / 检索）各自带 source 记账 —— 它们本来 +// 就会把错误原文写进各自的 WARN，因此这里再记一条只会是重复的第二份。 +// 单次超时也不可由选项更换：它是 budget.go 那条推演的输入，批次上限与它同源， +// 因此**只允许改常量**（改了之后批次上限自动跟着改，测试的常量断言立刻复核）。 func NewAPI(cfg config.EmbeddingConfig, opts ...APIOption) *APIEmbedder { e := &APIEmbedder{ cfg: cfg, client: defaultHTTPClient(), - log: slog.New(slog.NewTextHandler(io.Discard, nil)), } + // 冷启动放宽是一个**一次性**许可：见 APIEmbedder.warmupLeft。 + e.warmupLeft.Store(true) for _, opt := range opts { opt(e) } @@ -155,8 +157,10 @@ func (e *APIEmbedder) Dim() int { // returned vectors aligned by index. It returns ErrNoAPIURL for a blank URL // and a wrapped error for any HTTP or decode failure. // -// 单次调用带独立的、比上层审查总预算（20s）短得多的超时（DefaultRequestTimeout）， -// 卡住即快速失败，把预算留给后续步骤；失败语义与降级口径不变。 +// texts 长度不受调用方控制：本方法按 budget.go 推出来的上限（条数 / 字符总量） +// **自己切片**，分多次请求顺序发出，因此「一次请求的耗时 < 单次超时」这条不变式 +// 对任何调用方都成立 —— 调用方给多大的切片都不会把服务端顶住。中途任一段失败即 +// 整体失败（返回已发请求的错误），绝不半途改用混合表示。 func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, error) { if len(texts) == 0 { return []Vector{}, nil @@ -168,7 +172,37 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro return nil, ErrNoAPIURL } - callCtx, cancel := context.WithTimeout(ctx, DefaultRequestTimeout) + // 切片用的上限由**稳态**超时推出（budget.go 的 DefaultRequestTimeout）： + // 冷启动那一次只放宽**截止时刻**，批形状不变 —— 否则首个请求会按更宽的预算 + // 切出一个巨大的批，正好把冷启动的模型顶得更久。 + items, chars := batchLimitsFor(DefaultRequestTimeout) + out := make([]Vector, 0, len(texts)) + for _, chunk := range chunkTexts(texts, items, chars) { + vectors, err := e.embedChunk(ctx, chunk) + if err != nil { + return nil, err + } + out = append(out, vectors...) + } + return out, nil +} + +// callTimeout 返回本次请求该用的单次超时，并**当场花掉**一次性冷启动放宽许可 +// （成功、失败都算花掉）。因此： +// - 本进程的第一个请求最多等 warmupTimeout（冷启动）， +// - 从第二个请求起一律 DefaultRequestTimeout —— 服务挂死时不会每次都慢。 +func (e *APIEmbedder) callTimeout() time.Duration { + if e.warmupLeft.CompareAndSwap(true, false) { + return warmupTimeout + } + return DefaultRequestTimeout +} + +// embedChunk 发**一次** /embeddings 请求。chunk 必须已经落在 batchLimitsFor 推出来 +// 的边界内（Embed 负责切），因此它的耗时上界不超过单次超时的一半。 +func (e *APIEmbedder) embedChunk(ctx context.Context, texts []string) ([]Vector, error) { + // 先取超时（取走即花掉那次放宽），再建 ctx：放宽许可与这次请求一一对应。 + callCtx, cancel := context.WithTimeout(ctx, e.callTimeout()) defer cancel() req, err := e.newEmbeddingRequest(callCtx, texts) diff --git a/internal/runtime/embedding/api_url_test.go b/internal/runtime/embedding/api_url_test.go index 25a78b1..c7b8234 100644 --- a/internal/runtime/embedding/api_url_test.go +++ b/internal/runtime/embedding/api_url_test.go @@ -222,6 +222,11 @@ const reviewBudgetRef = 20 * time.Second // 审查总预算：否则一次卡死的 embedding 连接就能吃光整条审查的预算，排在后面的 // T1 灰区判定连执行机会都没有（2026-10-07 实测故障形态）。同时验证该超时确实 // 打在了发出去的请求上。 +// +// 🔴 冷启动放宽（warmupTimeout）必须一并守住同一条界：它只发生一次，但它那一次的 +// 等待**也**不能吃光整轮审查 —— 否则「一次挂死 = 整轮审查没有预算给 T1」这个形态 +// 会从常态超时挪到放宽预算上原样复现（reviewer round1 的 p1）。因此这里对 +// warmupTimeout 做与 DefaultRequestTimeout **同一条**断言。 func TestDefaultRequestTimeout(t *testing.T) { t.Parallel() @@ -236,6 +241,19 @@ func TestDefaultRequestTimeout(t *testing.T) { t.Fatalf("DefaultRequestTimeout = %s, want 5s（值变了要连带复核审查预算口径）", DefaultRequestTimeout) } + // 冷启动放宽的两条界：仍在常态超时的**同一量级**（不得成倍放大）， + // 且严格小于审查总预算（一次冷启动不许吃光整轮）。 + if warmupTimeout <= 0 { + t.Fatalf("warmupTimeout = %s, 必须为正", warmupTimeout) + } + if warmupTimeout > 2*DefaultRequestTimeout { + t.Fatalf("冷启动放宽 %s 超过常态超时 %s 的两倍 —— 它必须是同一量级的一次性放宽", + warmupTimeout, DefaultRequestTimeout) + } + if warmupTimeout >= reviewBudgetRef { + t.Fatalf("冷启动放宽 %s 必须严格小于审查总预算 %s（否则一次冷启动就吃光整轮审查）", + warmupTimeout, reviewBudgetRef) + } var slack time.Duration rt := roundTripFunc(func(r *http.Request) (*http.Response, error) { @@ -256,8 +274,12 @@ func TestDefaultRequestTimeout(t *testing.T) { config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"}, WithRoundTripper(rt), ) - if _, err := emb.Embed(context.Background(), []string{"hi"}); err != nil { - t.Fatalf("Embed() error: %v", err) + // 第一次是冷启动调用（warmupTimeout），结果由它自己的测试断言；这里连发两次， + // 只看稳态那一次打出去的 deadline。 + for i := 0; i < 2; i++ { + if _, err := emb.Embed(context.Background(), []string{"hi"}); err != nil { + t.Fatalf("第 %d 次 Embed() error: %v", i+1, err) + } } if slack <= 0 || slack > DefaultRequestTimeout { t.Fatalf("请求 deadline 余量 = %s, want (0, %s]", slack, DefaultRequestTimeout) diff --git a/internal/runtime/embedding/budget.go b/internal/runtime/embedding/budget.go new file mode 100644 index 0000000..815ffc8 --- /dev/null +++ b/internal/runtime/embedding/budget.go @@ -0,0 +1,145 @@ +package embedding + +import "time" + +// 本文件是**单次 /embeddings 调用的预算推演**：批次上限与单次超时必须由**同一个 +// 式子**得出。任一侧单独调整都会让「一次请求的实测耗时 < 单次超时」这条不变式 +// 静默失效 —— 而失效的后果不是报错，是**服务端还在算、客户端已经放弃**：被放弃的 +// 那一批仍占着本机 llama.cpp 的算力，后面每一个请求（包括写索引时的单条请求）都 +// 排在它后面等到自己的超时，于是整条记忆写入路径被拖住、召回全线降级。 +// +// ## 不变式 +// +// requestOverhead + 一次请求的字符总量 × perCharBudget ≤ DefaultRequestTimeout / callBudgetShare +// +// 左半是**推演出来的耗时上界**，右半是该超时下允许用掉的一半（另一半是这台机器的 +// 抖动余量）。批次上限由解这个不等式得到（见 DefaultBatchLimit / DefaultBatchChars）。 +// +// ⚠️ 这条不等式**只按稳态超时（DefaultRequestTimeout）推导，且只约束批次形状**； +// 冷启动那次放宽（warmupTimeout）**不改批形状**，只放宽进程第一个请求的**截止 +// 时刻**，而且只发生一次（见 warmupTimeout 的说明）。批形状与超时仍然同源： +// 无论第一个请求的截止多宽，它带的条数/字符数永远按稳态超时算出来。 +// +// ## 实测依据（2026-10-13，本机 = AMD RX 6600 XT + llama.cpp b11371 + +// Qwen3-Embedding-0.6B Q8_0，端点 127.0.0.1:11434，**服务器空闲**，正文取自活库 +// data/cyrene.db：短正文约 33 汉字、长正文约 168 汉字；每档 5~6 次取最坏） +// +// n=1 短 0.066~0.185s 长 0.740s +// n=8 短 0.86~1.11s +// n=12 短 1.27s 长 5.47s +// n=16 短 2.20~3.31s +// n=64 短 7.49~8.49s（改前的批次上限） +// +// 拟合：单次请求耗时 ≈ 固定开销 + 字符总量 × 每字符成本；固定开销实测约 50ms， +// 每字符成本实测 2.7~3.9ms。**同一批 n≥32 起出现过数倍放大**（8.97s / 26.6s）， +// 因此大批次的耗时不可预测 —— 这也是「余量取一半」而不是「卡着边界」的依据。 +// +// 复跑命令（读活库正文、只连本机 embedding 服务、不写任何库）： +// +// python - <<'PY' +// import json,sqlite3,time,urllib.request +// URL="http://127.0.0.1:11434/v1/embeddings"; MODEL="" +// t=[r[0] for r in sqlite3.connect("data/cyrene.db").execute( +// "select content from memory_items order by id limit 64")] +// for b in (t[:1],t[:8],t[:12],t[:16],t[:64]): +// body=json.dumps({"model":MODEL,"input":b}).encode() +// r=urllib.request.Request(URL,data=body,headers={"Content-Type":"application/json"}) +// s=time.perf_counter(); urllib.request.urlopen(r,timeout=300).read() +// print(len(b), round(time.perf_counter()-s,3)) +// PY +const ( + // requestOverhead 是单次 HTTP 调用的固定开销上界（实测 ~50ms，向上取整）。 + requestOverhead = 100 * time.Millisecond + + // perCharBudget 是**每字符**的墙钟预算上界（实测 2.7~3.9 ms/字符，取 5ms）。 + perCharBudget = 5 * time.Millisecond + + // perItemBudget 是**每条文本**的墙钟预算上界：按活库里最长的正文（168 汉字， + // 最坏摊销 0.456 s/条）向上取整到 500ms。 + perItemBudget = 500 * time.Millisecond + + // callBudgetShare 表示一次请求最多用掉超时的 1/2：另一半是抖动余量 + // （见上面 n≥32 的数倍放大）。调大它 = 余量变小，必须连带复核实测。 + callBudgetShare = 2 + + // DefaultBatchLimit 是单次请求的**条数**上限，由上面三个常量推出来（唯一推演）： + // (5s/2 − 100ms) / 500ms = 4。 + DefaultBatchLimit = (int(DefaultRequestTimeout/callBudgetShare) - int(requestOverhead)) / int(perItemBudget) + + // DefaultBatchChars 是单次请求的**字符总量**上限，同一个不等式换成每字符预算： + // (5s/2 − 100ms) / 5ms = 480。它防的是「条数很少但每条很长」： + // 4 条 × 168 汉字 = 672 字符已越过它，会被再切一刀。 + DefaultBatchChars = (int(DefaultRequestTimeout/callBudgetShare) - int(requestOverhead)) / int(perCharBudget) + + // warmupTimeout 是**一次性的**冷启动放宽：本 embedder 生命周期里的**第一个** + // 请求可以拿到这个截止（模型加载 / GPU 缓冲分配还没发生过），此后**每一个** + // 请求（含刚失败的那个之后的下一个）一律回到 DefaultRequestTimeout。 + // + // 🔴 它**不是**「只要还没成功过就一直放宽」—— 那个口径会在服务挂死时把 + // fail-fast 变成 fail-slow（每次都等 warmupTimeout），并把审查的 20s 预算、 + // 索引写入的 5s 预算整段吃掉。一次性 + 8s 两条一起才是有界的： + // - 一次性：放宽最多花掉**一个**请求的时长，最坏情形之后立刻回到 5s； + // - 8s：与常态 5s **同一量级**（1.6×），且严格小于审查单条总预算 20s + // （memory.defaultReviewTimeout）⇒ 一次冷启动不可能吃光整轮审查， + // 后面的 T1 灰区判定（8s）仍有预算。这条不等式由测试守住 + // （api_url_test.go 的 TestDefaultRequestTimeout）。 + // + // 依据：2026-10-13 部署后第一分钟的生产日志里有一条 + // `vector embed for index failed ... embed 1 texts ... context deadline exceeded` + // —— 模型还没热时 5s 不够；而稳态实测单条只要 0.03~0.19s（见上面的曲线）， + // 所以放宽只需要覆盖「进程的第一个请求」这一刻。 + warmupTimeout = 8 * time.Second +) + +// callBudget 返回在 timeout 这个单次超时下，一次请求允许用掉的墙钟预算。 +func callBudget(timeout time.Duration) time.Duration { + if timeout <= 0 { + timeout = DefaultRequestTimeout + } + return timeout/callBudgetShare - requestOverhead +} + +// batchLimitsFor 由**同一个超时**推出一对批次上限（条数 / 字符总量）。 +// 它保证：按 (条数 ≤ items 且 字符总量 ≤ chars) 切出来的任何一段，其耗时上界 +// 都不超过该超时的一半。两者都至少为 1 —— 单条文本本身越界时独占一段， +// 正文**绝不截断**（截断会让向量描述的不是这条记忆）。 +func batchLimitsFor(timeout time.Duration) (items, chars int) { + budget := callBudget(timeout) + items = int(budget / perItemBudget) + if items < 1 { + items = 1 + } + chars = int(budget / perCharBudget) + if chars < 1 { + chars = 1 + } + return items, chars +} + +// chunkTexts 按「条数 ≤ maxItems 且 字符总量 ≤ maxChars」把 texts 切成若干段。 +// 单条文本的 rune 数超过 maxChars 时它独占一段（见 batchLimitsFor 的口径）。 +// 空输入返回 nil。 +func chunkTexts(texts []string, maxItems, maxChars int) [][]string { + if len(texts) == 0 { + return nil + } + if maxItems < 1 { + maxItems = 1 + } + if maxChars < 1 { + maxChars = 1 + } + var out [][]string + start := 0 + chars := 0 + for i, text := range texts { + n := len([]rune(text)) + // 段内已有内容，再加这一条就越界 ⇒ 先收口，本段从这一条重新开始。 + if i > start && (i-start >= maxItems || chars+n > maxChars) { + out = append(out, texts[start:i]) + start, chars = i, 0 + } + chars += n + } + return append(out, texts[start:]) +} diff --git a/internal/runtime/embedding/budget_test.go b/internal/runtime/embedding/budget_test.go new file mode 100644 index 0000000..415489e --- /dev/null +++ b/internal/runtime/embedding/budget_test.go @@ -0,0 +1,393 @@ +package embedding + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" +) + +// 本文件把 budget.go 的**不变式**钉成可执行的断言，两态各一条： +// +// - 「批内耗时 < 超时」⇒ 不降级：DerivedBatchCompletesWithinTimeout； +// - 「批内耗时 ≥ 超时」⇒ 超时（本次故障形态）：LegacyBatchOf64TimesOut； +// +// 以及**变异自证**：把批上限调回旧值 64（或把推演调松）会让哪条测试红， +// 见 TestBatchBudget_InvariantHolds 与 TestAPIEmbedder_LegacyBatchOf64TimesOut。 +// +// 假客户端用的是**实测曲线**（不是随口编的常数）：单次请求耗时 +// ≈ measuredOverhead + 字符总量 × measuredPerChar，取值见下面两个常量 +// （2026-10-13 本机实测：固定开销 ~50ms、每字符 2.7~3.9ms，此处取上界）。 +const ( + measuredOverhead = 50 * time.Millisecond + measuredPerChar = 4 * time.Millisecond + latencyModelChars = 33 // 活库短正文的平均长度（实测 33 汉字） +) + +// latencyTripper 是一个按上面那条曲线**真的睡够时间**的 RoundTripper： +// 它模拟「服务端会算完，不管你客户端还在不在等」。这正是本次故障的形态 —— +// 客户端 5s 放弃、服务端还在算，后面每个请求都排在它后面。 +type latencyTripper struct { + mu sync.Mutex + requests int + maxItems int + maxChars int + // abandoned 计数「客户端先放弃、服务端还在算」的次数 —— 本次故障的形态。 + // 修好之后它必须是 0（见 TestAPIEmbedder_OversizedBatchIsSplit 的断言）。 + abandoned int + // firstSleep 非零时，第一次请求额外多睡这么久（模拟冷启动的模型加载）。 + firstSleep time.Duration +} + +func (l *latencyTripper) RoundTrip(r *http.Request) (*http.Response, error) { + body, err := io.ReadAll(r.Body) + if err != nil { + return nil, err + } + var req openAICompatibleRequest + if err := json.Unmarshal(body, &req); err != nil { + return nil, err + } + chars := 0 + for _, s := range req.Input { + chars += len([]rune(s)) + } + + l.mu.Lock() + l.requests++ + if len(req.Input) > l.maxItems { + l.maxItems = len(req.Input) + } + if chars > l.maxChars { + l.maxChars = chars + } + first := l.requests == 1 + extra := l.firstSleep + l.mu.Unlock() + + wait := measuredOverhead + time.Duration(chars)*measuredPerChar + if first && extra > 0 { + wait += extra + } + select { + case <-time.After(wait): + case <-r.Context().Done(): + // 客户端放弃了：服务端那一侧不会因为客户端走了就把已排队的算力还回来 + // （实测的级联堆积正是从这一步开始的），因此这里明确记一笔。 + l.mu.Lock() + l.abandoned++ + l.mu.Unlock() + return nil, r.Context().Err() + } + + resp := openAICompatibleResponse{Data: make([]openAICompatibleDatum, 0, len(req.Input))} + for i := range req.Input { + resp.Data = append(resp.Data, openAICompatibleDatum{Index: i, Embedding: []float32{1, 0}}) + } + raw, err := json.Marshal(resp) + if err != nil { + return nil, err + } + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(string(raw))), + Header: make(http.Header), + }, nil +} + +// shortTexts 造 n 条与活库短正文等长的文本。 +func shortTexts(n int) []string { + out := make([]string, n) + line := strings.Repeat("河", latencyModelChars) + for i := range out { + out[i] = line + } + return out +} + +// TestBatchBudget_InvariantHolds 是**推演本身**的断言（不连任何网络）： +// 由超时推出来的一对上限，必须使「一次请求的耗时上界 ≤ 超时的一半」成立。 +// +// 变异自证：把 DefaultBatchLimit 的推演改成旧口径（条数写回 64）⇒ 第一条断言红； +// 若改成「把超时放大到盖住 64 条」⇒ 最后一条断言红（旧配对必须仍是越界的）。 +func TestBatchBudget_InvariantHolds(t *testing.T) { + t.Parallel() + + if DefaultBatchLimit < 1 || DefaultBatchChars < 1 { + t.Fatalf("推出来的上限必须 ≥ 1：条数=%d 字符=%d", DefaultBatchLimit, DefaultBatchChars) + } + share := DefaultRequestTimeout / callBudgetShare + byItems := requestOverhead + time.Duration(DefaultBatchLimit)*perItemBudget + if byItems > share { + t.Fatalf("条数上限 %d 的耗时上界 %s 超过超时的一半 %s（推演被调松了）", + DefaultBatchLimit, byItems, share) + } + byChars := requestOverhead + time.Duration(DefaultBatchChars)*perCharBudget + if byChars > share { + t.Fatalf("字符上限 %d 的耗时上界 %s 超过超时的一半 %s（推演被调松了）", + DefaultBatchChars, byChars, share) + } + // 旧配对（批次 64 + 5s 超时）必须仍被判为越界 —— 那正是 2026-10-13 的故障形态： + // 实测 64 条 7.5~8.5s > 5s。这里用同一套预算常量复核它仍然出界。 + legacy := requestOverhead + 64*perItemBudget + if legacy <= DefaultRequestTimeout { + t.Fatalf("旧配对（64 条）的耗时上界 %s 竟然落在单次超时 %s 之内 —— "+ + "说明超时被放大到迁就批次了，本次修法不允许", legacy, DefaultRequestTimeout) + } +} + +// TestChunkTexts_RespectsBothLimits 钉住切片：条数与字符总量两道界都要守， +// 单条本身越界的文本独占一段且**不被截断**。 +func TestChunkTexts_RespectsBothLimits(t *testing.T) { + t.Parallel() + + got := chunkTexts(shortTexts(9), 2, 100) + if len(got) != 5 { + t.Fatalf("2 条/段 切 9 条 = %d 段，want 5", len(got)) + } + for i, seg := range got { + if len(seg) > 2 { + t.Fatalf("第 %d 段有 %d 条，超过条数上限 2", i, len(seg)) + } + } + + // 字符界绑定：4 条 33 字符 = 132 字符 ⇒ 100 字符/段时必须收口（3 条 = 99 字符满， + // 余下 1 条自成一段）。 + byChars := chunkTexts(shortTexts(4), 64, 100) + if len(byChars) != 2 { + t.Fatalf("100 字符/段 切 4×33 字符 = %d 段，want 2", len(byChars)) + } + for i, seg := range byChars { + total := 0 + for _, s := range seg { + total += len([]rune(s)) + } + if total > 100 { + t.Fatalf("第 %d 段 %d 字符，超过字符上限 100", i, total) + } + } + + // 超长单条：不切碎、不截断，独占一段。 + huge := strings.Repeat("长", 500) + segs := chunkTexts([]string{huge, "短"}, 64, 100) + if len(segs) != 2 || segs[0][0] != huge { + t.Fatalf("超长单条应独占一段且不改写，got %d 段", len(segs)) + } + + if chunkTexts(nil, 4, 100) != nil { + t.Fatal("空输入应返回 nil") + } +} + +// TestAPIEmbedder_DerivedBatchCompletesWithinTimeout 钉**不降级态**： +// 按推出来的上限发一次（Embed 内部切成 1 段），在实测曲线下必须在单次超时内拿到向量。 +func TestAPIEmbedder_DerivedBatchCompletesWithinTimeout(t *testing.T) { + t.Parallel() + + rt := &latencyTripper{} + emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"}, + WithRoundTripper(rt)) + + texts := shortTexts(DefaultBatchLimit) + vectors, err := emb.Embed(context.Background(), texts) + if err != nil { + t.Fatalf("上限内的批不该超时：%v", err) + } + if len(vectors) != len(texts) { + t.Fatalf("got %d vectors, want %d", len(vectors), len(texts)) + } + if rt.requests != 1 { + t.Fatalf("上限内的批应发 1 次请求，实际 %d 次", rt.requests) + } +} + +// TestAPIEmbedder_OversizedBatchIsSplit 钉**不变式与调用方无关**： +// 调用方给 3 倍上限的切片时，Embed 自己切成多段、每段都在界内，向量顺序与入参一致。 +func TestAPIEmbedder_OversizedBatchIsSplit(t *testing.T) { + t.Parallel() + + rt := &latencyTripper{} + emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"}, + WithRoundTripper(rt)) + + texts := shortTexts(DefaultBatchLimit * 3) + vectors, err := emb.Embed(context.Background(), texts) + if err != nil { + t.Fatalf("超大切片应被切开而不是超时：%v", err) + } + if len(vectors) != len(texts) { + t.Fatalf("got %d vectors, want %d（切开后必须按序拼回）", len(vectors), len(texts)) + } + if rt.requests < 3 { + t.Fatalf("3 倍上限应至少切成 3 段，实际 %d 段", rt.requests) + } + if rt.maxItems > DefaultBatchLimit { + t.Fatalf("有请求带了 %d 条，超过推出来的条数上限 %d", rt.maxItems, DefaultBatchLimit) + } + if rt.maxChars > DefaultBatchChars { + t.Fatalf("有请求带了 %d 字符，超过推出来的字符上限 %d", rt.maxChars, DefaultBatchChars) + } + // 🔴 本次修复的核心不变式，确定性断言（不依赖机器负载）： + // 没有任何一次请求发生「客户端先放弃、服务端还在算」——那正是级联堆积的起点。 + if rt.abandoned != 0 { + t.Fatalf("有 %d 次请求被客户端先放弃（服务端还在算）⇒ 不变式没守住", rt.abandoned) + } + // 服务端侧的同一条不变式：任何一次请求的建模耗时（实测曲线）都落在单次超时内。 + if cost := measuredOverhead + time.Duration(rt.maxChars)*measuredPerChar; cost > DefaultRequestTimeout { + t.Fatalf("最大一次请求的实测曲线耗时 %s 越过单次超时 %s", cost, DefaultRequestTimeout) + } +} + +// TestAPIEmbedder_LegacyBatchOf64TimesOut 是**变异自证**的另一半：同样是实测曲线， +// 旧配对（一次请求 64 条 + 5s 超时）必须**超时**。它证明把上限写回 64 不是「保守」，而是 +// 必然把客户端与服务端撕裂（客户端走了、服务端还在算 ⇒ 后续请求全排队）。 +func TestAPIEmbedder_LegacyBatchOf64TimesOut(t *testing.T) { + t.Parallel() + + rt := &latencyTripper{} + emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"}, + WithRoundTripper(rt)) + + // 先把一次性冷启动放宽花掉，否则第一次调用按 warmupTimeout 算、不会超时。 + if _, err := emb.Embed(context.Background(), shortTexts(1)); err != nil { + t.Fatalf("预热调用失败：%v", err) + } + start := time.Now() + _, err := emb.embedChunk(context.Background(), shortTexts(64)) + elapsed := time.Since(start) + if err == nil { + t.Fatalf("旧配对（64 条 + %s 超时）在实测曲线下不该成功", DefaultRequestTimeout) + } + if elapsed > DefaultRequestTimeout+2*time.Second { + t.Fatalf("超时保护没生效：等了 %s", elapsed) + } +} + +// TestAPIEmbedder_WarmupIsOneShot 钉 p1 的核心：冷启动放宽**一次性**，三种失败形态 +// 下都必须是「第一次最多等 warmupTimeout，第二次起一律 DefaultRequestTimeout」。 +// +// 这是 reviewer 在 round1 打回的那条：旧实现 `warmed` 只在 2xx 之后才翻，于是 +// 「服务接受连接但永不回包」时**每一次**调用都等完整放宽预算（30s），把 fail-fast +// 变成 fail-slow —— 一次挂死就能吃光审查的 20s 预算、把索引写入拖成 8×30s。 +func TestAPIEmbedder_WarmupIsOneShot(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + rt func(t *testing.T) http.RoundTripper + }{ + { + name: "连接被拒", + rt: func(t *testing.T) http.RoundTripper { + return roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("dial tcp 127.0.0.1:11434: connectex: 目标计算机积极拒绝") + }) + }, + }, + { + name: "HTTP 500", + rt: func(t *testing.T) http.RoundTripper { + return roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{ + StatusCode: http.StatusInternalServerError, + Body: io.NopCloser(strings.NewReader("server exploded")), + Header: make(http.Header), + }, nil + }) + }, + }, + { + name: "接受连接但永不回包", + rt: func(t *testing.T) http.RoundTripper { + return roundTripFunc(func(r *http.Request) (*http.Response, error) { + <-r.Context().Done() // 只等服务端超时，永不回包 + return nil, r.Context().Err() + }) + }, + }, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"}, + WithRoundTripper(tc.rt(t))) + + // 第一次：放宽预算。断言上界 = warmupTimeout（不得再宽）， + // 下界只对会真正等到截止的那一种（挂死）有意义。 + start := time.Now() + _, err := emb.Embed(context.Background(), shortTexts(1)) + first := time.Since(start) + if err == nil { + t.Fatal("这一组失败形态不该成功") + } + t.Logf("首次耗时 = %s", first.Round(time.Millisecond)) + if first > warmupTimeout+time.Second { + t.Fatalf("首次耗时 %s 超过冷启动放宽 %s", first, warmupTimeout) + } + + // 第二次：**必须**回到常态超时 —— 一次挂死绝不许变成每次都慢。 + start = time.Now() + if _, err := emb.Embed(context.Background(), shortTexts(1)); err == nil { + t.Fatal("这一组失败形态不该成功") + } + second := time.Since(start) + t.Logf("第二次耗时 = %s（常态超时 %s）", second.Round(time.Millisecond), DefaultRequestTimeout) + if second > DefaultRequestTimeout+time.Second { + t.Fatalf("第二次耗时 %s 超过常态超时 %s ⇒ 放宽成了长期的", second, DefaultRequestTimeout) + } + // 放宽许可必须已经被花掉（它是一次性的，任何路径都不许给它续期）。 + if emb.warmupLeft.Load() { + t.Fatal("第二次请求之后放宽许可仍为真 —— 一次性的约束被绕过了") + } + }) + } +} + +// TestAPIEmbedder_WarmupCoversColdStartOnce 钉放宽的**正面**作用：第一次请求 +// （模型加载 / GPU 缓冲分配）可以比常态超时多等一点，只要它落在 warmupTimeout +// 之内；而这一刻过去之后，请求立刻回到常态超时（deadline 断言）。 +func TestAPIEmbedder_WarmupCoversColdStartOnce(t *testing.T) { + t.Parallel() + + rt := &latencyTripper{firstSleep: DefaultRequestTimeout + time.Second} + emb := NewAPI(config.EmbeddingConfig{APIURL: "http://127.0.0.1:11434/v1/embeddings"}, + WithRoundTripper(rt)) + + if _, err := emb.Embed(context.Background(), shortTexts(1)); err != nil { + t.Fatalf("冷启动第一次调用应被 warmupTimeout(%s) 兜住：%v", warmupTimeout, err) + } + if emb.warmupLeft.Load() { + t.Fatal("放宽许可应在第一次请求时就被花掉") + } + + var slack time.Duration + steady := roundTripFunc(func(r *http.Request) (*http.Response, error) { + dl, ok := r.Context().Deadline() + if !ok { + t.Error("稳态请求没有 deadline") + } else { + slack = time.Until(dl) + } + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(`{"data":[{"index":0,"embedding":[1]}]}`)), + Header: make(http.Header), + }, nil + }) + emb.client = &http.Client{Transport: steady} + if _, err := emb.Embed(context.Background(), shortTexts(1)); err != nil { + t.Fatalf("稳态调用失败：%v", err) + } + if slack <= 0 || slack > DefaultRequestTimeout { + t.Fatalf("稳态单次超时余量 = %s, want (0, %s]", slack, DefaultRequestTimeout) + } +} diff --git a/internal/runtime/memory/tidy.go b/internal/runtime/memory/tidy.go index 1e5967a..a7e954f 100644 --- a/internal/runtime/memory/tidy.go +++ b/internal/runtime/memory/tidy.go @@ -119,12 +119,17 @@ func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) { return stats, err } stats.RunID = runID + // ledger 是**本轮实际发生**的召回来源记账口（不是配置里那一种）：生产轮可以 + // 多分区混合（有的走语义、有的降级），因此汇总行只能由实测记录推出。 + ledger := &recallLedger{} failures := t.scanPartitions(ctx, partitions, &stats, tidyRoundPolicy{ sink: func(ctx context.Context, cands []TidyCandidate, stats *TidyStats) { t.persistCandidates(ctx, runID, cands, stats) }, - recall: t.fallbackRecall, + recall: func(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) { + return t.fallbackRecall(ctx, p, items, stats, budget, ledger) + }, }) // 轮末自动落地（2026-10-07 主人口径）：规则层判为高置信的待定候选直接落地， @@ -150,7 +155,7 @@ func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) { failures = append(failures, err) } - t.logRun(stats, run) + t.logRun(stats, run, ledger) t.auditAutoApply(auto) switch { @@ -165,7 +170,14 @@ func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) { // logRun 输出本轮的可观测口径：扫描 N / 候选 M / 建议 K / 否决原因分布。 // 它带 source（memory/tidy），因此这条汇总永远可回溯到整理子系统。 -func (t *Tidier) logRun(stats TidyStats, run TidyRun) { +// +// ledger 给出的是**本轮实际发生**的召回来源（不是配置态）。这一条是 M6 2026-10-13 +// 的 medium finding 的落点：旧实现按「有没有配 embedder」写 recall_mode，全量降级 +// 在汇总行里被写成 embedding。现在： +// - recall_mode = ""：没有任何分区真正算过相似度（没有候选可比）； +// - recall_mode = lexical ∧ recall_degraded_partitions > 0：语义不可用、降级了； +// - recall_mode = mixed：多分区里一部分走了语义、一部分降级 —— 不再被抹平。 +func (t *Tidier) logRun(stats TidyStats, run TidyRun, ledger *recallLedger) { fields := []any{ "run_id", run.ID, "rule_version", run.RuleVersion, @@ -181,7 +193,8 @@ func (t *Tidier) logRun(stats TidyStats, run TidyRun) { "kept", stats.Kept, "errors", run.Errors, "duration_ms", run.FinishedAt.Sub(run.StartedAt).Milliseconds(), - "recall_mode", tidyRecallMode(t.deps.Embedder != nil), + "recall_mode", ledger.label(), + "recall_degraded_partitions", ledger.degradedPartitions(), "auto_apply", t.cfg.autoApplyEnabled(), "auto_applied", stats.AutoApplied, "auto_merge", stats.AutoMerged, @@ -232,11 +245,15 @@ func (t *Tidier) scanPartitions(ctx context.Context, partitions []tidyPartition, // fallbackRecall 是**生产轮**的召回：相似度来源不可用时降级到零依赖的词项 // 余弦（离线/无 embedding 配置时可跑，宁可比语义路径更保守），并记一条 WARN。 // 两个来源**不可混用阈值**：阈值只有一个（design §13 开放点 2 的 0.80）。 -func (t *Tidier) fallbackRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) { - pairs, err := t.recallPairs(ctx, p, items, stats, budget) +// +// 降级这一笔**必须记进 ledger**：汇总行/面板上的召回来源只能由实测记录推出， +// 否则「全量降级」会被报成「向量正常」（M6 2026-10-13 的 medium finding）。 +func (t *Tidier) fallbackRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget, ledger *recallLedger) ([]tidyPair, error) { + pairs, err := t.recallPairs(ctx, p, items, stats, budget, ledger) if err == nil { return pairs, nil } + ledger.noteLexical(true) t.log.Warn("memory tidy embedding recall unavailable, falling back to lexical", "partition", p.label(), "error", err) return t.collectPairs(p, items, t.cfg.RecallThreshold, diff --git a/internal/runtime/memory/tidy_config.go b/internal/runtime/memory/tidy_config.go index f351192..59eee80 100644 --- a/internal/runtime/memory/tidy_config.go +++ b/internal/runtime/memory/tidy_config.go @@ -4,6 +4,7 @@ import ( "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" ) // 本文件收拢记忆整理（design/23-记忆整理与去重）全部的标识、默认值与旋钮， @@ -74,7 +75,17 @@ const ( // tidyEmbedBatch 是单次 Embed 调用的文本条数上限，避免把整个分区的文本 // 一次性交给远端 embedding API。 - tidyEmbedBatch = 64 + // + // 🔴 它**不是**拍出来的数字，而是 internal/runtime/embedding 里**同一条推演** + // 的结果（budget.go：一次请求的耗时上界 ≤ 单次超时的一半 ⇒ 批上限 = 超时的一半 + // ÷ 单条预算上界）。2026-10-13：旧值 64 的实测耗时 7.5~8.5s 已越过客户端 5s + // 单次超时 ⇒ 客户端放弃、服务端继续算 ⇒ 本机 embedding 服务被顶住，单条请求 + // （写向量索引）也跟着超时。把批上限与超时绑到同一条推演上，就不会再出现 + // 「只调一侧、另一侧静默失效」。 + // + // 批上限在 embedder 内部**还有一道强制**（Embed 自己切片），因此这里即使被改大 + // 也不会把服务端顶住；这一行只是让调用方别去构造注定被切开的超大切片。 + tidyEmbedBatch = embedding.DefaultBatchLimit // minTidyContentRunes 是参与比对的最小内容长度（按 rune）。过短的内容 // （如单字）没有可判定的事实要素，进比对只会产出噪声，直接跳过。 @@ -126,6 +137,11 @@ const ( maxTidyRejectBatchLimit = 10000 ) +// 编译期不变式（2026-10-13）：批上限一旦被调回超过 embedding 推出来的值 +// （例：写回旧值 64），下面这行常量表达式会变成负数、uint 溢出 ⇒ **编译失败**。 +// 这条断言把「批上限与超时同源」从注释里的承诺变成编译器守的门。 +const _ = uint(embedding.DefaultBatchLimit - tidyEmbedBatch) + // TidyConfig 携带记忆整理的调优旋钮。零值经 normalize 选取安全默认值， // 因此半填的 Config 仍可用。 type TidyConfig struct { diff --git a/internal/runtime/memory/tidy_dryrun.go b/internal/runtime/memory/tidy_dryrun.go index b90ea9a..14e0dc3 100644 --- a/internal/runtime/memory/tidy_dryrun.go +++ b/internal/runtime/memory/tidy_dryrun.go @@ -52,11 +52,15 @@ type TidyDryRun struct { // 「完整的建议」展示）。 func (t *Tidier) DryRun(ctx context.Context) (TidyDryRun, error) { started := t.now() - out := TidyDryRun{RecallMode: tidyRecallMode(t.deps.Embedder != nil)} + out := TidyDryRun{} stats, partitions, err := t.beginRound(ctx) if err != nil { return out, err } + // ledger 是**本轮实际发生**的召回来源：试算会写进 RecallMode 返回给面板。 + // 旧实现按「有没有配 embedder」写（tidyRecallMode(deps.Embedder != nil)）， + // 那会让面板把「向量不可用」读成「向量正常」（M6 2026-10-13 的 medium finding）。 + ledger := &recallLedger{} // 收集出口：闭包只在 scanPartitions 的**串行**循环里被调用，因此无需加锁。 collected := make([]TidyCandidate, 0) @@ -64,25 +68,38 @@ func (t *Tidier) DryRun(ctx context.Context) (TidyDryRun, error) { sink: func(_ context.Context, cands []TidyCandidate, _ *TidyStats) { collected = append(collected, cands...) }, - recall: t.strictRecall, + recall: func(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) { + return t.strictRecall(ctx, p, items, stats, budget, ledger) + }, } failures := t.scanPartitions(ctx, partitions, &stats, policy) out.Stats = stats out.Candidates = collected + // 如实上报**实际发生**的来源：空串 = 没有任何分区真正算过相似度（「没候选」）。 + // 「尝试过但向量源不可用」是另一件事，由 recall_unavailable_partitions 单独给出 + // （见 logDryRun 的说明与 recallLedger.unavailable）。 + out.RecallMode = ledger.label() out.DurationMS = t.now().Sub(started).Milliseconds() + // 汇总行**成败都记**（与 RunOnce 同一口径）：失败那一路的 + // `recall_unavailable_partitions` 正是「向量不可用」与「没候选」的唯一区分点， + // 只在成功时记就等于把这条口径丢掉。 + t.logDryRun(out, ledger) if len(failures) > 0 { return out, fmt.Errorf("memory/tidy: dry-run finished with errors: %w", errors.Join(failures...)) } - t.logDryRun(out) return out, nil } // strictRecall 是**试算轮**的召回：语义召回不可用时**不**降级，如实上报。 // 日志只记分区与错误，绝不带记忆正文（候选预览里已含正文）。 -func (t *Tidier) strictRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) { - pairs, err := t.recallPairs(ctx, p, items, stats, budget) +// +// 失败时要往 ledger 记一笔 unavailable：否则「尝试过但向量源不可用」与「没有候选 +// 可比」在观测面上同值（reviewer round1 的 p2）。 +func (t *Tidier) strictRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget, ledger *recallLedger) ([]tidyPair, error) { + pairs, err := t.recallPairs(ctx, p, items, stats, budget, ledger) if err != nil { + ledger.noteUnavailable() t.log.Warn("memory tidy dry-run recall unavailable", "partition", p.label(), "items", len(items), "error", err) return nil, fmt.Errorf("%w: %w", ErrTidyRecallUnavailable, err) @@ -92,7 +109,17 @@ func (t *Tidier) strictRecall(ctx context.Context, p tidyPartition, items []Memo // logDryRun 输出试算的观测口径。它带 source（memory/tidy），**不带任何记忆 // 正文** —— 只记计数、分区数与截断标签。 -func (t *Tidier) logDryRun(out TidyDryRun) { +// +// 🔴 `recall_mode` + `recall_unavailable_partitions` 一起才把三件事分开 +// （reviewer round1 的 p2：它们曾同值）： +// - mode=""、unavailable=0：本轮**没有候选可比**（压根没算过相似度）； +// - mode=""、unavailable>0：**尝试过召回但向量源不可用**（严格口径不降级， +// 调用方拿到的是 `ErrTidyRecallUnavailable` ⇒ HTTP 500 `tidy_recall_unavailable`）； +// - mode=embedding/lexical：真的用该来源算过。 +// +// 没有为此扩 `recallMode` 的取值集合：试算失败时**不序列化** result（响应是错误 +// 体），一个新取值只会出现在这行日志里，却要动一次 API 契约与文档 —— 不划算。 +func (t *Tidier) logDryRun(out TidyDryRun, ledger *recallLedger) { t.log.Info("memory tidy dry-run finished", "rule_version", TidyRuleVersion, "partitions", out.Stats.Partitions, @@ -104,6 +131,7 @@ func (t *Tidier) logDryRun(out TidyDryRun) { "veto_reasons", tidyReasonsLabel(out.Stats.VetoReasons), "errors", out.Stats.Errors, "recall_mode", out.RecallMode, + "recall_unavailable_partitions", ledger.unavailablePartitions(), "duration_ms", out.DurationMS) } diff --git a/internal/runtime/memory/tidy_recall.go b/internal/runtime/memory/tidy_recall.go index 99e3d64..aed1c2b 100644 --- a/internal/runtime/memory/tidy_recall.go +++ b/internal/runtime/memory/tidy_recall.go @@ -25,8 +25,105 @@ const ( tidyRecallEmbedding = "embedding" // tidyRecallLexical 是本轮相似度来自零依赖的词项余弦。 tidyRecallLexical = "lexical" + // tidyRecallMixed 是多分区的一轮里**部分**分区走了语义、部分降级到词项余弦。 + // 它取代「挑一个代表值报上去」的做法：降级一旦发生就必须看得见。 + tidyRecallMixed = "mixed" ) +// recallLedger 记录**一轮里实际发生**的召回来源（不是配置里那一种）。 +// +// 为什么需要它（M6 2026-10-13 的 medium finding）：整理轮与试算轮的 `recall_mode` +// 曾按「有没有配 embedder」写出去（旧实现 tidyRecallMode(hasEmbedder bool)），于是 +// **全量降级**在汇总行/面板里被写成 `embedding` —— 汇总与事实相反，运维据此判读会 +// 把「向量不可用」读成「向量正常工作」。本类型把口径翻过来：只在**真的用了**那条 +// 路的地方记一笔，label() 完全由这些实测记录推出。 +type recallLedger struct { + // embedding / lexical 是**真正用该来源算过相似度**的分区数。 + embedding int + lexical int + // degraded 是**因相似度来源不可用**而降级到词项余弦的分区数（不是「没配 + // embedder」那种配置形态）。它把「没候选」（label 为空串）与「向量不可用」 + // （label 为 lexical 且 degraded > 0）分开 —— 这两件事在运维上不是一回事。 + degraded int + // unavailable 是**尝试过召回但向量源不可用、且本轮不降级**（试算轮的严格口径） + // 的分区数。它与 degraded 的区别是「有没有改用词项余弦」：生产轮会降级 + // （degraded），试算轮只报错。 + // + // 🔴 它存在的理由（reviewer round1 的 p2）：试算轮在向量源不可用时 + // `RecallMode` 与「没有候选可比」**同为空串**，两者在那个字段上分不开。 + // 现在 provenance 由 unavailablePartitions() 单独给出，日志行能区分； + // 调用方（面板）那条路本来就有明确错误码（500 tidy_recall_unavailable）， + // 因此**没有**为此扩 recallMode 的取值集合（见 logDryRun 的说明）。 + unavailable int +} + +// noteEmbedding 记一笔「这个分区真的用了语义 embedding」。 +func (l *recallLedger) noteEmbedding() { + if l != nil { + l.embedding++ + } +} + +// noteLexical 记一笔「这个分区用了词项余弦」；degraded 表示它是**降级**来的 +// （embedder 报错 / 向量条数错位），而不是「本来就没配 embedder」。 +func (l *recallLedger) noteLexical(degraded bool) { + if l == nil { + return + } + l.lexical++ + if degraded { + l.degraded++ + } +} + +// noteUnavailable 记一笔「这个分区尝试召回但向量源不可用，且本轮没有降级」 +// （试算轮的严格口径：宁可报错，也不把「没算出来」说成「算过了，没有」）。 +func (l *recallLedger) noteUnavailable() { + if l != nil { + l.unavailable++ + } +} + +// label 返回本轮**实际**的召回来源： +// - ""：没有任何分区真正算过相似度（没有候选可比）—— 「没候选」不是「降级」； +// - embedding / lexical：所有算过的分区都走同一条路； +// - mixed：两种都有（多分区里有一部分降级）。 +// +// ⚠️ 「尝试过但不可用」**不改变** label：它没有产出任何相似度，报 embedding / +// lexical 都是假的；它的存在由 unavailablePartitions() 单独给出（见 recallLedger +// 的 unavailable 字段说明）。 +func (l *recallLedger) label() string { + if l == nil { + return "" + } + switch { + case l.embedding == 0 && l.lexical == 0: + return "" + case l.embedding == 0: + return tidyRecallLexical + case l.lexical == 0: + return tidyRecallEmbedding + default: + return tidyRecallMixed + } +} + +// degradedPartitions 返回本轮因向量不可用而降级的分区数。 +func (l *recallLedger) degradedPartitions() int { + if l == nil { + return 0 + } + return l.degraded +} + +// unavailablePartitions 返回本轮「尝试召回但向量源不可用、且没有降级」的分区数。 +func (l *recallLedger) unavailablePartitions() int { + if l == nil { + return 0 + } + return l.unavailable +} + // tidyPair 是一条被召回的候选对。 type tidyPair struct { a, b MemoryItem @@ -35,18 +132,23 @@ type tidyPair struct { mode string } -// recallPairs 在分区内召回候选对。条目少于 2 条时为空操作（无候选、不报错）。 +// recallPairs 在分区内召回候选对。条目少于 2 条时为空操作（无候选、不报错）， +// 且**不往 ledger 记任何一笔** —— 「没有候选可比」不是「走哪条路召回」。 // // 相似度来源不可用时（embedder 报错、向量条数错位）返回错误 —— 回退决策不在 // 这里：生产轮把它降级成词项余弦，试算轮如实上报（见 tidyRecall 策略）。 // +// ledger 是**实际发生**的来源记账口（见 recallLedger）：它只在真正选定相似度 +// 函数的那一行记账，因此谁也无法把它报成「配置里那一种」。 +// // 返回的对按「分数降序 + id」确定性排序，使每分区候选上限的截断可复现 // （同一批数据两次运行产出同一批候选 ⇒ 幂等，design §4.1 A）。 -func (t *Tidier) recallPairs(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) { +func (t *Tidier) recallPairs(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget, ledger *recallLedger) ([]tidyPair, error) { if len(items) < 2 { return nil, nil } if t.deps.Embedder == nil { + ledger.noteLexical(false) return t.collectPairs(p, items, t.cfg.RecallThreshold, lexicalSimFunc(items), tidyRecallLexical, stats, budget), nil } @@ -57,6 +159,7 @@ func (t *Tidier) recallPairs(ctx context.Context, p tidyPartition, items []Memor if len(vecs) != len(items) { return nil, ErrTidyEmbedCount } + ledger.noteEmbedding() return t.collectPairs(p, items, t.cfg.RecallThreshold, vectorSimFunc(vecs), tidyRecallEmbedding, stats, budget), nil } @@ -205,10 +308,6 @@ func clampUnit(v float64) float64 { return v } -// tidyRecallMode 报告本轮相似度的来源，便于运维判断阈值是否在预期口径上。 -func tidyRecallMode(hasEmbedder bool) string { - if hasEmbedder { - return tidyRecallEmbedding - } - return tidyRecallLexical -} +// 旧的 tidyRecallMode(hasEmbedder bool) 已删除（M6 2026-10-13 的 medium finding）： +// 它按**配置态**报「本轮相似度来源」，把全量降级掩盖成 embedding。现在的口径是 +// recallLedger —— 只报**实际发生**的那一种，见本文件顶部的 recallLedger。 diff --git a/internal/runtime/memory/tidy_recall_mode_test.go b/internal/runtime/memory/tidy_recall_mode_test.go new file mode 100644 index 0000000..d955e1d --- /dev/null +++ b/internal/runtime/memory/tidy_recall_mode_test.go @@ -0,0 +1,216 @@ +package memory + +import ( + "bytes" + "context" + "errors" + "log/slog" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" +) + +// 本文件钉 M6（2026-10-13）的 medium finding：整理轮/试算轮的 `recall_mode` +// 必须是**实际发生**的那一种，不是配置态。 +// +// 旧实现（tidyRecallMode(hasEmbedder bool)）按「有没有配 embedder」写，于是 +// 「配了但是向量不可用 ⇒ 全量降级」在汇总行里被写成 embedding —— 汇总与事实相反。 +// +// 「配了但不可用」的 embedder 复用 tidy_dryrun_test.go 的 failingEmbedder +// （正是线上那 2568 次 `embedding recall unavailable` 的形态）。 + +// newLogCaptureTidier 构建一个把汇总行写进 buffer 的整理器（便于断言日志字段）。 +func newLogCaptureTidier(t *testing.T, store *Store, cfg TidyConfig, emb embedding.Embedder) (*Tidier, *bytes.Buffer) { + t.Helper() + buf := &bytes.Buffer{} + tr, err := NewTidier(cfg, TidyDeps{ + Store: store, + Embedder: emb, + Log: slog.New(slog.NewTextHandler(buf, nil)), + }) + if err != nil { + t.Fatalf("NewTidier: %v", err) + } + return tr, buf +} + +// TestRecallLedger_LabelComesFromActualUse 钉记账口本身：label 完全由「真的用了 +// 哪条路」推出，混合与降级都不许被抹平。 +func TestRecallLedger_LabelComesFromActualUse(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + notes func(l *recallLedger) + want string + degraded int + }{ + {name: "没算过（没候选）", notes: func(*recallLedger) {}, want: ""}, + {name: "全部语义", notes: func(l *recallLedger) { l.noteEmbedding() }, want: tidyRecallEmbedding}, + {name: "全部词项", notes: func(l *recallLedger) { l.noteLexical(false) }, want: tidyRecallLexical}, + { + name: "降级到词项", + notes: func(l *recallLedger) { l.noteLexical(true) }, + want: tidyRecallLexical, + degraded: 1, + }, + { + name: "多分区混合（部分降级）", + notes: func(l *recallLedger) { l.noteEmbedding(); l.noteLexical(true) }, + want: tidyRecallMixed, + degraded: 1, + }, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + var l recallLedger + tc.notes(&l) + if got := l.label(); got != tc.want { + t.Fatalf("label = %q, want %q", got, tc.want) + } + if got := l.degradedPartitions(); got != tc.degraded { + t.Fatalf("degraded = %d, want %d", got, tc.degraded) + } + }) + } +} + +// TestTidyRunLog_RecallModeReportsDegradation 是 finding 的正面回归：配了 embedder +// 但它报错 ⇒ 汇总行必须写 `recall_mode=lexical` 且 `recall_degraded_partitions=1`， +// **绝不能**写成 embedding。 +func TestTidyRunLog_RecallModeReportsDegradation(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + seedDryRunFixture(t, store) + + emb := failingEmbedder{err: errors.New("embedding: request: context deadline exceeded")} + tr, buf := newLogCaptureTidier(t, store, tidyTestCfg(0.80), emb) + if _, err := tr.RunOnce(ctx); err != nil { + t.Fatalf("RunOnce 应当降级跑完而不是失败：%v", err) + } + + line := lastLogLine(t, buf, "memory tidy run finished") + if !strings.Contains(line, "recall_mode=lexical") { + t.Fatalf("降级轮的汇总行必须写 lexical，实际：%s", line) + } + if strings.Contains(line, "recall_mode=embedding") { + t.Fatalf("降级轮被报成了 embedding（正是本 finding 的形态）：%s", line) + } + if !strings.Contains(line, "recall_degraded_partitions=1") { + t.Fatalf("降级分区数没写出来：%s", line) + } +} + +// TestTidyRunLog_RecallModeReportsEmbeddingWhenHealthy 是反面：embedder 正常时 +// 汇总行必须如实写 embedding、降级数为 0（不能因为「怕误报」而一律写降级）。 +func TestTidyRunLog_RecallModeReportsEmbeddingWhenHealthy(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + seedDryRunFixture(t, store) + + tr, buf := newLogCaptureTidier(t, store, tidyTestCfg(0.80), dryRunFixtureEmbedder()) + if _, err := tr.RunOnce(ctx); err != nil { + t.Fatalf("RunOnce: %v", err) + } + + line := lastLogLine(t, buf, "memory tidy run finished") + if !strings.Contains(line, "recall_mode=embedding") { + t.Fatalf("健康轮必须写 embedding，实际：%s", line) + } + if !strings.Contains(line, "recall_degraded_partitions=0") { + t.Fatalf("健康轮的降级数必须为 0：%s", line) + } +} + +// TestTidyRunLog_RecallModeEmptyWhenNoPairs 钉「没候选」与「向量不可用」可区分： +// 分区里只有一条（压根没有对可比）时，来源是**空串**，而不是任何一条召回路径。 +func TestTidyRunLog_RecallModeEmptyWhenNoPairs(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + seedTidyItem(t, store, habitItem(fixtureHabitShort, base, 0.9)) + + emb := failingEmbedder{err: errors.New("embedding: request: context deadline exceeded")} + tr, buf := newLogCaptureTidier(t, store, tidyTestCfg(0.80), emb) + if _, err := tr.RunOnce(ctx); err != nil { + t.Fatalf("RunOnce: %v", err) + } + + line := lastLogLine(t, buf, "memory tidy run finished") + if !strings.Contains(line, `recall_mode=""`) { + t.Fatalf("没有候选可比时来源应为空串（区别于降级），实际：%s", line) + } + if !strings.Contains(line, "recall_degraded_partitions=0") { + t.Fatalf("没有候选可比不算降级：%s", line) + } +} + +// TestTidyDryRun_UnavailableIsDistinctFromNoCandidates 钉 reviewer round1 p2： +// 试算轮「尝试过召回但向量源不可用」与「没有候选可比」**必须能区分**。 +// +// 两者的 `RecallMode` 都是空串（都不曾产出相似度，报 embedding/lexical 都是假的）， +// 区分点在 `recall_unavailable_partitions`：不可用时为 1，没候选时为 0。调用方 +// （面板）另有一条更硬的区分 —— 失败那一路不序列化 result，直接是 +// `500 tidy_recall_unavailable`。 +func TestTidyDryRun_UnavailableIsDistinctFromNoCandidates(t *testing.T) { + t.Parallel() + + t.Run("向量源不可用", func(t *testing.T) { + t.Parallel() + store := newTestStore(t) + seedDryRunFixture(t, store) + + emb := failingEmbedder{err: errors.New("embedding: request: context deadline exceeded")} + tr, buf := newLogCaptureTidier(t, store, tidyTestCfg(0.80), emb) + out, err := tr.DryRun(context.Background()) + if !errors.Is(err, ErrTidyRecallUnavailable) { + t.Fatalf("试算轮必须如实上报不可用，err = %v", err) + } + if out.RecallMode == tidyRecallEmbedding { + t.Fatalf("向量不可用却把 RecallMode 报成 embedding（配置态口径）：%q", out.RecallMode) + } + line := lastLogLine(t, buf, "memory tidy dry-run finished") + if !strings.Contains(line, "recall_unavailable_partitions=1") { + t.Fatalf("汇总行必须写出「尝试过但不可用」的计数，实际：%s", line) + } + }) + + t.Run("没有候选可比", func(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + seedTidyItem(t, store, habitItem(fixtureHabitShort, base, 0.9)) + + emb := failingEmbedder{err: errors.New("embedding: request: context deadline exceeded")} + tr, buf := newLogCaptureTidier(t, store, tidyTestCfg(0.80), emb) + if _, err := tr.DryRun(context.Background()); err != nil { + t.Fatalf("没有对可比时试算不该报错：%v", err) + } + line := lastLogLine(t, buf, "memory tidy dry-run finished") + if !strings.Contains(line, "recall_unavailable_partitions=0") { + t.Fatalf("没候选不是「不可用」，计数必须为 0，实际：%s", line) + } + if !strings.Contains(line, `recall_mode=""`) { + t.Fatalf("没候选时来源为空串，实际：%s", line) + } + }) +} + +// lastLogLine 返回 buffer 里最后一条含 marker 的日志行。 +func lastLogLine(t *testing.T, buf *bytes.Buffer, marker string) string { + t.Helper() + lines := strings.Split(strings.TrimSpace(buf.String()), "\n") + for i := len(lines) - 1; i >= 0; i-- { + if strings.Contains(lines[i], marker) { + return lines[i] + } + } + t.Fatalf("日志里没有找到 %q：\n%s", marker, buf.String()) + return "" +}