xiaoxue_admin/memoryweave
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(recall): 召回契约字段 + trace 回溯 + 生命周期/实体质量判据（工作树收口）

Browse Source 
这批是 10-26~27 已在生产二进制（zhiyid-protectrails-20261027）里运行、
但一直留在工作树未提交的改动。收口原因：三件改造要开分支，工作树脏 =
分支基准与回滚点都不存在。

主要内容：
- api/routes/contract.go + trace.go：召回响应契约字段（trace_id/status/scope/
  continuation/channel）与 /api/v1/trace 回溯端点
- governance/entity_quality.go：实体质量闸
- storage/trace.go + recall.go/recall_write_buffer.go：召回写入缓冲与轨迹
- 测试：contract/recall_contract_fields/recall_lifecycle/stale_lifecycle/
  entity_quality/trace

同时修一个 flaky：recall_lifecycle_test 原本硬编码「今天 11:00 已过」，
凌晨跑就会把「已失效」fixture 当成仍有效。改成相对时间（过期=now-1h），
生命周期判据与运行时刻无关。修完 go test ./internal/api/routes ./internal/storage 全绿。
This commit is contained in:
小唯
2026-10-28 02:09:19 +08:00
parent 0b2b4f01d8
commit 7f7171ac14
26 changed files with 2032 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/internal/api/routes/contract.go
+119
View File
@@ -0,0 +1,119 @@
// 织忆 MemoryWeave — 召回响应契约辅助（P0-3，2026-10-26）
//
// 背景（CS249r Vol III Ch6 检索契约）：响应只有 cache_hit/candidates_scanned/count/mode/graph_expanded，
// 空结果分不清「不存在」与「扫了一半就超时」——这是 Air Canada 类静默幻觉的结构性入口。
//
// 本文件提供最小可用契约三件：
// ① status/partial_reason：complete 的空结果才允许解读为「不存在」；
// ② scope{namespace,snapshot,...}：说明这次答案基于哪个数据视图（本服务无索引版本号，
// 故 snapshot = 观测时点 + 缓存状态，缓存命中代表 60s 内旧快照 —— 语义诚实，不假装有版本号）；
// ③ continuation：不透明翻页令牌（query 指纹 + ns 绑定，防止跨 query 串页）。
package routes
import (
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"strconv"
"time"
"github.com/xiaoxue/memoryweave/internal/models"
"github.com/xiaoxue/memoryweave/internal/storage"
)
// maxContinuationOffset 硬上限：防止把 topK 放大到爆内存/延迟。
const maxContinuationOffset = 500
type pageToken struct {
V int `json:"v"`
Q string `json:"q"` // query 指纹（sha256 前 8 hex）
NS string `json:"ns"` // 作用域绑定
Off int `json:"o"` // 下一页起点
Lim int `json:"l"` // 页大小
// Prev 已返回过的 ID（去重锚点）。活索引下排序会随 fetchLimit 变化而轻微漂移，
// 只靠 offset 会出现跨页重复；token 携带已返回 ID 才能机械保证「不重」。
Prev []string `json:"p,omitempty"`
}
// maxPrevIDs 令牌里最多携带的去重锚点数（3~5 页内足够；超长则丢最旧，令牌不会无限膨胀）。
const maxPrevIDs = 100
// queryFingerprint 8 hex 的 query 指纹（足够防串页，不泄露正文）。
func queryFingerprint(q string) string {
sum := sha256.Sum256([]byte(q))
return hex.EncodeToString(sum[:4])
}
// encodeContinuation 生成下一页令牌；没有下一页时返回空串。
func encodeContinuation(query, namespace string, offset, limit int, prev []string) string {
if limit <= 0 {
return ""
}
if len(prev) > maxPrevIDs {
prev = prev[len(prev)-maxPrevIDs:]
}
tok := pageToken{V: 1, Q: queryFingerprint(query), NS: namespace, Off: offset, Lim: limit, Prev: prev}
b, err := json.Marshal(tok)
if err != nil {
return ""
}
return base64.RawURLEncoding.EncodeToString(b)
}
// decodeContinuation 解析令牌。错误信息面向调用方（可据此修请求）。
func decodeContinuation(raw string) (pageToken, error) {
var tok pageToken
b, err := base64.RawURLEncoding.DecodeString(raw)
if err != nil {
return tok, errors.New("continuation 不是合法 base64url 令牌")
}
if err := json.Unmarshal(b, &tok); err != nil {
return tok, errors.New("continuation 内容不是合法 JSON")
}
if tok.V != 1 {
return tok, errors.New("continuation 版本不支持：" + strconv.Itoa(tok.V))
}
if tok.Off < 0 || tok.Lim <= 0 {
return tok, errors.New("continuation 字段非法（offset/limit）")
}
if tok.Off > maxContinuationOffset {
return tok, errors.New("continuation offset 超出上限 " + strconv.Itoa(maxContinuationOffset))
}
return tok, nil
}
// buildScope 组装本次召回的数据视图声明。
func buildScope(namespace string, topK int, mode string, cacheHit, truncated bool, started time.Time) map[string]interface{} {
snap := "t" + strconv.FormatInt(started.UnixMilli(), 10)
if cacheHit {
snap += "-cache" // 缓存命中：读的是 60s 内旧快照，不是本次实时扫描
} else {
snap += "-scan"
}
if truncated {
snap += "-truncated"
}
return map[string]interface{}{
"namespace": namespace,
"snapshot": snap,
"cache_hit": cacheHit,
"truncated": truncated,
"top_k": topK,
"mode": mode,
}
}
// traceItems 把最终返回集映射成 trace 的 item 归因（rank 从 1 起）。
func traceItems(results []models.RecallResult) []storage.TraceItem {
items := make([]storage.TraceItem, 0, len(results))
for i, r := range results {
ch := r.Channel
if ch == "" {
ch = "vector" // 老调用方/旧缓存条目无 channel ⇒ 主通道默认值
}
items = append(items, storage.TraceItem{Rank: i + 1, ID: r.ID, Channel: ch, Score: r.Score})
}
return items
}
go/internal/api/routes/contract_test.go
+116
View File
@@ -0,0 +1,116 @@
package routes
import (
"encoding/json"
"testing"
"time"
"github.com/xiaoxue/memoryweave/internal/models"
)
func TestContinuationRoundTrip(t *testing.T) {
tok := encodeContinuation("织忆怎么召回", "hermes-main", 10, 5, []string{"mem_1", "mem_2"})
if tok == "" {
t.Fatal("有下一页时令牌不应为空")
}
got, err := decodeContinuation(tok)
if err != nil {
t.Fatalf("解码失败：%v", err)
}
if got.Off != 10 || got.Lim != 5 {
t.Fatalf("令牌字段错误：%+v", got)
}
if got.Q != queryFingerprint("织忆怎么召回") {
t.Fatalf("query 指纹不一致")
}
if got.NS != "hermes-main" {
t.Fatalf("namespace 不一致：%s", got.NS)
}
if len(got.Prev) != 2 || got.Prev[0] != "mem_1" {
t.Fatalf("去重锚点未随令牌传递：%v", got.Prev)
}
}
func TestContinuationRejectsBadToken(t *testing.T) {
if _, err := decodeContinuation("not-base64!!"); err == nil {
t.Fatal("非法 base64 应报错")
}
if _, err := decodeContinuation("eyJ2Ijo5fQ"); err == nil { // {"v":9}
t.Fatal("未知版本应报错")
}
tok := encodeContinuation("q", "ns", 9999, 5, nil)
if _, err := decodeContinuation(tok); err == nil {
t.Fatal("offset 超上限应报错")
}
}
func TestQueryFingerprintStable(t *testing.T) {
a := queryFingerprint("同样的问句")
b := queryFingerprint("同样的问句")
c := queryFingerprint("不同的问句")
if a != b {
t.Fatal("同 query 指纹应稳定")
}
if a == c {
t.Fatal("不同 query 指纹应不同")
}
if len(a) != 8 {
t.Fatalf("指纹应为 8 hex，实际 %d", len(a))
}
}
func TestBuildScope(t *testing.T) {
started := time.UnixMilli(1700000000000)
sc := buildScope("hermes-main", 5, "hybrid", false, false, started)
for _, k := range []string{"namespace", "snapshot", "cache_hit", "truncated", "top_k", "mode"} {
if _, ok := sc[k]; !ok {
t.Fatalf("scope 缺字段 %s", k)
}
}
snap := sc["snapshot"].(string)
if snap != "t1700000000000-scan" {
t.Fatalf("实时扫描 snapshot 形态错误：%s", snap)
}
cached := buildScope("ns", 5, "hybrid", true, true, started)["snapshot"].(string)
if cached != "t1700000000000-cache-truncated" {
t.Fatalf("缓存+截断 snapshot 形态错误：%s", cached)
}
}
func TestTraceItems(t *testing.T) {
rs := []models.RecallResult{
{ID: "a", Channel: "vector"},
{ID: "b"}, // 老调用方无 channel ⇒ 默认 vector
{ID: "c", Channel: "graph"},
}
items := traceItems(rs)
if len(items) != 3 || items[0].Rank != 1 || items[2].Rank != 3 {
t.Fatalf("rank 编号错误：%+v", items)
}
if items[1].Channel != "vector" {
t.Fatalf("缺省 channel 应回落 vector，实际 %s", items[1].Channel)
}
}
// TestRecallResultContractFields 契约字段必须真的出现在 JSON 里（不是「加了结构体没接线」）。
func TestRecallResultContractFields(t *testing.T) {
r := models.RecallResult{
ID: "mem_1", Channel: "fts", Stale: true, Version: 3,
ValidFrom: "2026-10-26T10:00:00+08:00", ValidTo: "2026-10-26T11:00:00+08:00",
Supersedes: []string{"mem_0"}, SupersededBy: []string{"mem_2"}, Superseded: true,
}
b, err := json.Marshal(r)
if err != nil {
t.Fatalf("marshal: %v", err)
}
var m map[string]interface{}
if err := json.Unmarshal(b, &m); err != nil {
t.Fatalf("unmarshal: %v", err)
}
for _, k := range []string{"channel", "stale", "generation", "valid_from", "valid_to",
"supersedes", "superseded_by", "superseded"} {
if _, ok := m[k]; !ok {
t.Fatalf("响应缺字段 %s（JSON=%s）", k, string(b))
}
}
}
go/internal/api/routes/core.go
+207 -60
View File
@@ -2,6 +2,7 @@
package routes
import (
"context"
"encoding/json"
"fmt"
"log"
@@ -508,13 +509,15 @@ func cosineSimilarity(a, b []float32) float64 {
// 用途：区分「真无匹配」（results=[] 但候选数>0）与「依赖故障」（非 2xx）。
func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
var req struct {
Query string `json:"query"`
Limit int `json:"limit"`
TopK int `json:"top_k"` // 兼容插件 top_k 参数
Namespace string `json:"namespace"`
AgentID string `json:"agent_id"` // 用于推导默认 namespace
Diversity float64 `json:"diversity"`
Mode string `json:"mode"` // "hybrid"(default), "semantic", "keyword"
Query string `json:"query"`
Limit int `json:"limit"`
TopK int `json:"top_k"` // 兼容插件 top_k 参数
Namespace string `json:"namespace"`
AgentID string `json:"agent_id"` // 用于推导默认 namespace
Diversity float64 `json:"diversity"`
Mode string `json:"mode"` // "hybrid"(default), "semantic", "keyword"
DeadlineMS int `json:"deadline_ms"` // P0-3：本次召回的硬超时（毫秒），超时返回 partial
Continuation string `json:"continuation"` // P0-3：上一页返回的不透明翻页令牌
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
respondError(w, 400, "invalid body")
@@ -552,6 +555,43 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
nsDerivedFromAgent := req.Namespace == "" && req.AgentID != ""
req.Namespace = resolveRecallNamespace(req.Namespace, req.AgentID, req.Query)
// ── P0-3 契约 + P0-1 追踪（2026-10-26）──────────────────────────────
// 翻页令牌：绑定 query 指纹 + 作用域，跨 query 复用一律 400（防串页）。
offset := 0
var prevIDs []string
if req.Continuation != "" {
pt, perr := decodeContinuation(req.Continuation)
if perr != nil {
respondErrorSchema(w, 400, "bad_continuation", perr.Error(), 0)
return
}
if pt.Q != queryFingerprint(req.Query) {
respondErrorSchema(w, 400, "continuation_query_mismatch",
"continuation 属于另一个 query，不能跨 query 复用", 0)
return
}
if pt.NS != req.Namespace {
respondErrorSchema(w, 400, "continuation_namespace_mismatch",
"continuation 的 namespace 与本次请求不一致", 0)
return
}
offset = pt.Off
prevIDs = pt.Prev
}
// 取数上限 = 页尾（offset+limit）；MMR 的 k 随之上抬，保证前缀稳定。
// 多取 len(prevIDs)：用于过滤跨页重复项，过滤后仍能凑满一页（否则「不重」要靠「漏」换）。
fetchLimit := req.Limit + offset + len(prevIDs)
if fetchLimit > maxContinuationOffset+req.Limit {
fetchLimit = maxContinuationOffset + req.Limit
}
ctx := r.Context()
if req.DeadlineMS > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, time.Duration(req.DeadlineMS)*time.Millisecond)
defer cancel()
}
// H4: Default diversity — balance relevance & diversity
// 2026-10-14：改读 storage 的推送值（默认 0.3，与下方原硬编码一致 ⇒ 行为不变），
// 让调参器能真正影响召回 —— 此前调参改了 Tuner.Diversity 但没人读（空转）。
@@ -579,72 +619,68 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
if req.Mode == "" {
req.Mode = "hybrid"
}
// P0-1：trace 在 mode 归一之后创建（记录真实 mode）；所有路径共用同一条 trace。
tr := storage.NewRecallTrace("", req.Query, req.Namespace, req.Mode)
var lastMeta storage.RecallMeta
switch req.Mode {
case "keyword":
// Pure keyword search: LanceDB + BM25 re-rank, or graph.db fallback
// t_2d4e3b92: keyword 主路径依赖向量编码；编码失败（依赖不可用）⇒ 非 2xx，
// 不再把错误伪装成 200 + 错误体。
// t_1c12e000：keyword 路径同样收进 RecallWithMeta（Pipeline.Recall 内部
// 走同一缓存键），只为把 cache_hit 带回响应——旧版用 Recall 丢掉了 meta。
var kwMeta storage.RecallMeta
kwResults, kwMeta, kwErr := a.Pipeline.RecallWithMetaOpts(req.Query, req.Namespace, req.Limit, req.Diversity, nsDerivedFromAgent)
// t_2d4e3b92: keyword 主路径依赖向量编码；编码失败（依赖不可用）⇒ 非 2xx。
stKw := time.Now()
kwResults, kwMeta, kwErr := a.Pipeline.RecallWithMetaTraced(ctx, req.Query, req.Namespace, fetchLimit, req.Diversity, nsDerivedFromAgent, tr)
if kwErr != nil {
// 主路径硬错误（向量编码/rerank 依赖不可用）：即便兜底能补结果，也按 502 上报，
// 调用方据此走失败路径而不是把依赖故障误判成"无匹配"
tr.StageNote("keyword_path", stKw, 0, kwErr.Error())
tr.Error = kwErr.Error()
respondErrorSchema(w, 502, "recall_backend_error",
"keyword recall failed: "+kwErr.Error(), 5)
log.Printf("[recall:keyword] failed: %v", kwErr)
return
}
lastMeta = kwMeta
cacheHit = kwMeta.CacheHit
truncated = kwMeta.Truncated // t_3aa2a8c1
truncated = kwMeta.Truncated
if len(kwResults) == 0 {
// 真无匹配（正常业务）：成功 schema，results=[] + candidates_scanned 供区分
fallbackResults := a.GraphStore.FallbackTextSearch(req.Query, req.Namespace, req.Limit)
results = convertFallbackResults(fallbackResults)
respond(w, 200, map[string]interface{}{
"results": results,
"count": len(results),
"candidates_scanned": 0, // keyword 路径未做 ANN 扫描
"mode": "keyword",
"cache_hit": false, // graph 兜底结果不来自 search cache
})
return
// 真无匹配（正常业务）或纯词面路径空：graph.db 兜底
results = convertFallbackResults(a.GraphStore.FallbackTextSearch(req.Query, req.Namespace, fetchLimit))
if len(results) > 0 {
tr.StageNote("keyword_fallback", stKw, len(results), "主路径 0 命中 ⇒ graph.db 兜底")
} else {
tr.Stage("keyword_path", stKw, 0)
}
cacheHit = false // 兜底结果不来自 search cache
} else {
// Re-rank by BM25 only
for i := range kwResults {
kwResults[i].Score = storage.ComputeBM25Score(req.Query, kwResults[i].Content)
}
sort.Slice(kwResults, func(i, j int) bool { return kwResults[i].Score > kwResults[j].Score })
if len(kwResults) > fetchLimit {
kwResults = kwResults[:fetchLimit]
}
for i := range kwResults {
if kwResults[i].Channel == "" {
kwResults[i].Channel = "keyword"
}
}
results = kwResults
tr.Stage("keyword_path", stKw, len(results))
}
// Re-rank by BM25 only
for i := range kwResults {
kwResults[i].Score = storage.ComputeBM25Score(req.Query, kwResults[i].Content)
}
sort.Slice(kwResults, func(i, j int) bool { return kwResults[i].Score > kwResults[j].Score })
if len(kwResults) > req.Limit {
kwResults = kwResults[:req.Limit]
}
kwResp := map[string]interface{}{
"results": kwResults, "count": len(kwResults), "mode": "keyword",
"candidates_scanned": 0,
"cache_hit": cacheHit, // t_1c12e000：keyword 主路径命中 search cache 时显式标注
}
// t_3aa2a8c1：truncated 只在 true 时暴露（省略 = 未截断，omitempty 语义）。
if truncated {
kwResp["truncated"] = true
}
respond(w, 200, kwResp)
return
case "semantic":
// Pure semantic — BM25 off (H1 skipped, Pipeline Recall has BM25 built in — handled by switch)
// t_949840b1：RecallWithMeta 额外返回候选池真实大小（ANN 粗检索池，非最终产出数）
// Pure semantic — BM25 off（Pipeline 内部按 mode 处理）
var meta storage.RecallMeta
results, meta, err = a.Pipeline.RecallWithMetaOpts(req.Query, req.Namespace, req.Limit, req.Diversity, nsDerivedFromAgent)
results, meta, err = a.Pipeline.RecallWithMetaTraced(ctx, req.Query, req.Namespace, fetchLimit, req.Diversity, nsDerivedFromAgent, tr)
lastMeta = meta
annScanned = meta.CandidatesScanned
cacheHit = meta.CacheHit // t_1c12e000
truncated = meta.Truncated // t_3aa2a8c1
cacheHit = meta.CacheHit
truncated = meta.Truncated
case "hybrid":
// BM25 + vector combined (H1: BM25 scoring inside Pipeline.Recall)
var meta storage.RecallMeta
results, meta, err = a.Pipeline.RecallWithMetaOpts(req.Query, req.Namespace, req.Limit, req.Diversity, nsDerivedFromAgent)
results, meta, err = a.Pipeline.RecallWithMetaTraced(ctx, req.Query, req.Namespace, fetchLimit, req.Diversity, nsDerivedFromAgent, tr)
lastMeta = meta
annScanned = meta.CandidatesScanned
cacheHit = meta.CacheHit // t_1c12e000
truncated = meta.Truncated // t_3aa2a8c1
cacheHit = meta.CacheHit
truncated = meta.Truncated
}
if err != nil {
// P0: 降级到 graph.db 关键词搜索（仅 hybrid/semantic 主路径依赖向量/rerank）
@@ -657,6 +693,8 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
w.Header().Set("X-Fallback", "graph")
} else {
// t_2d4e3b92: 主路径与兜底都失败 ⇒ 非 2xx + 统一错误体（绝不在 200 里塞错误）
tr.MarkPartial("stage_error")
tr.Error = err.Error()
respondErrorSchema(w, 502, "recall_backend_error",
"recall failed: "+err.Error()+"; graph fallback also empty", 5)
log.Printf("[recall] failed: %v; fallback empty", err)
@@ -705,6 +743,13 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
QualityScore: m.QualityScore,
VValue: vv,
HumanUsefulCount: m.HumanUsefulCount, // A10 (t_d618dc7c)
Channel: "shared", // P0-1：来源通道
Version: m.Version,
ValidFrom: m.ValidFrom,
ValidTo: m.ValidTo,
Supersedes: m.Supersedes,
SupersededBy: m.SupersededBy,
Superseded: len(m.SupersededBy) > 0,
})
}
}
@@ -749,6 +794,13 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
QualityScore: m.QualityScore,
VValue: vv,
HumanUsefulCount: m.HumanUsefulCount, // A10 (t_d618dc7c)
Channel: "wiki", // P0-1：来源通道
Version: m.Version,
ValidFrom: m.ValidFrom,
ValidTo: m.ValidTo,
Supersedes: m.Supersedes,
SupersededBy: m.SupersededBy,
Superseded: len(m.SupersededBy) > 0,
})
}
}
@@ -778,6 +830,28 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
results = ordered
}
// ── 2026-10-26（P1-3 生命周期 fixture 抓到的真 bug）──────────────────
// 主候选池在 pipeline 里已做 bi-temporal 过滤，但 **shared / wiki 补充通道与图谱注入
// 不过那道闸** ⇒ 已失效事实会从补充通道回流进 top-k（实测：valid_to=11:00 的记录
// 被 shared 通道以 score=0.5 重新塞回）。这里在「所有注入完成之后、cap 之前」统一过滤，
// 保证读侧只有一条有效期口径。全失效时保留原结果（不把召回清空，non-lossy）。
if storage.BitemporalEnabled() {
now := time.Now()
live := make([]models.RecallResult, 0, len(results))
expired := 0
for _, r := range results {
if storage.ValidityExpired(r.ValidTo, now) {
expired++
continue
}
live = append(live, r)
}
if len(live) > 0 && expired > 0 {
log.Printf("[recall] 补充通道 bi-temporal 过滤 %d 条失效事实（统一读侧口径）", expired)
results = live
}
}
// Record recall hit rate (has results = hit, empty = miss)
selfoptimize.Dash.RecordRecall(len(results) > 0)
// Gap auto-close: recall 命中后自动关闭该 query 对应的 open gap
@@ -829,10 +903,76 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
// 主结果不足 req.Limit 时用扩展结果按原顺序补齐（稀疏查询仍有扩展收益）。
// graph_expanded = 扩展结果**单独成字段**（上限 req.Limit）——能力仍可观测、可统计，
// 但不再挤占主结果、不再污染 results/count。
results, graphExpanded := capRecallResults(results, req.Limit)
results, graphExpanded := capRecallResults(results, fetchLimit)
if results == nil {
results = []models.RecallResult{}
}
// ── P0-3 翻页：从 fetchLimit 结果里切出 [offset, offset+limit) ──
// 说明：翻页重算整段前 offset+limit 条再切片，保证 MMR 前缀稳定；
// 令牌绑定 query 指纹 + namespace，跨 query/跨域复用一律 400（防串页）。
totalFetched := len(results)
if offset > 0 {
if offset >= len(results) {
results = []models.RecallResult{}
} else {
results = results[offset:]
}
}
if len(prevIDs) > 0 {
seenPrev := make(map[string]bool, len(prevIDs))
for _, id := range prevIDs {
seenPrev[id] = true
}
kept := results[:0]
for _, r := range results {
if !seenPrev[r.ID] {
kept = append(kept, r)
}
}
results = kept
}
if len(results) > req.Limit {
results = results[:req.Limit]
}
nextCont := ""
if totalFetched >= fetchLimit && len(results) == req.Limit && offset+req.Limit < maxContinuationOffset {
returned := make([]string, 0, len(prevIDs)+len(results))
returned = append(returned, prevIDs...)
for _, r := range results {
returned = append(returned, r.ID)
}
nextCont = encodeContinuation(req.Query, req.Namespace, offset+req.Limit, req.Limit, returned)
}
// ── P0-1/P0-3：契约字段 + trace 落盘 ──
tr.SetItems(traceItems(results))
status := lastMeta.Status
if status == "" {
status = "complete"
}
partialReason := lastMeta.PartialReason
if tr.Status == "partial" && status == "complete" {
status = "partial"
partialReason = tr.PartialReason
}
if truncated && status == "complete" {
status = "partial"
partialReason = "row_cap_truncated"
}
tr.Status = status
tr.PartialReason = partialReason
tr.CacheHit = cacheHit
tr.Truncated = truncated
tr.CandidatesScanned = annScanned
tr.Namespace = req.Namespace
tr.Mode = req.Mode
tr.Scope = buildScope(req.Namespace, req.Limit, req.Mode, cacheHit, truncated, tr.StartedAt)
if offset > 0 {
tr.Scope["offset"] = offset
}
storage.SaveTrace(tr)
resp := map[string]interface{}{
"results": results,
"count": len(results),
@@ -842,14 +982,21 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
// 键恒存在（可为空数组），便于探针/统计区分「无扩展」与「字段缺失」。
"graph_expanded": graphExpanded,
// t_1c12e000：AC3 —— 缓存命中/回源扫表必须可区分。
// true ⇒ candidates_scanned=0 是「读了 60s 内的旧快照」（含 L1/L2）；
// false ⇒ 本次真扫了表（candidates_scanned 为真实候选池大小）。
"cache_hit": cacheHit,
// 2026-10-26（P0-1/P0-3 契约）：
// trace_id ⇒ GET /api/v1/trace/<id> 可重建本次调用链（阶段耗时 + 每通道贡献）；
// status ⇒ complete 的空结果才可解读为「不存在」；partial 必须看 partial_reason；
// scope ⇒ 本次答案基于哪个数据视图（namespace/snapshot/cache_hit/truncated）；
// continuation ⇒ 空串=无下一页，非空=拼回请求体取下一页。
"trace_id": tr.TraceID,
"status": status,
"scope": tr.Scope,
"continuation": nextCont,
}
if partialReason != "" {
resp["partial_reason"] = partialReason
}
// t_3aa2a8c1：truncated —— 只在 true 时暴露（键缺失 = 未截断，等价 omitempty 语义）。
// true = 本次 recall 的粗检索/补充搜至少一次命中 Rust 侧单次重读行数上限
// （默认 20000，ZHIYI_HEAVY_READ_MAX_ROWS）被截断为部分结果 ⇒ count/
// candidates_scanned 是截断后状态，调用方不得当作「全量」。
if truncated {
resp["truncated"] = true
}
go/internal/api/routes/recall_contract_fields_test.go
+67
View File
@@ -0,0 +1,67 @@
package routes
// 2026-10-26（P1-1）：supersede / 版本字段必须真的出现在 /recall 响应里。
// 用同包 fakeLanceDB 构造一条「已被取代 + 有效期已过 + 版本 2」的记录：
// - 这是「写侧有、读侧看不见」的复发点（t_4c810e02「加列不接 IPC=静默 no-op」同款坑）；
// - 判据只看 JSON 响应键值，不看结构体注释。
import (
"testing"
"github.com/xiaoxue/memoryweave/internal/models"
)
func TestRecall_SupersedeAndValidityFieldsExposed(t *testing.T) {
db := &fakeLanceDB{searchResults: []models.MemoryRecord{
{
ID: "mem_old_fact",
Content: "织忆默认 stale 阈值是 720 小时（旧事实）",
Category: "distilled",
Version: 2,
ValidFrom: "2026-10-26T10:00:00+08:00",
ValidTo: "2026-10-26T11:00:00+08:00",
SupersededBy: []string{"mem_new_fact"},
},
{ID: "mem_fresh", Content: "织忆默认 stale 阈值是 720 小时（现行事实）", Category: "distilled"},
}}
code, m, body := recallBody(t, db, `{"query":"stale 阈值多少小时","limit":5,"namespace":"hermes-main"}`)
if code != 200 {
t.Fatalf("expected 200, got %d; body=%s", code, body)
}
results, _ := m["results"].([]interface{})
if len(results) == 0 {
t.Fatalf("无结果，无法验证字段：%s", body)
}
var found map[string]interface{}
for _, raw := range results {
r, _ := raw.(map[string]interface{})
if r["id"] == "mem_old_fact" {
found = r
break
}
}
if found == nil {
t.Fatalf("目标记录未被召回（fake DB 应原样返回），body=%s", body)
}
if found["superseded_by"] == nil {
t.Fatalf("superseded_by 未暴露给读侧：%v", found)
}
if found["superseded"] != true {
t.Fatalf("superseded 便捷布尔应为 true：%v", found["superseded"])
}
if found["generation"] != float64(2) {
t.Fatalf("generation 应为 2：%v", found["generation"])
}
if found["valid_to"] != "2026-10-26T11:00:00+08:00" {
t.Fatalf("valid_to 未直通：%v", found["valid_to"])
}
if found["channel"] == nil || found["channel"] == "" {
t.Fatalf("channel 未暴露：%v", found)
}
if _, ok := m["trace_id"]; !ok {
t.Fatalf("响应缺 trace_id：%s", body)
}
if m["status"] != "complete" {
t.Fatalf("正常召回 status 应为 complete，实际 %v", m["status"])
}
}
go/internal/api/routes/recall_lifecycle_test.go
+115
View File
@@ -0,0 +1,115 @@
package routes
// 2026-10-26（P1-3）：stale-return + 生命周期 fixture（10:00 生效 / 11:00 失效）。
//
// 评审缺口：「只有离线检索指标，没有 stale-return、没有生命周期 fixture」。
// 本测试把三条读侧生命周期行为锁成机械判据：
// ① bi-temporal 开启时，valid_to 已过的事实**不进 top-k**（但仍在库，non-lossy）；
// ② valid_to 在未来 / 为空 ⇒ 正常返回（同一 fixture 的对照，证明不是「整类被吞」）；
// ③ updated_at 超 720h ⇒ 响应 stale=true；新记录不输出该键。
//
// 时窗用真实语义：fixture 事实 10:00 生效、11:00 失效，查询时点取「当日 11:00 之后」。
import (
"testing"
"time"
"github.com/xiaoxue/memoryweave/internal/models"
)
// 时窗用**相对时间**：曾经写成 dayAt(10,0)/dayAt(11,0) 假设「跑测试时已是晚上 20:00」，
// 凌晨 02:07 跑就是 flaky —— 11:00 还没到，「已失效」的 fixture 其实仍有效。
// 生命周期判据必须与运行时刻无关：过期 = now-1h，有效期 = now+24h。
func expiredPair() (from, to, future string) {
now := time.Now()
return now.Add(-2 * time.Hour).Format(time.RFC3339),
now.Add(-1 * time.Hour).Format(time.RFC3339),
now.Add(24 * time.Hour).Format(time.RFC3339)
}
func idsOf(m map[string]interface{}) map[string]bool {
out := map[string]bool{}
if res, ok := m["results"].([]interface{}); ok {
for _, raw := range res {
if r, ok := raw.(map[string]interface{}); ok {
if id, ok := r["id"].(string); ok {
out[id] = true
}
}
}
}
return out
}
func TestRecall_Lifecycle_ExpiredFactDroppedWhenBitemporal(t *testing.T) {
from, to, future := expiredPair()
db := &fakeLanceDB{searchResults: []models.MemoryRecord{
{ID: "mem_expired_window", Content: "生命周期 fixture：10:00 生效 11:00 失效的事实",
Category: "distilled", ValidFrom: from, ValidTo: to},
{ID: "mem_still_valid", Content: "生命周期 fixture：仍在有效期内的事实",
Category: "distilled", ValidFrom: from, ValidTo: future},
}}
t.Setenv("ZHIYI_BITEMPORAL", "1")
code, m, body := recallBody(t, db, `{"query":"生命周期 fixture 有效期","limit":5,"namespace":"hermes-main"}`)
if code != 200 {
t.Fatalf("expected 200, got %d; body=%s", code, body)
}
ids := idsOf(m)
if ids["mem_expired_window"] {
t.Fatalf("11:00 已失效的事实不应进 top-k（bi-temporal 开）；body=%s", body)
}
if !ids["mem_still_valid"] {
t.Fatalf("有效期未过的事实必须正常返回（对照缺失 ⇒ 可能是整类被吞）；body=%s", body)
}
}
func TestRecall_Lifecycle_ExpiredFactVisibleWhenBitemporalOff(t *testing.T) {
// 开关关（默认）⇒ 行为零变化：过期事实照常返回，但 valid_to 字段可见（读侧能自己判）。
from, to, _ := expiredPair()
db := &fakeLanceDB{searchResults: []models.MemoryRecord{
{ID: "mem_expired_visible", Content: "生命周期 fixture：过期但可见", Category: "distilled",
ValidFrom: from, ValidTo: to},
}}
t.Setenv("ZHIYI_BITEMPORAL", "0")
code, m, body := recallBody(t, db, `{"query":"过期但可见","limit":5,"namespace":"hermes-main"}`)
if code != 200 {
t.Fatalf("expected 200, got %d; body=%s", code, body)
}
if !idsOf(m)["mem_expired_visible"] {
t.Fatalf("bi-temporal 关时过期事实应照常返回（non-lossy 可见性）；body=%s", body)
}
var found map[string]interface{}
for _, raw := range m["results"].([]interface{}) {
if r, ok := raw.(map[string]interface{}); ok && r["id"] == "mem_expired_visible" {
found = r
}
}
if found == nil || found["valid_to"] != to {
t.Fatalf("valid_to 必须直通给读侧：%v", found)
}
}
func TestRecall_StaleReturn_DefaultThreshold(t *testing.T) {
old := time.Now().Add(-31 * 24 * time.Hour)
db := &fakeLanceDB{searchResults: []models.MemoryRecord{
{ID: "mem_old", Content: "陈年事实（31 天未更新）", Category: "distilled", UpdatedAt: old},
{ID: "mem_new", Content: "新鲜事实（今天更新）", Category: "distilled", UpdatedAt: time.Now()},
}}
code, m, body := recallBody(t, db, `{"query":"事实 新鲜度","limit":5,"namespace":"hermes-main"}`)
if code != 200 {
t.Fatalf("expected 200, got %d; body=%s", code, body)
}
got := map[string]map[string]interface{}{}
for _, raw := range m["results"].([]interface{}) {
if r, ok := raw.(map[string]interface{}); ok {
got[r["id"].(string)] = r
}
}
if got["mem_old"] == nil || got["mem_old"]["stale"] != true {
t.Fatalf("31 天未更新的记忆应 stale=true：%v", got["mem_old"])
}
if _, ok := got["mem_new"]["stale"]; ok {
t.Fatalf("新记忆不应输出 stale 键：%v", got["mem_new"])
}
}
go/internal/api/routes/trace.go
+49
View File
@@ -0,0 +1,49 @@
// 织忆 MemoryWeave — 召回链路追踪端点（P0-1，2026-10-26）
//
// GET /api/v1/trace/<trace_id>
//
// 返回该次召回的：各阶段耗时（encode/coarse_search/fts_recall/bm25/rerank_fusion/mmr_select/
// graph_expand/dedupe）、各通道贡献条数（channels）、每条结果的来源归因（items）、
// status/partial_reason、scope、cache_hit/truncated。
//
// 数据来源：内存环形缓冲（最近 2000 条）+ JSONL 落盘（/var/lib/memoryweave/traces/），
// 服务重启后仍可按 trace_id 回溯（落盘扫描）。
package routes
import (
"context"
"net/http"
"strings"
"time"
"github.com/xiaoxue/memoryweave/internal/storage"
)
// ── P0-3 召回硬超时：取消函数类型 + context.WithTimeout 包装 ──────────────
// core.go Recall() 在 req.DeadlineMS>0 时需要带类型标记的取消函数，以便 defer 取消。
// 2026-10-26 (t_e98cb1d6)：此处补齐使 routes 包可编译（否则 routes 全包不可构建，
// capRecallResults 的单测无法跑通）。这是缺失的声明，不是功能实现；P0-3 的 trace 完整
// 接线（tr.Stage 埋点、tr 落盘）仍待 P0-3 本任务补齐。
type ctxDeadlineFunc context.CancelFunc
func contextWithTimeout(ctx context.Context, timeout time.Duration) (context.Context, ctxDeadlineFunc) {
c, cancel := context.WithTimeout(ctx, timeout)
return c, ctxDeadlineFunc(cancel)
}
// Trace 查询单条召回链路。路径形如 /api/v1/trace/tr_xxxx。
func (a *API) Trace(w http.ResponseWriter, r *http.Request) {
id := strings.TrimSpace(strings.TrimPrefix(r.URL.Path, "/api/v1/trace/"))
if id == "" {
respondErrorSchema(w, 400, "trace_id_required", "用法：GET /api/v1/trace/<trace_id>", 0)
return
}
tr, ok := storage.GetTrace(id)
if !ok {
// 404 与「trace 过期/被环形缓冲覆盖」同义：调用方据此走「重新召回」而不是当成系统故障。
respondErrorSchema(w, 404, "trace_not_found",
"trace 不存在或已被环形缓冲覆盖（内存保留最近 2000 条，落盘按天分文件）", 0)
return
}
respond(w, 200, tr)
}
go/internal/api/server.go
+4 -1
View File
@@ -187,7 +187,8 @@ func NewServer() http.Handler {
// ─── t_d32e5f8a L2 模式检测（MemOS 血统）─────────────────────────
// 复活 L1→L2 pattern 蒸馏（原 Python daemon main_loop 退役后停摆）：
// pattern 节点存 graph_nodes(type='pattern')，与现有 85 存量同表，不另开新表。
// pattern 节点存 graph_nodes(type='pattern')，与旧 daemon 残留的 85 行
// type='pattern_stat' 同表但不同类，不另开新表。
// modernc sqlite 短命单发开/关（UpsertPatterns/ListPatterns 开一次用一次即关），
// 不持长连接、不碰 state.db —— 规避当年退役根因。
// 召回注入 env 门控 ZHIYI_PATTERN_RECALL，默认 0 ⇒ 不注册 Provider，行为零变化。
@@ -602,6 +603,8 @@ func NewServer() http.Handler {
})
mux.HandleFunc("/api/v1/recall", api.Recall)
mux.HandleFunc("/api/v1/recall/debug", api.RecallDebug)
// 2026-10-26 (P0-1 t_trace_p0)：召回链路追踪回溯（trace_id 贯穿各阶段/各通道）
mux.HandleFunc("/api/v1/trace/", api.Trace)
// t_23b6a940 AC7 热门度看门狗快照（report-only，传感器端读它转既有告警通道）
mux.HandleFunc("/api/v1/popwatch", api.PopWatch)
// 2026-10-13 (t_e9a4d18f P2) 常设答案直读/写入
go/internal/governance/entity_quality.go
+142
View File
@@ -0,0 +1,142 @@
// 织忆 MemoryWeave — 图谱实体质量闸（P0-4，2026-10-26）
//
// 背景（CS249r 评审实测）：graph_expanded 抽样 15 条里 8 条是 ≤4 字垃圾实体
// （"条""里存着""开发"），既污染上下文又抬延迟。根因是实体抽取「任意 2-8 字中文串都当实体」，
// 且扩展前不检查节点在图里的度数 —— 孤边节点（度=1）被当成语义邻居注入。
//
// 两道闸（都在**导航前**执行，省算力）：
// ① isJunkEntity：长度 <2 rune / 纯虚词量词代词 / 常见填充短语 / 纯 ASCII 停用词 ⇒ 丢弃；
// ② 度数闸：起点实体在图中的边数 <2 ⇒ 不导航、不注入（孤边不构成「关联」）。
//
// 观测：两道闸的拒绝计数写入 debug 日志（ZHIYI_GRAPH_EXPAND_DEBUG=1）与 recall trace
// （graph_expand 阶段 note），保证「下次坏了不沉默」。
package governance
import (
"strings"
"unicode/utf8"
)
// junkEntitySet 高置信垃圾实体词表：中文虚词/代词/量词/时间副词/填充短语 + 英文停用词。
// 原则：只收**几乎不可能**是真实实体名的词；宁可漏筛（由度数闸兜底），不可误杀专名。
var junkEntitySet = map[string]struct{}{}
func init() {
words := []string{
// 虚词 / 助词 / 连词
"的", "了", "着", "过", "地", "得", "所", "以", "于", "与", "及", "或", "而", "且", "则", "即",
"因为", "所以", "但是", "而且", "然后", "如果", "虽然", "不过", "而是", "就是", "只是", "还是",
"并且", "以及", "或者", "否则", "因此", "于是", "关于", "对于", "由于", "根据", "按照", "通过",
// 代词 / 指示词
"我", "你", "他", "她", "它", "我们", "你们", "他们", "她们", "它们", "自己", "别人", "大家",
"这个", "那个", "这些", "那些", "这样", "那样", "这种", "那种", "这里", "那里", "里面", "上面",
"下面", "前面", "后面", "其中", "其他", "其它", "什么", "怎么", "怎样", "为什么", "哪个", "哪些",
// 数量 / 量词 / 程度
"一个", "一条", "一次", "一些", "一种", "一件", "一点", "一下", "一直", "一样", "一起",
"两个", "三个", "几个", "多个", "很多", "更多", "多条", "几条", "每条", "各种", "某种", "全部", "所有", "每个", "各种", "部分",
"一条条", "一条里", "条", "个", "次", "些", "种", "只", "件", "篇", "位", "名", "层",
"非常", "比较", "特别", "十分", "更加", "最为", "可能", "应该", "必须", "可以", "能够", "需要",
// 常见谓词 / 填充动词短语（低信息量，几乎不可能是实体）
"里存着", "存着", "里存在", "有着", "存在", "进行", "使用", "支持", "包括", "包含", "属于",
"作为", "成为", "变成", "做到", "做好", "来看", "来说", "来讲", "看到", "看看", "发现",
"出现", "提供", "实现", "完成", "开始", "结束", "继续", "返回", "调用", "处理", "执行",
"开发", "编写", "添加", "修改", "删除", "更新", "检查", "确认", "验证", "测试", "部署",
// 时间 / 方位
"现在", "今天", "昨天", "明天", "之前", "之后", "以前", "以后", "当时", "当前", "目前",
"最近", "刚才", "马上", "已经", "正在", "曾经", "从来", "后来", "最终", "最后", "首先",
"时候", "时间", "期间", "上面", "下面", "左边", "右边", "中间", "附近", "里面", "外面",
// 英文停用词（小写比较）
"the", "and", "for", "are", "but", "not", "you", "all", "any", "can", "had", "her", "was",
"one", "our", "out", "day", "get", "has", "him", "his", "how", "its", "new", "now", "old",
"see", "two", "way", "who", "boy", "did", "she", "use", "her", "way", "may", "say",
"this", "that", "with", "from", "have", "will", "your", "they", "than", "then", "them",
"into", "more", "some", "such", "only", "over", "also", "just", "like", "make", "made",
}
for _, w := range words {
junkEntitySet[w] = struct{}{}
}
}
// IsJunkEntity 判定实体是否应被图谱扩展丢弃（P0-4 第①道闸）。
func IsJunkEntity(e string) bool {
e = strings.TrimSpace(e)
if e == "" {
return true
}
if utf8.RuneCountInString(e) < 2 {
return true // 单字（"条""里"）不构成实体
}
// 归一化：去首尾标点/引号，压缩空白
e = strings.Trim(e, " \t\r\n\"'`，。；：！？、（）()【】[]{}<>《》—-·.,;:!?")
if e == "" {
return true
}
if utf8.RuneCountInString(e) < 2 {
return true
}
lower := strings.ToLower(e)
if _, bad := junkEntitySet[lower]; bad {
return true
}
if _, bad := junkEntitySet[e]; bad {
return true
}
// 全由虚词/量词字组成（如「里存着」之外的「的了着」「个条」组合）
if allFunctionChars(e) {
return true
}
// 纯标点/符号（无任何字母/数字/汉字）
if !hasSemanticChar(e) {
return true
}
return false
}
// functionChars 单字虚词/量词集合：整串都由这些字组成 ⇒ 无信息量。
var functionChars = map[rune]struct{}{}
func init() {
for _, r := range "的了着过地得所以于与及或而且则即是在有和与就都也还又很更最可会能要没对为从到把被让给吗呢吧啊呀哦嗯这那我你他她它们个条次些种只件篇位名层里上下中前后内外" {
functionChars[r] = struct{}{}
}
}
func allFunctionChars(s string) bool {
for _, r := range s {
if _, ok := functionChars[r]; !ok {
return false
}
}
return true
}
func hasSemanticChar(s string) bool {
for _, r := range s {
switch {
case r >= 0x4E00 && r <= 0x9FFF: // 汉字
return true
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z':
return true
case r >= '0' && r <= '9':
return true
}
}
return false
}
// FilterEntitiesForGraph 批量过滤垃圾实体，返回保留集与被拒数（观测用）。
func FilterEntitiesForGraph(entities []string) ([]string, int) {
if len(entities) == 0 {
return entities, 0
}
out := make([]string, 0, len(entities))
rejected := 0
for _, e := range entities {
if IsJunkEntity(e) {
rejected++
continue
}
out = append(out, e)
}
return out, rejected
}
go/internal/governance/entity_quality_test.go
+44
View File
@@ -0,0 +1,44 @@
package governance
import "testing"
func TestIsJunkEntity(t *testing.T) {
junk := []string{
"条", "个", "里", "的", "了", "这个", "那些", "里存着", "开发", "进行", "the", "and",
"一个", "多条", "存在", "什么", "为什么", "", " ", "，。", "的了着",
}
for _, e := range junk {
if !IsJunkEntity(e) {
t.Errorf("%q 应判为垃圾实体", e)
}
}
keep := []string{
"织忆", "MemoryWeave", "淇县", "生物制造", "卫都街道", "bge-m3", "KOCR", "llama-server-4b",
"8081", "graph.db", "牧尘", "飞书",
}
for _, e := range keep {
if IsJunkEntity(e) {
t.Errorf("%q 是专名/标识符，不应被判为垃圾", e)
}
}
}
func TestFilterEntitiesForGraph(t *testing.T) {
in := []string{"织忆", "条", "这个", "8081", "里存着"}
out, rejected := FilterEntitiesForGraph(in)
if rejected != 3 {
t.Fatalf("应拒绝 3 个垃圾实体，实际 %d（输出 %v）", rejected, out)
}
if len(out) != 2 || out[0] != "织忆" || out[1] != "8081" {
t.Fatalf("保留集错误：%v", out)
}
}
func TestAllFunctionChars(t *testing.T) {
if !allFunctionChars("的了着") {
t.Fatal("全虚词串应判 true")
}
if allFunctionChars("织忆") {
t.Fatal("专名不应判 true")
}
}
go/internal/governance/graph_expander.go
+14 -3
View File
@@ -36,9 +36,15 @@ func (g *InMemoryGraph) ExpandFromResults(results []models.RecallResult, namespa
startEntities := 0 // 可观测性：去重后实际导航的起点实体数 N
totalPathsHit := 0 // 可观测性：命中路径总数 M（补前去重）
rejectedJunk := 0 // P0-4：被质量闸①丢弃的垃圾实体数
rejectedLowDegree := 0 // P0-4：被质量闸②丢弃的低度数实体数
// 从每个结果正文提实体出发扩展
for _, r := range results {
for _, entity := range extractPotentialEntitiesFromContent(r.Content) {
entities := extractPotentialEntitiesFromContent(r.Content)
// P0-4 第①道闸：垃圾实体不导航
entities, rejected := FilterEntitiesForGraph(entities)
rejectedJunk += rejected
for _, entity := range entities {
if seenEntities[entity] {
continue
}
@@ -50,6 +56,11 @@ func (g *InMemoryGraph) ExpandFromResults(results []models.RecallResult, namespa
if err != nil {
continue
}
// P0-4 第②道闸：路径数 <2 ⇒ 孤边节点，不构成语义关联（内存图无 degree 查询，用路径数代理）
if len(paths) < 2 {
rejectedLowDegree++
continue
}
totalPathsHit += len(paths)
for _, p := range paths {
to, _ := p["target"].(string)
@@ -70,8 +81,8 @@ func (g *InMemoryGraph) ExpandFromResults(results []models.RecallResult, namespa
}
// 可观测性：默认关（ZHIYI_GRAPH_EXPAND_DEBUG=1 打开），保证「下次坏了不沉默」。
if graphExpandDebugEnabled() {
log.Printf("[graph-expand] ns=%s maxHops=%d 起点实体 %d 个 / 命中 %d 条 / 补入 %d 条",
namespace, maxHops, startEntities, totalPathsHit, len(expanded))
log.Printf("[graph-expand] ns=%s maxHops=%d 起点实体 %d 个 / 命中 %d 条 / 补入 %d 条 / 质量闸拒 垃圾 %d 低度 %d",
namespace, maxHops, startEntities, totalPathsHit, len(expanded), rejectedJunk, rejectedLowDegree)
}
return expanded
}
go/internal/governance/graph_file.go
+13 -3
View File
@@ -469,8 +469,13 @@ func (fg *FileGraph) ExpandFromResults(results []models.RecallResult, namespace
// 当实体去 Navigate，永远匹配不到任何图节点 ⇒ 结构性零产出（t_818f77be）。
startEntities := 0 // 可观测性：去重后实际导航的起点实体数 N
totalPathsHit := 0 // 可观测性：命中路径总数 M（补前去重）
rejectedJunk := 0 // P0-4：质量闸①拒绝数
rejectedLowDegree := 0 // P0-4：质量闸②拒绝数
for _, r := range results {
for _, entity := range extractFileGraphEntities(r.Content) {
entities := extractFileGraphEntities(r.Content)
entities, rejected := FilterEntitiesForGraph(entities)
rejectedJunk += rejected
for _, entity := range entities {
if seenEntities[entity] {
continue
}
@@ -482,6 +487,11 @@ func (fg *FileGraph) ExpandFromResults(results []models.RecallResult, namespace
if err != nil {
continue
}
// P0-4 第②道闸：路径数 <2 ⇒ 孤边节点不注入
if len(paths) < 2 {
rejectedLowDegree++
continue
}
totalPathsHit += len(paths)
for _, p := range paths {
target, _ := p["target"].(string)
@@ -502,8 +512,8 @@ func (fg *FileGraph) ExpandFromResults(results []models.RecallResult, namespace
}
// 可观测性：默认关（ZHIYI_GRAPH_EXPAND_DEBUG=1 打开），保证「下次坏了不沉默」。
if graphExpandDebugEnabled() {
log.Printf("[graph-expand] ns=%s maxHops=%d 起点实体 %d 个 / 命中 %d 条 / 补入 %d 条",
namespace, maxHops, startEntities, totalPathsHit, len(expanded))
log.Printf("[graph-expand] ns=%s maxHops=%d 起点实体 %d 个 / 命中 %d 条 / 补入 %d 条 / 质量闸拒 垃圾 %d 低度 %d",
namespace, maxHops, startEntities, totalPathsHit, len(expanded), rejectedJunk, rejectedLowDegree)
}
return expanded
}
go/internal/governance/graph_sqlite.go
+42
View File
@@ -16,6 +16,7 @@ import "C"
import (
"encoding/json"
"fmt"
"log"
"os"
"strconv"
"strings"
@@ -817,6 +818,8 @@ func (gs *SQLiteGraphStore) ExpandFromResults(results []models.RecallResult, nam
seen := make(map[string]bool) // 追踪已访问的记忆 ID
seenEntities := make(map[string]bool) // 追踪已访问的实体（用于去重扩展结果）
rejectedJunk := 0
rejectedLowDegree := 0
for _, r := range results {
if seen[r.ID] {
continue
@@ -824,6 +827,9 @@ func (gs *SQLiteGraphStore) ExpandFromResults(results []models.RecallResult, nam
seen[r.ID] = true
entities := extractPotentialEntities(r.Content)
// P0-4 第①道闸：垃圾实体（虚词/量词/代词/填充短语）不导航
entities, rejected := FilterEntitiesForGraph(entities)
rejectedJunk += rejected
for _, entity := range entities {
if seenEntities[entity] {
continue
@@ -834,6 +840,11 @@ func (gs *SQLiteGraphStore) ExpandFromResults(results []models.RecallResult, nam
// Navigate 的 from/to 都是**节点 ID**（n_织忆），不能用裸名 entity 比对，
// 否则恒不相等 ⇒ 路径全被 continue ⇒ 结构性 0 条附加（t_be4d070a 诊断）。
nodeID := normalizeEntityID(entity)
// P0-4 第②道闸：度数 <2 的孤边节点不构成语义关联 ⇒ 不导航、不注入。
if gs.EdgeDegree(nodeID) < 2 {
rejectedLowDegree++
continue
}
paths, _ := gs.Navigate(nodeID, maxHops, namespace, nil)
for _, p := range paths {
// Navigate 返回字段: from, to, relation, weight, hop（均为节点 ID）
@@ -866,9 +877,40 @@ func (gs *SQLiteGraphStore) ExpandFromResults(results []models.RecallResult, nam
}
}
}
// P0-4 可观测性：两道闸的拒绝量必须能读出来（默认关，debug 开才打印，避免刷屏）。
if graphExpandDebugEnabled() && (rejectedJunk > 0 || rejectedLowDegree > 0) {
log.Printf("[graph-expand] ns=%s 质量闸：垃圾实体拒 %d / 低度数拒 %d / 补入 %d",
namespace, rejectedJunk, rejectedLowDegree, len(expanded)-len(results))
}
return expanded
}
// EdgeDegree 返回节点在图中的边数（入度+出度，P0-4 度数闸用）。
// 与 GetEntityDegree 的区别：后者返回 evidence_count 之和（证据强度），
// 本函数才是字面意义的「度数」—— 孤边节点（度=1）不足以构成语义关联。
func (gs *SQLiteGraphStore) EdgeDegree(nodeID string) int {
if nodeID == "" {
return 0
}
gs.mu.RLock()
defer gs.mu.RUnlock()
rows := queryRows(gs.db, fmt.Sprintf(
"SELECT COUNT(*) as c FROM graph_edges WHERE source = '%s' OR target = '%s'",
escape(nodeID), escape(nodeID)))
if len(rows) == 0 {
return 0
}
switch v := rows[0]["c"].(type) {
case int:
return v
case int64:
return int(v)
case float64:
return int(v)
}
return 0
}
// displayNodeName 把节点 ID 转成可读显示名（去 n_ / n_mem_ / mem_ 前缀）。
// 图节点 ID 形态：实体 n_织忆 / 概念 n_mem_<记忆id>（name 列才是 mem_<id>）。
// 仅用于格式化 Content 的可读展示，不做任何记忆正文编造。
go/internal/governance/graph_sqlite_expand_test.go
+38
View File
@@ -39,9 +39,47 @@ func newTestSQLiteGraph(t *testing.T) *SQLiteGraphStore {
if err := gs.AddEdge("e-ab", "n_织忆", "n_图谱", "LINKS_TO", testNS724, 1.0); err != nil {
t.Fatalf("AddEdge: %v", err)
}
// P0-4 度数闸（2026-10-26）：degree<2 的孤边节点不再注入 ⇒ 正向用例的起点实体
// 必须真的在图里有 ≥2 条边（两跳邻居即真实图谱形态，不是为过测试造数据）。
if err := gs.AddNode("n_召回", "召回", "entity", testNS724); err != nil {
t.Fatalf("AddNode n_召回: %v", err)
}
if err := gs.AddEdge("e-ac", "n_织忆", "n_召回", "LINKS_TO", testNS724, 1.0); err != nil {
t.Fatalf("AddEdge e-ac: %v", err)
}
return gs
}
// TestSQLiteGraphStore_DegreeGate P0-4：度数=1 的孤边实体不得被扩展注入。
func TestSQLiteGraphStore_DegreeGate(t *testing.T) {
path := filepath.Join(t.TempDir(), "graph-degree.db")
gs, err := NewSQLiteGraphStore(path)
if err != nil {
t.Fatalf("NewSQLiteGraphStore: %v", err)
}
t.Cleanup(func() { _ = gs.Close() })
// 孤边：n_孤边 --LINKS_TO--> n_邻居（起点实体度=1）
if err := gs.AddNode("n_孤边", "孤边", "entity", testNS724); err != nil {
t.Fatal(err)
}
if err := gs.AddNode("n_邻居", "邻居", "entity", testNS724); err != nil {
t.Fatal(err)
}
if err := gs.AddEdge("e-one", "n_孤边", "n_邻居", "LINKS_TO", testNS724, 1.0); err != nil {
t.Fatal(err)
}
if got := gs.EdgeDegree("n_孤边"); got != 1 {
t.Fatalf("EdgeDegree(n_孤边) 应为 1，实际 %d", got)
}
results := []models.RecallResult{{
ID: "mem_degree_1", Content: "孤边 的扩展", Category: "distilled", Score: 0.9, Namespace: testNS724,
}}
got := gs.ExpandFromResults(results, testNS724, 1)
if len(got) != len(results) {
t.Fatalf("度数=1 的实体不应扩展注入，实际补入 %d 条", len(got)-len(results))
}
}
// TestSQLiteGraphStore_ExpandFromResults_Inserts 正向：实体在图中 ⇒ 补入 graph_expanded 且 ID 非空。
func TestSQLiteGraphStore_ExpandFromResults_Inserts(t *testing.T) {
gs := newTestSQLiteGraph(t)
go/internal/memprotect/memprotect.go
+46
View File
@@ -30,6 +30,7 @@ package memprotect
import (
"log"
"os"
"strconv"
"strings"
@@ -67,6 +68,41 @@ const (
RedisKeyProtectedIDs = "zhiyi:protected_ids"
)
// ── GA-P2 强保护栏（2026-10-27）──────────────────────────────────────────────
//
// 规则5（可选，ZHIYI_PROTECT_RAILS=1 启用）：recall_verified_count >= 2 ⇒ 保护。
// 口径与 Python 侧 zhiyi_protection.py 的 MIN_VERIFIED_FOR_RAILS 及
// ~/.hermes/plugins/zhiyi/protect_rails.py 规则② 逐字同口径 —— 任何改动必须两侧同步，
// 禁止两套标准（卡 t_944344b5 边界条款）。
//
// 政策边界（牧尘 2026-10-20 拍板）：rv == 1（只被确认过一次）不纳入保护，
// 保留「验证满 2 次才算硬证据」的门槛。
//
// 默认关（未设/env != "1"）⇒ 行为零变化；启用后为纯增保护（只增加「不遗忘」，绝不增加删除）。
const MinVerifiedForRails = 2
// ProtectRailsEnv 强保护栏 env 开关名（与 protect_rails.py / zhiyi_protection.py 同名）。
const ProtectRailsEnv = "ZHIYI_PROTECT_RAILS"
// ProtectRailsEnabled 读 env，每次调用现读（支持不重启热切换，开销可忽略）。
func ProtectRailsEnabled() bool {
return strings.TrimSpace(os.Getenv(ProtectRailsEnv)) == "1"
}
// rowInt 从候选 map 取整型字段，兼容 int/int64/float64
// （IPC 侧候选 map 是 int；直连/其它实现可能给 int64，避免规则因类型断言失败而空转）。
func rowInt(row map[string]interface{}, key string) int {
switch v := row[key].(type) {
case int:
return v
case int64:
return int(v)
case float64:
return int(v)
}
return 0
}
// State 一次 LoadState() 加载的保护状态快照。
// 衰减 tick / admin forget 开始时 LoadState 一次，循环内对每条候选复用同一 State，
// 避免每条记忆都 HGETALL/SMEMBERS 一次 Redis。
@@ -123,6 +159,16 @@ func IsProtected(row map[string]interface{}, st *State) (bool, string) {
}
}
// 规则5（GA-P2 强保护栏，可选）：recall_verified_count >= 2 ⇒ 保护。
// 默认关（ZHIYI_PROTECT_RAILS != "1"）⇒ 行为零变化。启用后：被验证使用 ≥2 次的
// 记忆永不自动遗忘（纯增保护，不增加任何删除）。字段已在候选 map（lancedb_ipc.go
// GetCandidatesForForgetting 已 SELECT recall_verified_count），不依赖 Redis 缓存。
if ProtectRailsEnabled() {
if rv := rowInt(row, "recall_verified_count"); rv >= MinVerifiedForRails {
return true, "recall_verified_ge2"
}
}
return false, ""
}
go/internal/memprotect/memprotect_test.go
+53
View File
@@ -403,3 +403,56 @@ func TestDifferential_FixtureTable(t *testing.T) {
}
}
}
// TestRule5_RailsVerifiedCount (GA-P2, 2026-10-27)：强保护栏 recall_verified_count>=2 规则。
// 口径与 Python zhiyi_protection.py 的 recall_verified>=2 一致；默认关（零行为变化）。
// 政策边界（牧尘 2026-10-20 拍板）：rv==1 不纳入保护（验证满 2 次才算硬证据）。
func TestRule5_RailsVerifiedCount(t *testing.T) {
mkRV := func(rv int) map[string]interface{} {
return map[string]interface{}{
"id": "mem_rails",
"category": "normal", // 不命中规则1
"freshness": "", // 不命中规则3
"content": "普通日常内容", // 不命中规则4b
"recall_verified_count": rv,
}
}
st := synthState(nil, nil) // 空 VProp/Protected ⇒ 规则2/4a 不命中
// 1) 默认关（未设 env）：rv 再高也不保护（行为零变化）
t.Setenv(ProtectRailsEnv, "")
if p, r := IsProtected(mkRV(50), st); p {
t.Fatalf("默认关: rv=50 不应保护, got true rule=%s", r)
}
// 2) ENV="1" 启用：rv>=2 保护（含边界 2），rv=1/0 不保护
t.Setenv(ProtectRailsEnv, "1")
if p, r := IsProtected(mkRV(2), st); !p || r != "recall_verified_ge2" {
t.Errorf("rv=2(边界) 应保护 rule=recall_verified_ge2, got %v rule=%s", p, r)
}
if p, _ := IsProtected(mkRV(5), st); !p {
t.Error("rv=5 应保护")
}
if p, _ := IsProtected(mkRV(1), st); p {
t.Error("rv=1 不应保护（政策边界：验证满 2 次才算硬证据）")
}
if p, _ := IsProtected(mkRV(0), st); p {
t.Error("rv=0 不应保护")
}
// 3) ENV="0" 显式关：不保护
t.Setenv(ProtectRailsEnv, "0")
if p, _ := IsProtected(mkRV(5), st); p {
t.Error("ENV=0 不应保护")
}
// 4) rowInt 类型兼容：int64 也应正确识别（防类型断言失配致规则空转）
t.Setenv(ProtectRailsEnv, "1")
row := map[string]interface{}{
"id": "mem_i64", "category": "normal", "content": "x",
"recall_verified_count": int64(3),
}
if p, _ := IsProtected(row, st); !p {
t.Error("rv=int64(3) 应保护（rowInt 类型兼容）")
}
}
go/internal/models/memory.go
+16
View File
@@ -197,7 +197,23 @@ type RecallResult struct {
// 2026-10-13 时间维检索（t_e9a4d18f P0）：该条目是否可能过期。
// 仅当 stale 标记开启（ZHIYI_STALE_AFTER_HOURS>0）且 updated_at 超期时为 true；
// 默认关闭 ⇒ 恒 false 且 omitempty 不输出 ⇒ 行为零变化（向后兼容）。
// 2026-10-26（t_contract_p0）：默认阈值改为 720h（30 天）——**只做展示标记，
// 不参与排序**（排序侧的有效期判据走 valid_to，见 ValidTo）；环境变量仍可覆盖/关闭。
Stale bool `json:"stale,omitempty"`
// 2026-10-26（t_trace_p0）：本条结果的来源通道（vector|fts|graph|shared|wiki|
// standing|pattern|peer）。此前「为什么被召回」不可答，调参只能盲修（CS249r Ch14）。
Channel string `json:"channel,omitempty"`
// 2026-10-26（P1 supersede 读侧暴露）：版本与取代关系。
// 写入侧一直有 supersedes / superseded_by / valid_from / valid_to / version，
// 但 RecallResult 不带出 ⇒ 读侧永远看不见「这条已被谁取代 / 是否过期」。
// 存量行零值（Version=0=未记录版本），omitempty 保持旧响应兼容。
Version int `json:"generation,omitempty"` // 记忆版本号（DB version 列，语义=第几代）
ValidFrom string `json:"valid_from,omitempty"` // 语义有效期起点（RFC3339，空=永久）
ValidTo string `json:"valid_to,omitempty"` // 语义有效期终点（空=永久）
Supersedes []string `json:"supersedes,omitempty"` // 我取代了谁
SupersededBy []string `json:"superseded_by,omitempty"` // 我被谁取代（反向指针）
// Superseded 便捷布尔：SupersededBy 非空 ⇒ 已有更新版本存在（读侧一个字段可判）。
Superseded bool `json:"superseded,omitempty"`
// A10 (t_d618dc7c)：把「人工有用」单独计数直通给调用方。此前 RecallResult 缺它 ⇒
// 插件/调用方读不到（t_4c810e02「加列不接 IPC=静默 no-op」同款坑）。
// nil=该条从未被人工点过（库里 NULL）；非 nil=真实人工 useful 次数（NULL≠0）。
go/internal/pattern/pattern.go
+5 -3
View File
@@ -2,15 +2,17 @@
//
// 背景：MemOS 血统的 L2「跨任务重复模式」检测。原实现活在 Python daemon 的
// main_loop（30s tick），2026-09-24 为 state-db 写安全整体退役后 L2 pattern 蒸馏
// 停摆（graph_nodes type='pattern' 停在 85 个 / 2026-09-18）。本包把 L1→L2 蒸馏
// 停摆（graph_nodes type='pattern' 停为 0 行；旧 Python daemon 残留的 85 行是
// type='pattern_stat'，id 全 NULL、语义为 L1 命名空间统计快照，非 L2 pattern / 2026-09-18）。本包把 L1→L2 蒸馏
// 复活成**短命单发**（modernc sqlite 直连 graph.db，开/关不占长连接，不碰 state.db），
// 并按设计 §4.1 叠加两条检测规则：
// - 工具组合频率：同一 tool 组合出现 ≥3 次 → tool_combo pattern
// （confidence = min(freq*0.1, 1.0)）
// - 相似解决路径聚类：cosine ≥0.85 成簇且簇内 ≥3 条 → workflow pattern
//
// 存储：复用现有 graph_nodes(type='pattern')，**不另开新表**（避免与 85 存量 + 现有
// 图谱查询割裂）。namespace='daemon-distill' 沿用旧 daemon 约定。
// 存储：复用现有 graph_nodes(type='pattern')，**不另开新表**（避免与现存图谱查询割裂；
// 表内另有 85 行 type='pattern_stat' 属旧 daemon 统计快照，与 L2 pattern 无关）。
// namespace='daemon-distill' 沿用旧 daemon 约定。
package pattern
import (
go/internal/pattern/shadow_diff_test.go
+1 -1
View File
@@ -169,7 +169,7 @@ func peerOpenReadOnly(path string) (*sql.DB, error) {
return sql.Open("sqlite", "file:"+path+"?mode=ro&_pragma=busy_timeout(5000)")
}
// ── 深度根因：即使修好 NULL-id，85 行 pattern 无 data/tools 键（只有 description），
// ── 深度根因：即使修好 NULL-id，85 行 pattern_stat 无 data/tools 键（只有 description），
// Match 只能 embed 命名键（如 "pattern:hermes-main:entity"）⇒ 对金标余弦 < 0.6 阈值。
// 本测试量化「修好 NULL-id 后开态到底能注入几类金标」—— 证明开态收益仍≈0。
func TestShadowPattern_CosineIfIDFixed(t *testing.T) {
go/internal/pattern/store.go
+3 -2
View File
@@ -1,7 +1,8 @@
// 织忆 MemoryWeave — L2 pattern 存储（t_d32e5f8a）
//
// 存储沿用现有 graph_nodes(type='pattern')，**不另开新表**（避免与 85 存量 +
// 现有图谱查询割裂）。modernc sqlite 短命单发开/关，不持长连接、不碰 state.db。
// 存储沿用现有 graph_nodes(type='pattern')，**不另开新表**（避免与现存图谱查询割裂；
// 表内另有 85 行 type='pattern_stat' 属旧 daemon 统计快照，与 L2 pattern 无关）。
// modernc sqlite 短命单发开/关，不持长连接、不碰 state.db。
//
// 写入幂等语义（复活 L1→L2 occurrence_count 驱动）：
// - 同名 pattern 节点存在 → occurrence_count += 1、last_seen 刷新、props 合并（新键覆盖、旧键保留）
go/internal/storage/recall.go
+155 -10
View File
@@ -2,6 +2,7 @@
package storage
import (
"context"
"encoding/json"
"fmt"
"log"
@@ -63,7 +64,7 @@ var (
// 2026-10-13 时间维检索（t_e9a4d18f P0）：stale 标记。
// updated_at 距今超过该小时数 ⇒ 响应带 stale:true，让调用方「看见」可能过期。
// 0 = 关闭（默认）⇒ 不产生任何 stale 标记，行为零变化。
staleAfterHours = envFloat("ZHIYI_STALE_AFTER_HOURS", 0)
staleAfterHours = envFloat("ZHIYI_STALE_AFTER_HOURS", 720)
)
// ── t_d32e5f8a L2 模式召回注入（env 门控 ZHIYI_PATTERN_RECALL，默认 0 ⇒ 行为零变化）──
@@ -108,6 +109,12 @@ func bitemporalEnabled() bool {
return false
}
// BitemporalEnabled 导出 bi-temporal 开关（供 API 层在补充通道/图谱注入后做同一口径过滤）。
func BitemporalEnabled() bool { return bitemporalEnabled() }
// ValidityExpired 导出有效期判定（同上；空串=永久有效，解析失败保守视为未过期）。
func ValidityExpired(validTo string, now time.Time) bool { return validityExpired(validTo, now) }
// validityExpired 判断记忆的语义有效期是否已过（valid_to < now ⇒ 失效）。
// valid_to 空串 = 永久有效（NULL 语义，永不失效）。解析失败保守视为未失效（不误伤）。
func validityExpired(validTo string, now time.Time) bool {
@@ -358,6 +365,12 @@ func fuseModeProof() bool {
type RecallMeta struct {
CandidatesScanned int
CacheHit bool
// TraceID / Status / PartialReason（2026-10-26 t_trace_p0 + t_contract_p0）：
// 响应契约的最小可观测面 —— trace_id 贯穿整条链路，status=partial 时
// 调用方不得把「空结果」解读为「不存在」。
TraceID string
Status string // complete | partial（空串等价 complete，向后兼容）
PartialReason string
// Truncated（t_3aa2a8c1）：本次召回的粗检索单次重读是否被 Rust 侧行数上限截断
// （lancedb_search 单次重读 cap，默认 20000，env ZHIYI_HEAVY_READ_MAX_ROWS 可调）。
// true = 部分结果：candidates_scanned/count 反映截断后状态，调用方不得当作「全量」。
@@ -415,7 +428,25 @@ func (p *RecallPipeline) RecallWithMeta(query, namespace string, topK int, diver
// 只有 **agent_id 派生** 的作用域才该传 true —— 显式传 namespace 的调用方传 false，
// 否则会跨域泄漏不该看到的记忆。
func (p *RecallPipeline) RecallWithMetaOpts(query, namespace string, topK int, diversity float64, allowNSFallback bool) ([]models.RecallResult, RecallMeta, error) {
results, meta, err := p.recallWithMetaInner(query, namespace, topK, diversity, allowNSFallback)
return p.RecallWithMetaTraced(context.Background(), query, namespace, topK, diversity, allowNSFallback, nil)
}
// RecallWithMetaTraced（2026-10-26 t_trace_p0）：带 ctx（deadline 可中断）+ 链路追踪的召回入口。
// 旧方法全部委托到本方法、tr=nil ⇒ 旧调用方行为逐位不变。tr 非 nil 时记录各阶段耗时/通道贡献。
// ctx 在每个阶段边界检查，超时 ⇒ 立即返回已得结果并把 status 标为 partial（reason=deadline_exceeded），
// 调用方据此区分「真的没有」与「扫了一半就超时」（Air Canada 式静默幻觉的结构性入口）。
func (p *RecallPipeline) RecallWithMetaTraced(ctx context.Context, query, namespace string, topK int, diversity float64, allowNSFallback bool, tr *RecallTrace) ([]models.RecallResult, RecallMeta, error) {
if ctx == nil {
ctx = context.Background()
}
results, meta, err := p.recallWithMetaInner(ctx, query, namespace, topK, diversity, allowNSFallback, tr)
if tr != nil {
meta.TraceID = tr.TraceID
if tr.Status == "partial" {
meta.Status = "partial"
meta.PartialReason = tr.PartialReason
}
}
if err != nil {
return results, meta, err
}
@@ -423,12 +454,15 @@ func (p *RecallPipeline) RecallWithMetaOpts(query, namespace string, topK int, d
if sa := LookupStandingAnswer(query); sa != nil {
results = PrependStandingAnswer(results, sa, topK)
meta.StandingAnswerHit = sa.QuestionID
if tr != nil {
tr.StandingAnswerHit = sa.QuestionID
}
}
}
return results, meta, err
}
func (p *RecallPipeline) recallWithMetaInner(query, namespace string, topK int, diversity float64, allowNSFallback bool) ([]models.RecallResult, RecallMeta, error) {
func (p *RecallPipeline) recallWithMetaInner(ctx context.Context, query, namespace string, topK int, diversity float64, allowNSFallback bool, tr *RecallTrace) ([]models.RecallResult, RecallMeta, error) {
// topK 先归一：缓存键必须与写入侧用的 topK 一致。recallUncached 内部也会归一，
// 若此处不先归一，topK<=0 的调用读到的是"归一前的键"、永远命中不了自己写的条目。
if topK <= 0 {
@@ -445,11 +479,17 @@ func (p *RecallPipeline) recallWithMetaInner(query, namespace string, topK int,
}
p.recordRecallWrites(filtered)
// t_1c12e000：命中缓存 ⇒ 显式告知调用方（candidates_scanned 恒 0）
return filtered, RecallMeta{CacheHit: true}, nil
st := time.Now()
if tr != nil {
tr.CacheHit = true
tr.TopK = topK
tr.StageNote("cache_hit", st, len(filtered), "候选池不可重建（读 60s 内快照）")
}
return filtered, RecallMeta{CacheHit: true, Status: "complete"}, nil
}
}
}
return p.recallUncached(query, namespace, topK, diversity, allowNSFallback)
return p.recallUncached(ctx, query, namespace, topK, diversity, allowNSFallback, tr)
}
// Recall 保持原签名（缓存 + 全量流程），行为不变。
@@ -460,16 +500,30 @@ func (p *RecallPipeline) Recall(query, namespace string, topK int, diversity flo
// recallUncached Recall 的非缓存路径。候选池大小在 Step 2.5 过滤后、
// Step 3 截断前捕获（真实粗检索池，不是最终产出数）。
func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diversity float64, allowFallback bool) ([]models.RecallResult, RecallMeta, error) {
func (p *RecallPipeline) recallUncached(ctx context.Context, query, namespace string, topK int, diversity float64, allowFallback bool, tr *RecallTrace) ([]models.RecallResult, RecallMeta, error) {
if topK <= 0 {
topK = 10
}
if ctx == nil {
ctx = context.Background()
}
if tr != nil {
tr.TopK = topK
}
// Step 1: Encode query
stEncode := time.Now()
queryVec, err := p.embedder.EncodeQuery(query)
if err != nil {
if tr != nil {
tr.StageNote("encode", stEncode, 0, err.Error())
tr.Error = err.Error()
}
return nil, RecallMeta{}, fmt.Errorf("recall encode: %w", err)
}
if tr != nil {
tr.Stage("encode", stEncode, 0)
}
// Step 2: Coarse search — own namespace + shared（§1.3 召回规则）
// 默认搜自己的 namespace + shared，显式 include_namespaces 时才跨域
@@ -480,6 +534,7 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
// ⇒ 默认召回只能看到极小一片（全表 10172 条，默认调用实测仅 11 个候选）。
// 显式传 namespace / agent_id 时保持原作用域行为，不改变调用方语义。
var candidates []models.MemoryRecord
stCoarse := time.Now()
// t_3aa2a8c1：Rust 侧单次重读行数上限的 truncated 上抛到 recall 响应。
// 只有实现了可选扩展接口 SearchTruncationReporter 的后端（RustLanceDBClient）
// 会上报截断；未实现的（纯 Go / 测试 fake）保持旧 Search ⇒ 恒未截断。
@@ -517,6 +572,18 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
candidates = append(candidates, nsResults...)
}
}
if tr != nil {
tr.Stage("coarse_search", stCoarse, len(candidates))
tr.CandidatesScanned = len(candidates)
}
// deadline 边界（t_contract_p0）：粗检索已超时 ⇒ 立刻返回 partial，
// 不继续做重排/MMR（省算力，也让调用方拿到明确的「不完整」信号）。
if err := ctx.Err(); err != nil {
if tr != nil {
tr.MarkPartial("deadline_exceeded")
}
return nil, RecallMeta{CandidatesScanned: len(candidates), Truncated: truncated, Status: "partial", PartialReason: "deadline_exceeded"}, nil
}
// 2026-10-26：**ns 作用域兜底**（牧尘令，可逆）。
// 背景：请求带 agent_id 时 resolveRecallNamespace 会派生 <agent>-main，
// 而该 ns 里几乎没有货 —— 实测 default-main 池 6 条 / xiaoyi-main 6 / prof-cloud-main 7，
@@ -559,6 +626,7 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
// 门控关 ⇒ 本块不执行（零行为变化）。
var ftsTail []models.MemoryRecord
ftsScoreOf := map[string]float64{} // id → 词面分（只用于尾部排序）
stFts := time.Now()
if ftsEnabled() {
ftsScores := ftsInstance.FtsSearch(query, 30)
for id, s := range ftsScores {
@@ -590,6 +658,9 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
log.Printf("[recall] FTS 词面通道命中 %d 条（候选池 %d）", len(ftsTail), len(candidates))
}
}
if tr != nil {
tr.Stage("fts_recall", stFts, len(ftsTail))
}
// Step 2.5: 过滤原始对话内容（category=episodes/对话/对话记录）
// 原因：原始对话未蒸馏，不应出现在 recall 结果中
@@ -662,6 +733,7 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
// 因为下面会原地覆盖为 BM25 融合值（0.7*dbQuality + 0.3*kwScore）。
// 响应需要展示 DB 原值（"响应值==库值"），不能展示被融合改写后的值。
dbQuality := make([]float64, len(candidates))
stBm25 := time.Now()
if len(candidates) > 0 {
for i := range candidates {
dbQuality[i] = candidates[i].QualityScore
@@ -697,6 +769,10 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
}
}
if tr != nil {
tr.Stage("bm25", stBm25, len(candidates))
}
if len(candidates) == 0 && len(ftsTail) == 0 {
// 自动记录召回缺口（gap_scan 触发器会在 30min 冷却后分类）
// t_2afae95e：FTS 尾部位命中也算有结果 ⇒ 不记 miss、不走空返回
@@ -713,6 +789,7 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
// 存储的 Vector，直接算 cosine 即可。旧实现把 query + 50~100 条候选文档重新送进
// 嵌入服务（int8 ONNX CPU），占 recall 总耗时 90%+，是 Hermes 记忆预取 8s 预算
// 被打爆的直接原因。向量不齐/维度不符时自动回退文本重排（行为与旧版一致）。
stRerank := time.Now()
docs := make([]string, len(candidates))
docVecs := make([][]float32, len(candidates))
for i, c := range candidates {
@@ -832,6 +909,17 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
}
// Step 4: MMR diversity
if tr != nil {
tr.Stage("rerank_fusion", stRerank, len(reranked))
}
// deadline 边界：重排已超时 ⇒ 返回 partial（不继续 MMR/图谱扩展）。
if ctx.Err() != nil {
if tr != nil {
tr.MarkPartial("deadline_exceeded")
}
return nil, RecallMeta{CandidatesScanned: poolSize, Truncated: truncated, Status: "partial", PartialReason: "deadline_exceeded"}, nil
}
stMMR := time.Now()
finalIndices := MMRSelect(reranked, candidates, topK*2, diversity)
results := make([]models.RecallResult, 0, topK)
@@ -902,11 +990,22 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
VValue: vv,
// A10 (t_d618dc7c)：人工有用单独计数直通（nil=NULL=从未被人工点过）
HumanUsefulCount: c.HumanUsefulCount,
// 2026-10-13 (t_e9a4d18f P0)：过期标记（默认关闭 ⇒ 恒 false 不输出）
// 2026-10-13 (t_e9a4d18f P0)：过期标记（默认 720h，仅展示不参与排序）
Stale: isStale(c.UpdatedAt, staleAfterHours),
// 2026-10-26（t_trace_p0 / P1 supersede）：来源通道 + 版本/取代关系直通
Channel: "vector",
Version: c.Version,
ValidFrom: c.ValidFrom,
ValidTo: c.ValidTo,
Supersedes: c.Supersedes,
SupersededBy: c.SupersededBy,
Superseded: len(c.SupersededBy) > 0,
})
}
}
if tr != nil {
tr.Stage("mmr_select", stMMR, len(results))
}
// Step 3.8 (t_2afae95e FTS 词面通道尾部保证位, 门控同 Step 2.2):
// 词面命中在 **rerank+MMR 之后**才并入最终结果（诊断第 3 条：rerank 之前
@@ -989,6 +1088,14 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
QualityScore: rec.QualityScore,
QualityScoreReliable: false, // FTS 记录未经过质量融合 ⇒ 标不可信（审计可辨）
FtsBoosted: true,
Channel: "fts",
Version: rec.Version,
ValidFrom: rec.ValidFrom,
ValidTo: rec.ValidTo,
Supersedes: rec.Supersedes,
SupersededBy: rec.SupersededBy,
Superseded: len(rec.SupersededBy) > 0,
Stale: isStale(rec.UpdatedAt, staleAfterHours),
// A10 (t_d618dc7c)：人工有用单独计数直通（nil=NULL=从未被人工点过）
HumanUsefulCount: rec.HumanUsefulCount,
})
@@ -997,6 +1104,8 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
}
// Step 5: 图谱融合扩展（不再作为后备补丁，每次 recall 都跑 1 跳）
stGraph := time.Now()
graphAdded := 0
if p.graph != nil {
expanded := p.graph.ExpandFromResults(results, namespace, 1)
for _, e := range expanded {
@@ -1008,12 +1117,18 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
}
}
if !already {
e.Channel = "graph"
results = append(results, e)
graphAdded++
}
}
}
if tr != nil {
tr.Stage("graph_expand", stGraph, graphAdded)
}
// Step 6: 去重（相同 content 只保留最高分）
stDedup := time.Now()
seenContent := make(map[string]bool)
deduped := make([]models.RecallResult, 0, len(results))
for _, r := range results {
@@ -1023,6 +1138,9 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
}
}
results = deduped
if tr != nil {
tr.Stage("dedupe", stDedup, len(results))
}
// Step 6.2: t_d32e5f8a L2 模式召回注入（env 门控 ZHIYI_PATTERN_RECALL，默认 0）
// 命中 pattern 时以 Category="pattern" 附带 top-k 返回（设计「命中即注入」）。
@@ -1038,6 +1156,7 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
}
}
if !dup {
pr.Channel = "pattern"
results = append(results, pr)
}
}
@@ -1055,6 +1174,7 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
}
}
if !dup {
pr.Channel = "peer"
results = append(results, pr)
}
}
@@ -1127,7 +1247,18 @@ func (p *RecallPipeline) recallUncached(query, namespace string, topK int, diver
}
}
return results, RecallMeta{CandidatesScanned: poolSize, Truncated: truncated}, nil
// 契约（t_contract_p0）：status 默认 complete；行数上限截断 ⇒ partial（
// 调用方不得把截断后的空/少结果解读为「不存在」）。
status := "complete"
partialReason := ""
if truncated {
status = "partial"
partialReason = "row_cap_truncated"
if tr != nil {
tr.MarkPartial("row_cap_truncated")
}
}
return results, RecallMeta{CandidatesScanned: poolSize, Truncated: truncated, Status: status, PartialReason: partialReason}, nil
}
// recallV2 开关（2026-10-14）。默认关 ⇒ 行为零变化。
@@ -1330,7 +1461,7 @@ func cosineSimilarity(a, b []float32) float64 {
func (p *RecallPipeline) recordRecallWrites(results []models.RecallResult) {
ids := make([]string, 0, len(results))
for _, r := range results {
if r.ID != "" {
if isMemoryRowID(r.ID) {
ids = append(ids, r.ID)
}
}
@@ -1344,12 +1475,26 @@ func (p *RecallPipeline) recordRecallWrites(results []models.RecallResult) {
go p.incrementRecallCount(results)
}
// isMemoryRowID 只认真实落在 memories 表里的行 ID（mem_*）。
//
// 2026-10-26（t_0a54c12b 写入放大复发）：召回结果集里混着大量**不属于 memories 表**的 ID：
// - ep_* = episodes 原始轮次（另一张表）
// - n_* = 图谱扩展节点（graph_expanded；实测 memories 表 40383 行里 n_* 前缀 = 0）
//
// 这些 ID 送进 update_recall_batch 只会产生「谓词命中 0 行」的提交 —— 而 LanceDB 是 MVCC，
// **每次提交无论命中几行都是一个版本**。实测 2026-10-26 21:26 一条批：
// `delta=1 ids=791 rows=10`（命中率 1.3%）、`delta=2 ids=32 rows=0`（纯空转版本）。
// 过滤后 ① 每条 batch 命中率回到 ~100% ② pending 涨得慢 ⇒ 阈值 flush 次数降一个量级。
func isMemoryRowID(id string) bool {
return strings.HasPrefix(id, "mem_")
}
// incrementRecallCount 旧版逐条同步写（每条结果 = 1 个 LanceDB 版本）。
// 已不作为主路径，仅保留为缓冲不可用时的兜底。
func (p *RecallPipeline) incrementRecallCount(results []models.RecallResult) {
now := time.Now().Format(time.RFC3339)
for _, r := range results {
if r.ID != "" {
if isMemoryRowID(r.ID) {
// t_268b0353（治根）：兜底路径同样不写 freshness=verified（"被召回"≠"被验证"）
// GA-P0-2 (t_c6d41ac6)：召回命中 → recall_seen_count（recall_count 冻结）
_ = p.lancedb.Update("memories", r.ID, map[string]any{
go/internal/storage/recall_write_buffer.go
+104 -30
View File
@@ -5,6 +5,7 @@ import (
"os"
"sort"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
@@ -44,7 +45,13 @@ type BatchRecallUpdater interface {
const (
defaultRecallFlushInterval = 300 * time.Second
defaultRecallFlushMaxItems = 256
// 2026-10-26 (t_0a54c12b)：256 → 2048。
// 256 在搜索突发期每 1~2 秒就触发一次 flush（每次 flush = 2~4 个 delta 组 = 2~4 个
// LanceDB 版本），实测 20~200 版本/分钟。待写条目是「不同记忆 ID」，2048 只占几百 KB 内存，
// 把窗口内重复命中合并得更彻底、flush 次数降一个量级。
defaultRecallFlushMaxItems = 2048
// requeue 硬上限：IPC 长时间不可用时 pending 不能无限涨（召回计数是观测值，丢了可接受）。
recallRequeueHardCap = 16384
// 优雅停止窗口（t_fa797edd）：Stop 内的「等后台协程退出 + 最终落盘」都必须有界。
// 否则 Rust/LanceDB 批量更新在活 IPC 流量下阻塞时，Stop 永远不返回 → 信号协程
// 到不了 srv.Shutdown → 进程跑满 systemd 90s TimeoutStopSec 被 SIGKILL
@@ -77,6 +84,8 @@ type RecallWriteBuffer struct {
flushRows int64
flushErrors int64
fallbackRows int64
// requeuedItems: IPC 通道不可用时整批退回缓冲的条目数（t_0a54c12b 观测项）
requeuedItems int64
}
// RecallWriteBufferInstance 进程级单例（与 SearchCacheInstance / CoOccurTrackerInstance 同模式）
@@ -205,30 +214,27 @@ func (b *RecallWriteBuffer) Flush() (int, error) {
usedFallback := false
if b.batch != nil {
rows, err = b.batch.UpdateRecallBatch("memories", batch, ts)
}
if b.batch == nil || err != nil {
usedFallback = true
if err != nil {
// 2026-10-26 (t_0a54c12b)：**通道不可用**与**语句被拒**必须分开处置。
// 通道不可用（sidecar 重启/拒绝连接/超时）时逐条降级毫无收益：
// ① IPC 不可用 ⇒ 每条都失败，2026-10-26 21:16 实测 **9142 次失败拨号**（日志风暴）；
// ② sidecar 恢复瞬间逐条成功 ⇒ **每条 = 1 个 LanceDB 版本**，正是 _versions 冲到 8.5k 的机制。
// 正确做法：整批退回 pending，下个窗口连同新命中一起批量重试（召回计数延迟可接受）。
if isRecallTransportErr(err) {
atomic.AddInt64(&b.flushErrors, 1)
atomic.AddInt64(&b.requeuedItems, int64(len(batch)))
n := b.requeue(batch, lastAt)
log.Printf("[recall-buffer] IPC 通道不可用(%v) → 整批 %d 条退回缓冲（pending=%d），下个窗口重试；不逐条降级（避免拨号风暴 + N 个版本）", err, len(batch), n)
return 0, err
}
log.Printf("[recall-buffer] 单事务批量提交失败(%v) → 退化逐条 Update（数据不丢，版本数不优化）", err)
atomic.AddInt64(&b.flushErrors, 1)
usedFallback = true
rows = b.fallbackRowUpdates(batch, ts)
}
rows = 0
for _, it := range batch {
// t_268b0353（2026-10-11 治根）：召回只更新 recall_count / last_recalled_at，
// 不再写 freshness="verified"。freshness=verified 的语义是「经过校验/用户
// 反馈」；"被召回" ≠ "被验证"。此前每次召回都盖 verified → 越常被召回的
// 旧快照越"可信"，时效判断全被污染（过期快照被当现状的机制根因之一）。
// GA-P0-2 (t_c6d41ac6)：召回命中写 recall_seen_count；不再写 recall_count（冻结）。
if uerr := b.ldb.Update("memories", it.ID, map[string]any{
"recall_seen_count": map[string]string{"$inc": "1"},
"last_recalled_at": ts,
}); uerr != nil {
log.Printf("[recall-buffer] 逐条 Update 失败 id=%s: %v", it.ID, uerr)
continue
}
rows++
}
atomic.AddInt64(&b.fallbackRows, rows)
} else {
usedFallback = true
rows = b.fallbackRowUpdates(batch, ts)
}
// 落盘成功后同步进程内缓存（缓存只是热数据，权威值在 LanceDB）
@@ -254,6 +260,73 @@ func (b *RecallWriteBuffer) Flush() (int, error) {
return int(rows), nil
}
// fallbackRowUpdates 逐条 Update（仅用于「批量语句被拒」或后端不支持批量的事务语义兜底）。
// 每条 = 1 个 LanceDB 版本，故**不得**在 IPC 通道不可用时走这里（见 Flush 里的通道判定）。
func (b *RecallWriteBuffer) fallbackRowUpdates(batch []RecallWriteItem, ts string) int64 {
var rows int64
for _, it := range batch {
// t_268b0353（2026-10-11 治根）：召回只更新 recall_count / last_recalled_at，
// 不再写 freshness="verified"。freshness=verified 的语义是「经过校验/用户
// 反馈」；"被召回" ≠ "被验证"。此前每次召回都盖 verified → 越常被召回的
// 旧快照越"可信"，时效判断全被污染（过期快照被当现状的机制根因之一）。
// GA-P0-2 (t_c6d41ac6)：召回命中写 recall_seen_count；不再写 recall_count（冻结）。
if uerr := b.ldb.Update("memories", it.ID, map[string]any{
"recall_seen_count": map[string]string{"$inc": "1"},
"last_recalled_at": ts,
}); uerr != nil {
log.Printf("[recall-buffer] 逐条 Update 失败 id=%s: %v", it.ID, uerr)
continue
}
rows++
}
atomic.AddInt64(&b.fallbackRows, rows)
return rows
}
// requeue 把一批未落盘的增量退回 pending（IPC 通道不可用时用）：delta 累加、时间取更晚者。
// 硬上限 recallRequeueHardCap：IPC 长期不可用时不能无限涨（召回计数是观测值，丢了可接受）。
func (b *RecallWriteBuffer) requeue(batch []RecallWriteItem, lastAt time.Time) int {
b.mu.Lock()
defer b.mu.Unlock()
for _, it := range batch {
if len(b.pending) >= recallRequeueHardCap {
if _, ok := b.pending[it.ID]; !ok {
continue // 满了且是新 ID：丢弃（观测值，可接受）
}
}
p, ok := b.pending[it.ID]
if !ok {
p = &pendingRecall{}
b.pending[it.ID] = p
}
p.delta += it.Delta
if lastAt.After(p.lastAt) {
p.lastAt = lastAt
}
}
return len(b.pending)
}
// isRecallTransportErr 判断是否为「IPC 通道不可用」类错误（可重试、且绝不能逐条降级）。
// 判据是**通道层**错误关键字，不是业务/谓词错误 —— 后者逐条降级有意义（换语句能成），
// 前者逐条只会制造拨号风暴 + 恢复后 N 个版本。
func isRecallTransportErr(err error) bool {
if err == nil {
return false
}
msg := strings.ToLower(err.Error())
for _, k := range []string{
"ipc dial", "connection refused", "connection reset", "broken pipe",
"i/o timeout", "context deadline exceeded", "unexpected eof",
"no such file or directory", // unix socket 文件不存在（sidecar 正在重启）
} {
if strings.Contains(msg, k) {
return true
}
}
return false
}
// Stop 停止后台 flush 协程，并把最后一个窗口落盘（进程优雅退出时调用）。
// 全程有界（t_fa797edd）：等后台协程 ≤ defaultRecallStopLoopWait，最终落盘
// ≤ defaultRecallStopFlushWait。Rust/LanceDB 批量更新在活 IPC 流量下可能阻塞
@@ -296,14 +369,15 @@ func (b *RecallWriteBuffer) Stats() map[string]interface{} {
pendingN := len(b.pending)
b.mu.Unlock()
return map[string]interface{}{
"pending": pendingN,
"flush_count": atomic.LoadInt64(&b.flushCount),
"flush_items": atomic.LoadInt64(&b.flushItems),
"flush_rows": atomic.LoadInt64(&b.flushRows),
"flush_errors": atomic.LoadInt64(&b.flushErrors),
"fallback_rows": atomic.LoadInt64(&b.fallbackRows),
"window": b.interval.String(),
"max_items": b.maxItems,
"single_tx": b.batch != nil,
"pending": pendingN,
"flush_count": atomic.LoadInt64(&b.flushCount),
"flush_items": atomic.LoadInt64(&b.flushItems),
"flush_rows": atomic.LoadInt64(&b.flushRows),
"flush_errors": atomic.LoadInt64(&b.flushErrors),
"fallback_rows": atomic.LoadInt64(&b.fallbackRows),
"requeued_items": atomic.LoadInt64(&b.requeuedItems),
"window": b.interval.String(),
"max_items": b.maxItems,
"single_tx": b.batch != nil,
}
}
go/internal/storage/recall_write_buffer_test.go
+88
View File
@@ -18,6 +18,8 @@ type fakeLDB struct {
updateCalls int
updateIDs []string
forceBatchErr bool
// batchErr：指定批量提交返回的错误（用于验证「通道不可用」与「语句被拒」分流）。
batchErr error
}
// Scan mock：write-buffer 测试只需占位实现（被测路径不实际调用）。
@@ -72,6 +74,9 @@ func (f *fakeLDB) PurgeInvalid() (int64, error) { return 0, nil }
// BatchRecallUpdater 实现（单事务批量）
func (f *fakeLDB) UpdateRecallBatch(table string, items []RecallWriteItem, lastRecalledAt string) (int64, error) {
if f.batchErr != nil {
return 0, f.batchErr
}
if f.forceBatchErr {
return 0, fmt.Errorf("forced batch failure")
}
@@ -227,6 +232,89 @@ func TestRecallWriteBufferFallbackToPerItem(t *testing.T) {
}
}
// t_0a54c12b：IPC 通道不可用（sidecar 重启/拒绝连接）时必须**整批退回缓冲**，
// 不许逐条降级 —— 逐条在通道断时是 N 次失败拨号，恢复瞬间又变成 N 个 LanceDB 版本
// （2026-10-26 实测 9142 次逐条 Update、_versions 冲 8.5k）。
func TestRecallWriteBufferTransportErrRequeues(t *testing.T) {
fake := &fakeLDB{batchErr: fmt.Errorf("ipc dial: dial unix /tmp/zhiyi-ipc.sock: connect: connection refused")}
b := NewRecallWriteBuffer(fake, 0, 1000)
defer b.Stop()
b.Record([]string{"mem_r1", "mem_r2"})
if _, err := b.Flush(); err == nil {
t.Fatal("通道不可用时 Flush 应返回 err 供上层观测")
}
fake.mu.Lock()
if fake.updateCalls != 0 {
t.Fatalf("通道不可用时禁止逐条降级，实际 updateCalls=%d", fake.updateCalls)
}
fake.mu.Unlock()
if st := b.Stats(); st["pending"].(int) != 2 {
t.Fatalf("整批应退回缓冲 pending=2，实际 %v", st["pending"])
}
// 通道恢复 → 下个窗口连同新命中一起批量落盘，delta 不丢
fake.mu.Lock()
fake.batchErr = nil
fake.mu.Unlock()
b.Record([]string{"mem_r1"}) // 窗口内再命中一次 → r1 delta=2
rows, err := b.Flush()
if err != nil {
t.Fatalf("恢复后 flush 应成功: %v", err)
}
if rows != 2 {
t.Fatalf("恢复后 rows = %d, want 2", rows)
}
fake.mu.Lock()
defer fake.mu.Unlock()
deltas := map[string]int{}
for _, it := range fake.batchItems {
deltas[it.ID] = it.Delta
}
if deltas["mem_r1"] != 2 || deltas["mem_r2"] != 1 {
t.Fatalf("退回缓冲后 delta 应累加不丢，实际 %v", deltas)
}
if st := b.Stats(); st["requeued_items"].(int64) != 2 {
t.Fatalf("requeued_items = %v, want 2", st["requeued_items"])
}
}
// t_0a54c12b：只有 mem_* 会进缓冲。ep_*（episodes 表）与 n_*（图谱扩展节点，
// memories 表实测 0 行）送进 update_recall_batch 只会产生「命中 0 行」的空转版本。
func TestRecallWriteBufferSkipsNonMemoryIDs(t *testing.T) {
// 过滤逻辑本体（recordRecallWrites 用的就是它）
got := memoryRowIDs([]models.RecallResult{
{ID: "mem_keep1"}, {ID: "mem_keep2"},
{ID: "ep_1793020608863378731"}, {ID: "n_静默失效"}, {ID: "n_api_key"}, {ID: ""},
})
if len(got) != 2 || got[0] != "mem_keep1" || got[1] != "mem_keep2" {
t.Fatalf("只应保留 mem_* 两条，实际 %v", got)
}
if isMemoryRowID("ep_x") || isMemoryRowID("n_y") || isMemoryRowID("") || !isMemoryRowID("mem_z") {
t.Fatal("isMemoryRowID 判据错误（只应认 mem_ 前缀）")
}
// 写进缓冲的也只有 mem_*（其余 ID 不进 pending ⇒ 不会触发空转提交）
fake := &fakeLDB{}
b := NewRecallWriteBuffer(fake, 0, 1000)
defer b.Stop()
b.Record(got)
if st := b.Stats(); st["pending"].(int) != 2 {
t.Fatalf("pending = %v, want 2", st["pending"])
}
}
// memoryRowIDs 供测试复用生产过滤逻辑。
func memoryRowIDs(rs []models.RecallResult) []string {
out := make([]string, 0, len(rs))
for _, r := range rs {
if isMemoryRowID(r.ID) {
out = append(out, r.ID)
}
}
return out
}
// Stop 时落盘最后一个窗口（优雅退出不丢统计）
func TestRecallWriteBufferStopFlushes(t *testing.T) {
fake := &fakeLDB{}
go/internal/storage/stale_lifecycle_test.go
+99
View File
@@ -0,0 +1,99 @@
// 织忆 MemoryWeave — stale-return / 生命周期 fixture（P1-3，2026-10-26）
//
// 评审指出的缺口：只有离线检索指标，没有 stale-return 与生命周期 fixture ——
// 即「记忆过期后读侧表现」从未被机械验证过。
//
// 本文件锁三件事：
// ① stale 标记默认开（720h / 30 天）但**只做展示**：老记忆被标 stale，新记忆不标；
// ② bi-temporal 有效期窗口（fixture 取 10:00 生效 / 11:00 失效）：
// T<11:00 未过期、T>11:00 已过期 —— 用 10:00→11:00 的真实时窗语义，而非「随便设个过去时间」；
// ③ supersede 反向指针在读侧可见（P1-1）：SupersededBy 非空 ⇒ Superseded=true，且 JSON 输出。
package storage
import (
"encoding/json"
"testing"
"time"
"github.com/xiaoxue/memoryweave/internal/models"
)
// TestStale_Default720h 默认阈值 720h：29 天不标、31 天标。
func TestStale_Default720h(t *testing.T) {
if got := envFloat("ZHIYI_STALE_AFTER_HOURS", 720); got != 720 {
t.Fatalf("默认 stale 阈值应为 720h，实际 %v", got)
}
now := time.Now()
if isStale(now.Add(-29*24*time.Hour), 720) {
t.Fatal("29 天前更新不应标 stale")
}
if !isStale(now.Add(-31*24*time.Hour), 720) {
t.Fatal("31 天前更新应标 stale")
}
// 关闭开关（0）⇒ 恒 false，保持旧行为可回滚
if isStale(now.Add(-1000*time.Hour), 0) {
t.Fatal("阈值 0 应关闭 stale 标记")
}
}
// TestValidityWindow_1000to1100 生命周期 fixture：事实 10:00 生效、11:00 失效。
// 判据：只在 (valid_from, valid_to) 区间内被视为有效；区间外（含恰好 11:00 之后）判过期。
func TestValidityWindow_1000to1100(t *testing.T) {
day := time.Date(2026, 10, 26, 0, 0, 0, 0, time.Local)
from := day.Add(10 * time.Hour).Format(time.RFC3339)
to := day.Add(11 * time.Hour).Format(time.RFC3339)
cases := []struct {
name string
now time.Time
want bool // true = 已过期
}{
{"09:30 尚未生效（未过期，只是还没到生效点）", day.Add(9*time.Hour + 30*time.Minute), false},
{"10:30 有效区间内", day.Add(10*time.Hour + 30*time.Minute), false},
{"11:01 已过失效点", day.Add(11*time.Hour + time.Minute), true},
{"23:00 当日深夜", day.Add(23 * time.Hour), true},
}
for _, c := range cases {
if got := validityExpired(to, c.now); got != c.want {
t.Errorf("%s：validityExpired(%s)=%v，期望 %v", c.name, to, got, c.want)
}
}
// valid_to 为空 = 永久有效（NULL 语义）
if validityExpired("", day.Add(1000*time.Hour)) {
t.Fatal("valid_to 为空应视为永久有效")
}
// 非法格式保守视为未过期（不误伤存量脏数据）
if validityExpired("not-a-time", day) {
t.Fatal("valid_to 解析失败应保守视为未过期")
}
_ = from
}
// TestSupersedeReadSideVisible P1-1：版本/取代关系必须真的进 JSON（不是「加了结构体没接线」）。
func TestSupersedeReadSideVisible(t *testing.T) {
r := models.RecallResult{
ID: "mem_old", Content: "旧事实", Channel: "vector",
Version: 2, ValidFrom: "2026-10-26T10:00:00+08:00", ValidTo: "2026-10-26T11:00:00+08:00",
SupersededBy: []string{"mem_new"}, Superseded: true,
}
b, err := json.Marshal(r)
if err != nil {
t.Fatalf("marshal: %v", err)
}
var m map[string]interface{}
if err := json.Unmarshal(b, &m); err != nil {
t.Fatalf("unmarshal: %v", err)
}
if m["superseded"] != true {
t.Fatalf("superseded 便捷布尔应输出 true：%s", string(b))
}
if got, _ := m["superseded_by"].([]interface{}); len(got) != 1 || got[0] != "mem_new" {
t.Fatalf("superseded_by 未输出：%s", string(b))
}
if m["generation"] != float64(2) {
t.Fatalf("generation（version）未输出：%s", string(b))
}
if m["valid_to"] == "" || m["valid_to"] == nil {
t.Fatalf("valid_to 未输出：%s", string(b))
}
}
go/internal/storage/standing_match.go
+1
View File
@@ -178,6 +178,7 @@ func PrependStandingAnswer(results []models.RecallResult, sa *StandingAnswer, to
Score: 1.0,
Timestamp: sa.UpdatedAt.Format(time.RFC3339),
Source: "standing_answer",
Channel: "standing",
})
for _, r := range results {
if r.ID == id {
go/internal/storage/trace.go
+331
View File
@@ -0,0 +1,331 @@
// 织忆 MemoryWeave — 召回链路追踪（trace_id 贯穿，t_trace_p0）
//
// 背景（CS249r Vol III Ch14 判据）：全仓 grep trace_id|span = 0 ⇒「这条为什么被召回」
// 永远答不出，后续所有调参都是盲修。本文件提供最小可用链路追踪：
//
// ① 每次 recall 生成 trace_id（tr_<16字节hex>），贯穿 编码→粗检索→FTS→重排→MMR→图谱→返回；
// ② 各阶段耗时 + 各通道贡献条数 + 每条结果的来源通道（RecallResult.Channel）；
// ③ GET /api/v1/trace/<trace_id> 可重建整条链路（内存环形缓冲 + JSONL 落盘，重启后仍可查）。
//
// 设计约束：
// - 不引入新依赖（crypto/rand 生成 ID，不新增 uuid 直接依赖）；
// - 落盘异步（有界 channel + 单写协程），recall 主路径只做内存 append，绝不为追踪拖慢召回；
// - 落盘失败只降级为「仅内存」，不报错、不影响召回结果（追踪是观测设施，不是业务依赖）；
// - **这不是缓存**（不参与召回读写路径、不返回旧结果），与 searchcache/Redis 层的职责不同；
// 写一次读多次的观测日志，用环形缓冲 + JSONL 是合适的形态，无需引入 Redis 键。
package storage
import (
"bufio"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"time"
)
// TraceStage 单个阶段耗时。Count 语义随阶段不同（候选数/重排数/扩展数…）。
type TraceStage struct {
Name string `json:"name"`
MS float64 `json:"ms"`
Count int `json:"count,omitempty"`
Note string `json:"note,omitempty"`
}
// TraceItem 单条返回结果的来源归因。
type TraceItem struct {
Rank int `json:"rank"`
ID string `json:"id"`
Channel string `json:"channel"`
Score float64 `json:"score"`
}
// RecallTrace 一次召回的完整链路记录。
type RecallTrace struct {
TraceID string `json:"trace_id"`
Query string `json:"query"`
Namespace string `json:"namespace"`
Mode string `json:"mode"`
Status string `json:"status"` // complete | partial
PartialReason string `json:"partial_reason,omitempty"`
StartedAt time.Time `json:"started_at"`
DurationMS float64 `json:"duration_ms"`
CacheHit bool `json:"cache_hit"`
CandidatesScanned int `json:"candidates_scanned"`
Truncated bool `json:"truncated"`
StandingAnswerHit string `json:"standing_answer_hit,omitempty"`
TopK int `json:"top_k"`
Channels map[string]int `json:"channels"`
Stages []TraceStage `json:"stages"`
Items []TraceItem `json:"items"`
Scope map[string]interface{} `json:"scope,omitempty"`
Error string `json:"error,omitempty"`
started time.Time
}
const (
traceRingCap = 2000
traceMaxFileSize = 32 << 20 // 32MB 轮转
)
// traceDir 可通过 ZHIYI_TRACE_DIR 覆盖（测试用）。
func traceDir() string {
if d := os.Getenv("ZHIYI_TRACE_DIR"); d != "" {
return d
}
return "/var/lib/memoryweave/traces"
}
func tracePersistEnabled() bool {
switch strings.ToLower(os.Getenv("ZHIYI_TRACE_PERSIST")) {
case "0", "false", "off":
return false
}
return true
}
// NewTraceID 生成 tr_<32hex>（16 字节随机，碰撞概率可忽略）。
func NewTraceID() string {
var b [16]byte
if _, err := rand.Read(b[:]); err != nil {
// 极端情况下退化为纳秒时间戳，仍保证非空、可区分。
return fmt.Sprintf("tr_%d", time.Now().UnixNano())
}
return "tr_" + hex.EncodeToString(b[:])
}
// NewRecallTrace 创建一条追踪记录（StartedAt 同时作为计时起点）。
func NewRecallTrace(traceID, query, namespace, mode string) *RecallTrace {
now := time.Now()
if traceID == "" {
traceID = NewTraceID()
}
return &RecallTrace{
TraceID: traceID,
Query: query,
Namespace: namespace,
Mode: mode,
Status: "complete",
StartedAt: now,
Channels: map[string]int{},
Stages: []TraceStage{},
Items: []TraceItem{},
started: now,
}
}
// Stage 记录一个阶段耗时。start 由调用方在阶段开始前取；count<0 表示不记录。
func (t *RecallTrace) Stage(name string, start time.Time, count int) {
if t == nil {
return
}
t.Stages = append(t.Stages, TraceStage{
Name: name,
MS: float64(time.Since(start).Microseconds()) / 1000.0,
Count: count,
})
}
// StageNote 记录带备注的阶段（如失败原因）。
func (t *RecallTrace) StageNote(name string, start time.Time, count int, note string) {
if t == nil {
return
}
t.Stages = append(t.Stages, TraceStage{
Name: name,
MS: float64(time.Since(start).Microseconds()) / 1000.0,
Count: count,
Note: note,
})
}
// MarkPartial 标记本次召回为部分结果（deadline / 行数上限 / 阶段错误）。
func (t *RecallTrace) MarkPartial(reason string) {
if t == nil || reason == "" {
return
}
t.Status = "partial"
if t.PartialReason == "" {
t.PartialReason = reason
} else if !strings.Contains(t.PartialReason, reason) {
t.PartialReason += "," + reason
}
}
// SetItems 用最终返回集填充 item 归因（Rank 从 1 起）。
func (t *RecallTrace) SetItems(items []TraceItem) {
if t == nil {
return
}
t.Items = items
t.Channels = map[string]int{}
for _, it := range items {
ch := it.Channel
if ch == "" {
ch = "unknown"
}
t.Channels[ch]++
}
}
var (
traceMu sync.RWMutex
traceRing = make([]*RecallTrace, 0, traceRingCap)
traceIdx = map[string]*RecallTrace{}
traceFileMu sync.Mutex
traceFile *os.File
traceFileSize int64
traceFileDay string
traceCh = make(chan *RecallTrace, 1024)
traceOnce sync.Once
)
// SaveTrace 写入内存环形缓冲，并异步落盘（非阻塞；缓冲满则丢落盘、保留内存）。
func SaveTrace(t *RecallTrace) {
if t == nil {
return
}
if t.DurationMS == 0 {
t.DurationMS = float64(time.Since(t.started).Microseconds()) / 1000.0
}
traceMu.Lock()
if len(traceRing) >= traceRingCap {
// 覆盖最旧：同时从索引删除
old := traceRing[0]
copy(traceRing, traceRing[1:])
traceRing = traceRing[:len(traceRing)-1]
if old != nil {
delete(traceIdx, old.TraceID)
}
}
traceRing = append(traceRing, t)
traceIdx[t.TraceID] = t
traceMu.Unlock()
if !tracePersistEnabled() {
return
}
traceOnce.Do(func() { go traceWriter() })
select {
case traceCh <- t:
default:
// 落盘队列满：丢这一条（内存仍可查），不阻塞召回主路径。
}
}
// GetTrace 先查内存，再回落到落盘文件（支持服务重启后按 trace_id 回溯）。
func GetTrace(traceID string) (*RecallTrace, bool) {
if traceID == "" {
return nil, false
}
traceMu.RLock()
if t, ok := traceIdx[traceID]; ok {
traceMu.RUnlock()
return t, true
}
traceMu.RUnlock()
return getTraceFromDisk(traceID)
}
// TraceCount 当前内存中可查的 trace 数（自检/测试用）。
func TraceCount() int {
traceMu.RLock()
defer traceMu.RUnlock()
return len(traceRing)
}
func traceWriter() {
for t := range traceCh {
writeTraceLine(t)
}
}
// writeTraceLine 追加一行 JSONL。失败静默（仅内存可查），不向调用方冒泡。
func writeTraceLine(t *RecallTrace) {
line, err := json.Marshal(t)
if err != nil {
return
}
line = append(line, '\n')
dir := traceDir()
if err := os.MkdirAll(dir, 0o755); err != nil {
return
}
day := time.Now().Format("20060102")
traceFileMu.Lock()
defer traceFileMu.Unlock()
if traceFile == nil || traceFileDay != day || traceFileSize > traceMaxFileSize {
if traceFile != nil {
_ = traceFile.Close()
traceFile = nil
}
path := filepath.Join(dir, "recall-"+day+".jsonl")
if traceFileSize > traceMaxFileSize {
// 轮转：同日超大时改名 .1（覆盖上一份）
_ = os.Rename(path, path+".1")
}
f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
if err != nil {
return
}
traceFile = f
traceFileDay = day
if st, err := f.Stat(); err == nil {
traceFileSize = st.Size()
} else {
traceFileSize = 0
}
}
if _, err := traceFile.Write(line); err == nil {
traceFileSize += int64(len(line))
}
}
// getTraceFromDisk 从落地文件逆序扫描（只读，不影响运行中的服务）。
func getTraceFromDisk(traceID string) (*RecallTrace, bool) {
dir := traceDir()
entries, err := os.ReadDir(dir)
if err != nil {
return nil, false
}
needle := `"trace_id":"` + traceID + `"`
// 文件名含日期，倒序 = 新文件先查。
for i := len(entries) - 1; i >= 0; i-- {
e := entries[i]
if e.IsDir() || !strings.HasPrefix(e.Name(), "recall-") {
continue
}
f, err := os.Open(filepath.Join(dir, e.Name()))
if err != nil {
continue
}
var hit *RecallTrace
sc := bufio.NewScanner(f)
sc.Buffer(make([]byte, 0, 64*1024), 4<<20)
for sc.Scan() {
b := sc.Bytes()
if !strings.Contains(string(b), needle) {
continue
}
var t RecallTrace
if json.Unmarshal(b, &t) == nil && t.TraceID == traceID {
hit = &t
}
}
_ = f.Close()
if hit != nil {
return hit, true
}
}
return nil, false
}
go/internal/storage/trace_test.go
+160
View File
@@ -0,0 +1,160 @@
// 织忆 MemoryWeave — 召回链路追踪单元测试（P0-1）
//
// 验收要点：
// - trace_id 唯一、可查（内存 + 落盘两条路径）；
// - 阶段/通道/条目归因完整（100 条全可重建，等价线上「100 请求 100% 可回溯」的 CI 断言）；
// - 环形缓冲按容量淘汰最旧且索引同步删除。
package storage
import (
"fmt"
"os"
"path/filepath"
"testing"
"time"
)
func resetTraceState(t *testing.T) {
t.Helper()
traceMu.Lock()
traceRing = make([]*RecallTrace, 0, traceRingCap)
traceIdx = map[string]*RecallTrace{}
traceMu.Unlock()
}
// TestRecallTrace_100RequestsAllRetrievable 对应验收：「CI 断言 100 个请求 100% 可重建调用链」。
func TestRecallTrace_100RequestsAllRetrievable(t *testing.T) {
resetTraceState(t)
t.Setenv("ZHIYI_TRACE_PERSIST", "0")
dir := t.TempDir()
t.Setenv("ZHIYI_TRACE_DIR", dir)
ids := make([]string, 0, 100)
for i := 0; i < 100; i++ {
tr := NewRecallTrace("", fmt.Sprintf("query-%d", i), "hermes-main", "hybrid")
tr.Stage("encode", time.Now(), 0)
tr.Stage("coarse_search", time.Now(), 200)
tr.Stage("graph_expand", time.Now(), 2)
tr.SetItems([]TraceItem{
{Rank: 1, ID: "mem_a", Channel: "vector", Score: 0.9},
{Rank: 2, ID: "mem_b", Channel: "fts", Score: 0.8},
{Rank: 3, ID: "mem_c", Channel: "graph", Score: 0.7},
{Rank: 4, ID: "mem_d", Channel: "standing", Score: 1.0},
{Rank: 5, ID: "mem_e", Channel: "vector", Score: 0.6},
})
SaveTrace(tr)
ids = append(ids, tr.TraceID)
}
if TraceCount() != 100 {
t.Fatalf("应保存 100 条 trace，实际 %d", TraceCount())
}
for _, id := range ids {
got, ok := GetTrace(id)
if !ok {
t.Fatalf("trace %s 不可回溯", id)
}
if len(got.Items) != 5 {
t.Fatalf("trace %s item 归因应 5 条，实际 %d", id, len(got.Items))
}
if len(got.Stages) < 3 {
t.Fatalf("trace %s 阶段应 ≥3，实际 %d", id, len(got.Stages))
}
for _, it := range got.Items {
if it.Channel == "" {
t.Fatalf("trace %s item %s 缺 channel（无法回答「它为什么被召回」）", id, it.ID)
}
}
}
}
// TestRecallTrace_DiskFallback 服务重启后仍可按 trace_id 回溯（内存清空 → 读 JSONL）。
func TestRecallTrace_DiskFallback(t *testing.T) {
resetTraceState(t)
dir := t.TempDir()
t.Setenv("ZHIYI_TRACE_DIR", dir)
t.Setenv("ZHIYI_TRACE_PERSIST", "1")
tr := NewRecallTrace("", "落盘回溯测试", "hermes-main", "hybrid")
tr.SetItems([]TraceItem{{Rank: 1, ID: "mem_x", Channel: "vector"}})
tr.Status = "complete"
// 直写文件（避免依赖异步协程时序），模拟落盘后的状态。
writeTraceLine(tr)
if _, ok := GetTrace(tr.TraceID); !ok {
t.Fatalf("落盘 trace %s 不可从文件回溯", tr.TraceID)
}
// 再验证一次真实异步路径（轮询最多 2s）
tr2 := NewRecallTrace("", "异步落盘", "hermes-main", "hybrid")
SaveTrace(tr2)
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if _, ok := getTraceFromDisk(tr2.TraceID); ok {
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("异步落盘 2s 内未完成（trace %s）", tr2.TraceID)
}
// TestRecallTrace_RingEviction 环形缓冲满后淘汰最旧，且索引同步删除。
func TestRecallTrace_RingEviction(t *testing.T) {
resetTraceState(t)
t.Setenv("ZHIYI_TRACE_PERSIST", "0")
first := NewRecallTrace("", "first", "ns", "hybrid")
SaveTrace(first)
for i := 0; i < traceRingCap; i++ {
SaveTrace(NewRecallTrace("", fmt.Sprintf("q-%d", i), "ns", "hybrid"))
}
if _, ok := GetTrace(first.TraceID); ok {
t.Fatalf("环形缓冲应淘汰最旧的 trace")
}
if TraceCount() != traceRingCap {
t.Fatalf("环形缓冲容量应保持 %d，实际 %d", traceRingCap, TraceCount())
}
}
// TestNewTraceID_Unique 1000 个 ID 不重复、带 tr_ 前缀。
func TestNewTraceID_Unique(t *testing.T) {
seen := map[string]bool{}
for i := 0; i < 1000; i++ {
id := NewTraceID()
if len(id) < 10 || id[:3] != "tr_" {
t.Fatalf("trace id 形态非法：%q", id)
}
if seen[id] {
t.Fatalf("trace id 重复：%q", id)
}
seen[id] = true
}
}
// TestRecallTrace_MarkPartial partial 原因必须可读（不覆盖首个原因）。
func TestRecallTrace_MarkPartial(t *testing.T) {
tr := NewRecallTrace("tr_test", "q", "ns", "hybrid")
if tr.Status != "complete" {
t.Fatalf("新 trace 状态应为 complete，实际 %q", tr.Status)
}
tr.MarkPartial("deadline_exceeded")
tr.MarkPartial("row_cap_truncated")
if tr.Status != "partial" {
t.Fatalf("MarkPartial 后状态应为 partial，实际 %q", tr.Status)
}
if tr.PartialReason != "deadline_exceeded,row_cap_truncated" {
t.Fatalf("partial_reason 应累积，实际 %q", tr.PartialReason)
}
}
// TestTraceDir_Default 默认目录形态（不创建文件，只校验路径选择逻辑）。
func TestTraceDir_Default(t *testing.T) {
os.Unsetenv("ZHIYI_TRACE_DIR")
if got := traceDir(); got != "/var/lib/memoryweave/traces" {
t.Fatalf("默认 trace 目录错误：%s", got)
}
t.Setenv("ZHIYI_TRACE_DIR", filepath.Join(t.TempDir(), "x"))
if got := traceDir(); got == "/var/lib/memoryweave/traces" {
t.Fatalf("ZHIYI_TRACE_DIR 未生效")
}
}