diff --git a/internal/runtime/embedding/api.go b/internal/runtime/embedding/api.go index 92cda5f..1ff6a61 100644 --- a/internal/runtime/embedding/api.go +++ b/internal/runtime/embedding/api.go @@ -11,6 +11,7 @@ import ( "net/http" "strings" "sync" + "sync/atomic" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" @@ -38,23 +39,21 @@ 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 @@ -93,6 +92,12 @@ type APIEmbedder struct { cfg config.EmbeddingConfig client *http.Client log *slog.Logger + // timeout 是**单次**请求的墙钟上限,与批次上限同源(见 budget.go)。 + // 默认 DefaultRequestTimeout;WithRequestTimeout 可换,换了批次上限也跟着换。 + timeout time.Duration + // warmed 报告是否已经拿到过成功响应。未热之前用 warmupTimeout(模型加载 / + // GPU 缓冲分配),热之后一律用 timeout。 + warmed atomic.Bool mu sync.RWMutex dim int @@ -130,12 +135,24 @@ func WithLogger(l *slog.Logger) APIOption { } } +// WithRequestTimeout 换掉单次请求的墙钟上限(非正数忽略,保留默认)。 +// 它同时决定批次上限 —— 两者是同一条推演(budget.go),因此这里是**唯一**的 +// 调优点,不允许调用方各自去改批次。 +func WithRequestTimeout(d time.Duration) APIOption { + return func(e *APIEmbedder) { + if d > 0 { + e.timeout = d + } + } +} + // NewAPI builds an APIEmbedder from cfg using the default HTTP client. func NewAPI(cfg config.EmbeddingConfig, opts ...APIOption) *APIEmbedder { e := &APIEmbedder{ - cfg: cfg, - client: defaultHTTPClient(), - log: slog.New(slog.NewTextHandler(io.Discard, nil)), + cfg: cfg, + client: defaultHTTPClient(), + log: slog.New(slog.NewTextHandler(io.Discard, nil)), + timeout: DefaultRequestTimeout, } for _, opt := range opts { opt(e) @@ -155,8 +172,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 +187,34 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro return nil, ErrNoAPIURL } - callCtx, cancel := context.WithTimeout(ctx, DefaultRequestTimeout) + // 切片用的上限由**稳态**超时推出(e.timeout),与 requestTimeout() 无关: + // 冷启动那一次只是**截止时刻**更宽(warmupTimeout),批次形状不变 —— 否则 + // 首个请求会按 30s 的预算切出一个巨大的批,正好把冷启动的模型顶得更久。 + items, chars := batchLimitsFor(e.timeout) + 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 +} + +// requestTimeout 返回本次调用该用的单次超时:拿到第一个成功响应之前用 +// warmupTimeout(冷启动:模型加载 / GPU 缓冲分配),之后一律用配置的 timeout。 +func (e *APIEmbedder) requestTimeout() time.Duration { + if e.warmed.Load() { + return e.timeout + } + return warmupTimeout +} + +// embedChunk 发**一次** /embeddings 请求。chunk 必须已经落在 batchLimitsFor 推出来 +// 的边界内(Embed 负责切),因此它的耗时上界不超过超时的一半。 +func (e *APIEmbedder) embedChunk(ctx context.Context, texts []string) ([]Vector, error) { + callCtx, cancel := context.WithTimeout(ctx, e.requestTimeout()) defer cancel() req, err := e.newEmbeddingRequest(callCtx, texts) @@ -186,6 +232,8 @@ func (e *APIEmbedder) Embed(ctx context.Context, texts []string) ([]Vector, erro msg := readErrorBody(resp.Body) return nil, fmt.Errorf("embedding: status %d: %s", resp.StatusCode, msg) } + // 服务端确实答了 ⇒ 冷启动这一关过了,后续调用回到常态超时。 + e.warmed.Store(true) data, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) if err != nil { diff --git a/internal/runtime/embedding/api_url_test.go b/internal/runtime/embedding/api_url_test.go index 25a78b1..7980b49 100644 --- a/internal/runtime/embedding/api_url_test.go +++ b/internal/runtime/embedding/api_url_test.go @@ -222,6 +222,10 @@ const reviewBudgetRef = 20 * time.Second // 审查总预算:否则一次卡死的 embedding 连接就能吃光整条审查的预算,排在后面的 // T1 灰区判定连执行机会都没有(2026-10-07 实测故障形态)。同时验证该超时确实 // 打在了发出去的请求上。 +// +// ⚠️ 冷启动那一次走的是 warmupTimeout(见 budget.go),因此这里**以稳态请求** +// (第二次及以后)为准;冷启动那一路由 budget_test.go 的 +// TestAPIEmbedder_WarmupFirstCallBudget 单独钉。 func TestDefaultRequestTimeout(t *testing.T) { t.Parallel() @@ -256,8 +260,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..ed33f6e --- /dev/null +++ b/internal/runtime/embedding/budget.go @@ -0,0 +1,129 @@ +package embedding + +import "time" + +// 本文件是**单次 /embeddings 调用的预算推演**:批次上限与单次超时必须由**同一个 +// 式子**得出。任一侧单独调整都会让「一次请求的实测耗时 < 单次超时」这条不变式 +// 静默失效 —— 而失效的后果不是报错,是**服务端还在算、客户端已经放弃**:被放弃的 +// 那一批仍占着本机 llama.cpp 的算力,后面每一个请求(包括写索引时的单条请求)都 +// 排在它后面等到自己的超时,于是整条记忆写入路径被拖住、召回全线降级。 +// +// ## 不变式 +// +// requestOverhead + 一次请求的字符总量 × perCharBudget ≤ DefaultRequestTimeout / callBudgetShare +// +// 左半是**推演出来的耗时上界**,右半是该超时下允许用掉的一半(另一半是这台机器的 +// 抖动余量)。批次上限由解这个不等式得到(见 DefaultBatchLimit / DefaultBatchChars)。 +// +// ## 实测依据(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 是**首次**调用(模型加载 / GPU 缓冲分配尚未发生)的墙钟上限。 + // 它**只在拿到第一个成功响应之前**生效,之后所有调用立刻回到 + // DefaultRequestTimeout —— 所以它不会让「真挂死的请求」长期拖住整轮审查, + // 又不让冷启动那一次注定失败(2026-10-13 部署后第一分钟生产日志里 + // `vector embed for index failed ... embed 1 texts ... context deadline exceeded` + // 就是这一路:模型还没热,5s 不够)。 + warmupTimeout = 30 * 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..3120e7d --- /dev/null +++ b/internal/runtime/embedding/budget_test.go @@ -0,0 +1,310 @@ +package embedding + +import ( + "context" + "encoding/json" + "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_WarmupFirstCallBudget 钉冷启动那一路:第一次调用(模型加载 / +// GPU 缓冲分配)拿的是 warmupTimeout,**只有第一次**;一旦拿到成功响应,后续请求 +// 立刻回到 DefaultRequestTimeout(真挂死的请求不会被 30s 拖住)。 +func TestAPIEmbedder_WarmupFirstCallBudget(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.warmed.Load() { + t.Fatal("拿到成功响应后 warmed 应为真") + } + + 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..2b23451 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,12 +68,17 @@ 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 + // 如实上报**实际发生**的来源:空串 = 没有任何分区真正算过相似度(「没候选」, + // 与「向量不可用」不是一回事)。 + out.RecallMode = ledger.label() out.DurationMS = t.now().Sub(started).Milliseconds() if len(failures) > 0 { return out, fmt.Errorf("memory/tidy: dry-run finished with errors: %w", errors.Join(failures...)) @@ -80,8 +89,8 @@ func (t *Tidier) DryRun(ctx context.Context) (TidyDryRun, error) { // 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) +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 { t.log.Warn("memory tidy dry-run recall unavailable", "partition", p.label(), "items", len(items), "error", err) diff --git a/internal/runtime/memory/tidy_recall.go b/internal/runtime/memory/tidy_recall.go index 99e3d64..65973e1 100644 --- a/internal/runtime/memory/tidy_recall.go +++ b/internal/runtime/memory/tidy_recall.go @@ -25,8 +25,75 @@ 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 +} + +// 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++ + } +} + +// label 返回本轮**实际**的召回来源: +// - "":没有任何分区真正算过相似度(没有候选可比)—— 「没候选」不是「降级」; +// - embedding / lexical:所有算过的分区都走同一条路; +// - mixed:两种都有(多分区里有一部分降级)。 +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 +} + // tidyPair 是一条被召回的候选对。 type tidyPair struct { a, b MemoryItem @@ -35,18 +102,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 +129,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 +278,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..28be31e --- /dev/null +++ b/internal/runtime/memory/tidy_recall_mode_test.go @@ -0,0 +1,185 @@ +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_RecallModeIsActualNotConfigured 是**试算轮**(面板那一条)的同口径 +// 回归:配了 embedder 但向量不可用 ⇒ 试算如实报错(严格模式),且返回的 RecallMode +// **不**写成 embedding。 +func TestTidyDryRun_RecallModeIsActualNotConfigured(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, _ := newLogCaptureTidier(t, store, tidyTestCfg(0.80), emb) + out, err := tr.DryRun(ctx) + if !errors.Is(err, ErrTidyRecallUnavailable) { + t.Fatalf("试算轮必须如实上报不可用,err = %v", err) + } + if out.RecallMode == tidyRecallEmbedding { + t.Fatalf("向量不可用却把 RecallMode 报成 embedding(配置态口径):%q", out.RecallMode) + } +} + +// 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 "" +}