Compare commits

...

16 Commits

Author SHA1 Message Date
小唯 995e93598a fix(tests): 修 pre-existing 测试失配,恢复 make test 全绿 (t_8496e8b6)
四处均为本任务之前就存在的测试漂移(与优化B 无关,但让 go test ./... 无法通过):

1. selfoptimize: ae73982 给 NewGapDetector 增加了 (emb, ldb) 两个参数,
   8 处测试调用点未跟进,整包 test 构建失败。测试只用到 RecordMiss/Close/List,
   不需要真实依赖,传 nil, nil 即可。
2. integration_test.go:296 selfoptimize.NewGapDetector() 同上。
3. integration_test.go:323 InMemoryGraph.Navigate 新增 relFilter 参数,补 nil。
4. integration_test.go:339 /api/v1/triggers/fire 的夹具用了 "t1"(那是 executor
   内部编号),routes.Triggers 的真实 ID 是 t_decay/t_distill/t_merge 等,
   导致 Fire 返回 404 trigger not found。夹具改用 t_decay。

验证: go vet ./... 无输出; go test ./... -count=1 全部 ok(根包 8.1s)
2026-09-12 18:07:40 +08:00
小唯 b9107fa925 docs+test(recall): 修正批量更新注释(delta 分组,非 CASE WHEN)+ 补 IPC 只读校验用例 (t_8496e8b6)
- recall_write_buffer.go / lancedb_ipc.go: 注释与实现对齐 —— lance 不支持 CASE WHEN,
  Rust 侧按 delta 分组、每组一次 update 提交;整批版本数 = 不同 delta 的个数(通常 1 个)
- batch_ipc_integration_test.go: +TestBatchIPCScanReadOnly(只读扫 socket 校验 recall_count 真落盘)

实测(临时 sidecar,/tmp/mw-verify-180517):
  VERSION_ACCOUNTING: 批量3条 → 1 个版本 | 逐条3次 → 3 个版本
  BATCH_IPC_OK id=ep_1786982837511335732 recall_count 0 → 3
  SCAN_READONLY: 总 1814 条, recall_count>0 的 742 条, 最大 828
2026-09-12 18:07:30 +08:00
小怡 2e04adf054 perf(recall): 累积延迟更新治理写放大 — 读路径零逐条写 + 单事务批量落盘 (t_8496e8b6)
根因: recall 读路径对每条结果同步 lancedb.Update,LanceDB MVCC 每次提交=1 版本
      实测 15 版本/分钟 / _versions 17.7G(真实数据 22M)

改动:
- Go: 新增 RecallWriteBuffer(窗口合并+阈值触发+优雅退出落盘);recall.go 读路径改 Record();RustLanceDBClient.UpdateRecallBatch
- Rust: lancedb_update_batch IPC — 按 delta 分组,每组一次 update(`recall_count + delta` + id IN (...)) 提交
- 单测 6 个 + IPC 端到端版本计数测试;bench_test.go 修 NewReranker 签名失配(阻塞包测试)

实测: 批量 3 条 → 1 个版本;逐条 3 次 → 3 个版本 (lance 不支持 CASE WHEN,已按 delta 分组规避)
2026-09-12 17:57:31 +08:00
xiaowei 38c31eeede fix(recall): 笔记知识对召回不可见 —— 补搜 wiki-curator-main
问题: wiki_curator 把「笔记→知识」写进 namespace=wiki-curator-main,
而 recall 只搜 req.Namespace 与 shared ⇒ 33+ 条笔记知识长期在盲区,
用户问"笔记里的资料"永远召回不到。

改动: internal/api/routes/core.go 在 shared 补搜之后增加额外命名空间补搜
(wiki-curator-main), 去重 + 降权 0.45 (不挤掉主记忆但能被看到)。

踩坑: 初版条件写成 len(results) < req.Limit 才补搜 —— 主库记忆充足时
该条件永不成立, 补搜形同虚设。改为无条件补搜后生效。

验证: 部署后实测 recall「Hermes平台核心配置文件」→ 11 条,
含笔记知识「Hermes平台的核心配置文件包括SOUL.md、AGENTS.md…」

已知未修(预先存在, 与本次无关): selfoptimize/storage 的 bench_test.go
调用签名已变更的 NewGapDetector/NewReranker 导致 test build failed。
2026-09-11 21:22:22 +08:00
小唯 2504174a75 fix(security): 写入侧凭证过滤 — 密钥不再进记忆/图谱/Obsidian 镜像
背景(2026-09-10 mc P0 事故):某轮蒸馏把含真 Tailscale 预授权 key 的原始对话
当事实入库 → 知识图谱按其文本建实体 → ObsidianSyncer 用实体标题当文件名/正文,
把 tskey-auth-… 落成可读镜像文件,又被 git 跟踪并推送 Gitea。
本提交在三个写入边界统一加脱敏。

新增 internal/redact(唯一真源):
- SecretPattern 覆盖 sk- / tskey- / github_pat_ / ghp_ / nvapi- / AKIA
- Go RE2 不支持 lookbehind,故"前一字符是否属于更长标识符"手工判定(isTokenByte)
- sk- 段只吃 [A-Za-z0-9],避免误伤 URL 里的 task-driven 片段(事故中的假阳性来源)
- RedactSecrets 幂等;ContainsSecret 供"直接丢弃"场景用

四处接入:
- distill/quality.go       IsQualityFact 判定前脱敏 + FilterQualityFacts 返回值脱敏
- governance/graph_auto.go cleanEntityName 命中→空串;UpdateFromDistill 只对有效实体
                           建节点/建边;extractEntitiesFromText 过滤;含凭证的 fact 不建节点
- api/routes/obsidian.go   sanitizeFilename 入参脱敏 + PushToObsidian 镜像正文脱敏
- api/server.go            蒸馏直写记忆库路径(绕过 FilterQualityFacts 的那条)同步脱敏

测试:13 条新用例(redact 5 / distill 3 / governance 4 / routes 1),
含 URL 假阳性回归与幂等性断言。全部使用构造的假密钥,不含任何真实凭证。
证据:go build ./... OK;
go test -count=1 ./internal/redact/ ./internal/distill/ ./internal/governance/ ./internal/api/routes/ 全绿。

顺带修既有 test 腐化(否则 governance 包根本无法编译测试,非本任务引入):
governance_test.go isContradiction→IsContradiction、Navigate 补 relFilter 参数;
bench_test.go 同。仅测试文件,无生产行为改动。
2026-09-11 00:41:59 +08:00
小唯 e9a07cd678 fix(obsidian-sync): 文件名 slug 不再把中文替换成下划线 + rune 级截断 + 撞名消歧
根因(kanban t_5f9c56ef):ObsidianSyncer.PushToObsidian 生成的镜像文件名全废 —
sanitizeFilename() 逐个 rune 判断 "是否 ASCII",非 ASCII(含全部中文)一律替换为 _,
于是 "修复了respond重复声明问题" 变成 "____respond______.md";实测 mc vault
小唯/07-Wiki/织忆 1628 个文件 + 小唯/02-Memory 59 个文件全是这种名字,Obsidian 里无法按名浏览。

改动:
- sanitizeFilename: 只替换文件系统非法字符(/ \ : * ? " < > | 控制符等),
  保留 Unicode 字母/数字(中文正常入名);连续非法字符折叠成一个 _;trim 首尾 " ._-"
- 去掉 content[:30] 的字节切片(会把多字节字符切半个 → 乱码),改用 truncateRunes()
- truncate() 从字节计数改为 rune 计数(H1 标题行同样会被切出乱码)
- 撞名消歧:同一批推送里 base 重名时补 8 位内容哈希后缀,不再静默互相覆盖
- 删除已无用的 minz()

顺带修复(原本就让整个 routes 包测试编译不过,不修则无法验证):
- routes_test.go: selfoptimize.NewGapDetector() 缺参 → (nil, nil)
- routes_test.go: 删掉引用已移除的 SSEBus/SSEMessage/PushMemoryCommitted/PushGapFound 的用例,
  改为按当前 WebSocket API 验证全部 Push* helper 不 panic
- routes_test.go: TestTriggers_List 期望 4 → 8(triggers.go 实际注册 8 个)

验证:go build ./... 通过;go vet ./internal/api/routes/ 通过;
go test ./internal/api/routes/ 全绿(此前 build failed)。
新增 obsidian_test.go 锁死回归:中文标题不得出现连续下划线、截断结果必须合法 UTF-8、同名不覆盖。
实际落盘文件名样例:
  LanceDB 已设天花板_不会再膨胀到 35G.md / 修复了 respond 重复声明问题.md /
  同一段开头的内容 abcdefghijklmnopqrstu.md / 同一段开头的内容 abcdefghijklmnopqrstu-4bad72a0.md

⚠️ 未部署:需重建 /home/muc/bin/zhiyid-new 并重启 zhiyid.service 才生效(另开任务卡跟踪)。
2026-09-10 23:15:01 +08:00
小唯 3872294045 fix(graph-cleanup): CleanupNoiseNodes NULL-id panic + 新增 scoped 卫安/A03/A04 测试污染清理端点 (t_1dfaaa4a)
- graph_sqlite.go CleanupNoiseNodes: row[id] 为 NULL(daemon-distill pattern 模板行)时直接
  .(string) panic → HTTP 连接被关(HTTP=000);改为 nil 安全跳过
- 新增 CleanupScopedNodes(dryRun, namespaces, nameContains):namespace+名称子串精确圈定,
  清理 2026-06 多 agent 测试污染(openclaw-main/a06-main/hermes-main 的 A03/A04/卫安 节点)
- graph_cache.go: cachedGraphStore 透传(inner 可选接口断言,非 SQLite 后端静默 0)
- server.go: GET/POST /api/v1/graph/cleanup/scoped?dry_run=&namespace=&name=
  实际执行后 InvalidateAll 保证 graph_cache 一致性 —— 禁外部 SQL 的官方通道
2026-09-08 09:02:35 +08:00
小唯 4714efaab9 fix(distill-quality): 严格质量门槛 1b (t_cc07ef4b) — <20字/纯疑问句/状态汇报拒绝; 20-50需实体或动词
方案B 严格版(替代 946f0c9 宽松版): quality.go IsQualityFact 重写
- <20字一律拒绝(无用户信号豁免, 与质量看门狗 FRAG_LEN=20 对齐 → AC-2 碎片率<10%)
- 纯疑问句拒绝(带?/?、疑问前缀+吗呢么、无陈述主语短问句)
- 状态汇报型拒绝(无主状态句/健康检查/桥接/reflection/完成重启类)
- 20-50字需含实体(织忆/hermes/牧尘/大写词/数字)或动词短语才入库
engine.go: distillOne 源头过滤(图谱/AAAK/记忆库同源干净) + fallbackSingle 过滤
server.go 写回 IsQualityFact 双保险(同一函数自动变严格)

单测 quality_test.go + quality_engine_test.go (mock LLM 短对话episode → 0碎片入库)
AC-1 dev 验证: 构造含短对话episodes → 蒸馏后碎片不入库 
2026-09-07 21:50:37 +08:00
xiaowei 4130e6b668 feat(memory-quality): 方案B写时质量门槛+方案C provenance 标注
方案B: distill 结果写 memories 前 IsQualityFact 过滤(碎片<8字拒/8-20字需用户信号/噪声前缀拦)+完全重复查重跳过
  - quality.go: 与 LightMem 保留细节哲学兼容(咖啡偏好保留), 拦进程噪声/桥接/reflection
  - 单测 8/8 通过
方案C: distill MemoryRecord 标 Source=llm_distill(provenance); volatile_flag 降权已确认生效(×0.5)
  - ContentMD5/VersionHistory 字段已在, bi-temporal valid_to 加列缓行(风险高)
2026-09-07 21:50:37 +08:00
xiaowei a5b5e217f3 fix(forgetting): 遗忘机制方案A — 碎片快速道+episodes/长内容保护+stats修正
- governance.go: recall加成 0.05→0.005(防 recall_count 500+ 虚高保命); decayRate 试 0.03 误删长内容后回滚 0.015
- server.go decay: 碎片快速道(<30字&>20天→auto_forget_fragment); episodes 不参与遗忘(曾误删67条对话已恢复); >200字长记忆保护
- admin.go Forget: 同步三保护
- rust lancedb_ops.rs: tombstone_count 数 is_deleted=true(移植 f6c4383+类型修)

验证: admin forget 触发 tombstone 0→210(碎片清除); episodes/长内容 0 误删; stats 真实计数
2026-09-07 21:50:37 +08:00
小唯 1bf23c42e8 fix(stats): count_rows filter 类型修正 Option<&str> -> Option<String>
f6c4383 的 count_rows(Some("is_deleted = true")) 编译失败(E0308: 此 lance 版本
签名 Option<String>)。修正为 .to_string()。实测 stats tombstone_count=830
(memories 表 is_deleted=true 行数, 之前恒 0)。
2026-09-06 09:25:27 +08:00
小唯 f6c4383f98 fix(stats): tombstone_count 统计源修正 — 数 memories is_deleted=true 行
SoftDelete 持久化语义 = memories 表内标记 is_deleted=true(不写独立 tombstones 表),
但 Rust stats() 的 tombstone_count 数的是空 tombstones 表 → 恒 0(指标失真)。
修正:stats() 对 memories 表 count_rows(Some("is_deleted = true")) 作为 tombstone_count。
验证: DELETE 一条记忆后 stats tombstone_count 应 >0。
2026-09-06 09:22:49 +08:00
xiaowei f6cd93fbd7 fix(forgetting): P2 候选源升级 — Rust scan 全表替代纯缓存遍历
GetCandidatesForForgetting 原实现只遍历进程内 _local.memories 缓存
(commit/recall/search 才触碰), 重启后缓存空 → decay scanned=3 扫不全,
9-25 到期记忆若不在缓存则永远无法被遗忘(闭环断)。

安全版 scan_for_forgetting (防 febc2c9 风暴复发):
- 强制 limit (IPC 钳制硬上限 5000, decay 侧 2000)
- 不读 vector 列 (1024维×3000条≈12MB+, IPC JSON 会卡)
- 真读 last_recalled_at 列 (scan_all 旧实现 String::new() 恒空 → 遗忘判定失真)
- 消费方截断护栏已就位 (merge 桶500 / conflict 500 / sanity 1000)
- IPC 失败/空 → fallback 缓存遍历 (保持可用)

验证: [decay] scanned=3 → 2000 (重启即扫全表), CPU 0%, 无 panic
2026-09-06 02:32:58 +08:00
xiaowei 6c1c21fadf perf(pagerank): O(V+E) 累加实现替代 O(V²) 全源遍历
旧实现每目标节点遍历全部源节点找边匹配 (943-961):
- 17628 nodes × 20 迭代 = 6.2亿次内层操作 → 单次 PageRank ~3 分钟
- 重启后多套触发器零值连锁 fire full consolidation → CPU 100% 持续 10+ 分钟

新实现先归一化每源节点总权重, 再沿出边把贡献累加到目标 (O(V+E)):
- 17628 nodes PageRank + 图谱维护: 3分钟 → <1 秒
- 重启风暴从 10+ 分钟 CPU 100% → 秒级无感

实测: 01:16:40 图谱维护开始 → 01:16:41 PageRank 更新 17628 nodes 完成
2026-09-06 01:20:21 +08:00
xiaowei e67f0bc7a6 fix(recall): Embedder http client 8s 超时 + decay 分批 AC-1 日志
- Embedder http.Client{} 无超时 → 远端 bge 半开 TCP 时 recall 永久挂起
  (gateway 10s 断 → 织忆异常误报)。加 8s 超时快速失败降级。
- decay 每 tick 限量 2000 条(最久未访问优先) + scanned/forgotten 日志
  (AC-1 验收证据)
2026-09-06 01:08:42 +08:00
xiaowei 4376f1f2ce fix(forgetting): P1 防风暴 - 限长/分桶/抽样下推/decay分批
febc2c9 全表扫描 O(n²) 风暴的根治防护(生产基线 506f76b+ 之上,不含 febc2c9 扫描代码):

1. mergeSimilar 按 category 分桶 + 单桶 maxMergePerBucket=500 截断
   - 原双层循环全量 O(n²),全表时触发 620% CPU
   - 分桶后 O(Σ桶²) << O(n²),语义不变(原 catI!=catJ continue 等价分桶)
2. scanConflicts 显式 created_at 倒序取最近 500 条
   - 原注释'最近100条'实际全量 O(n·50)
3. checkTimestampSanity 先截断 maxSanitySample=1000 再均匀抽样
   - 原全量拉取仅为 5% 抽样
4. decay 每 tick 限量 maxDecayBatch=2000,按最久未访问优先
5. 存储层输出契约统一 RFC3339 string + 补 category/created_at
   - lancedb_ipc: last_recalled_at 此前直接放 time.Time → server.go .(string)
     断言恒失败 → lastAccessed 永远默认30天前 → 遗忘判定失真(本 bug + 数据
     年龄未到 = 遗忘从不触发的双根因之一)
   - sqlite: 补 category/created_at + LIMIT 100→200(对齐 lancedb 契约)
   - 新增 memoryTimeVal/fmtTimeRFC3339 helper 兼容多类型

测试: go build ./... 全绿;gofmt 仅历史文件遗留(未全文件重排,保持最小 diff)
2026-09-06 00:56:06 +08:00
34 changed files with 2408 additions and 130 deletions

View File

@ -11,6 +11,7 @@ import (
"time"
"github.com/xiaoxue/memoryweave/internal/api"
"github.com/xiaoxue/memoryweave/internal/storage"
)
func main() {
@ -34,6 +35,8 @@ func main() {
<-sigCh
log.Println("[zhiyid] 收到关闭信号,正在退出...")
// 累积延迟更新:把最后一个窗口的召回元数据落盘(防止统计增量丢失)
storage.StopRecallWriteBuffer()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(ctx); err != nil {

View File

@ -293,7 +293,7 @@ func TestEdge_ConcurrentCommits(t *testing.T) {
}
func TestEdge_GapDetectionThreshold(t *testing.T) {
gd := selfoptimize.NewGapDetector()
gd := selfoptimize.NewGapDetector(nil, nil)
// 2 misses — no gap
if gap := gd.RecordMiss("test"); gap != nil {
@ -320,7 +320,7 @@ func TestEdge_GraphEmpty(t *testing.T) {
t.Error("empty graph should return all zeros")
}
paths, err := g.Navigate("nonexistent", 2, "shared")
paths, err := g.Navigate("nonexistent", 2, "shared", nil)
if err != nil {
t.Errorf("navigate on empty graph should not error: %v", err)
}
@ -336,7 +336,8 @@ func TestEdge_TriggersFired(t *testing.T) {
"X-API-Key": "zhiyi-dev-key-2026",
}
body := map[string]string{"trigger_id": "t1"}
// 触发器 ID 与 routes.Triggers 一致t1..t8 是 executor 的内部编号,不是 API ID
body := map[string]string{"trigger_id": "t_decay"}
req := httptest.NewRequest("POST", "/api/v1/triggers/fire", bytesBody(body))
for k, v := range header {
req.Header.Set(k, v)

View File

@ -72,6 +72,22 @@ func (aa *AdminAPI) Forget(w http.ResponseWriter, r *http.Request) {
tier := strVal(mem["tier"])
content := strVal(mem["content"])
// 🔒 2026-09-07 修复(误删事故): episodes(原始对话)不参与遗忘
if cat := strVal(mem["category"]); cat == "episodes" {
continue
}
// 🔒 长内容保护: >200 字记忆不参与 auto 遗忘(有实质信息)
if len([]rune(content)) > 200 {
continue
}
// 2026-09-07 方案A 碎片快速道(与 server.go decay trigger 一致):
// <30 字且 >20 天未访问 → 直接遗忘(碎片 recall_count 可能虚高, 绕过保命)
if len([]rune(content)) < 30 && time.Since(lastAccess).Hours()/24 > 20 {
aa.LanceDB.SoftDelete(strVal(mem["id"]), "auto_forget_fragment")
forgotten++
continue
}
// E4.3 修复: 从 content 提取实体(而非读 namespace取图谱最大度
degree := extractTopEntityDegree(content, aa.GraphStore)

View File

@ -4,6 +4,7 @@ package routes
import (
"fmt"
"log"
"sort"
"strings"
"time"
@ -207,7 +208,10 @@ func (cp *ConsolidationPipeline) RunMerge() (int, error) {
return cp.mergeSimilar()
}
// mergeSimilar 合并相似记忆(向量相似度 > 0.8 → 保留最新)
// mergeSimilar 合并相似记忆(相同 category + 高内容重叠 → 保留较新)
// 2026-09-06 P1 fix: 原实现双层循环全量 O(n²) —— febc2c9 全表扫描时触发 620% CPU 风暴。
// 改为按 category 分桶后桶内两两比较(原逻辑 catI != catJ continue 本就等价),
// 单桶超 maxMergePerBucket 截断,总计算量 O(Σ桶²) << O(n²)。
func (cp *ConsolidationPipeline) mergeSimilar() (int, error) {
// 获取全部 distilled 记忆
memories, err := cp.ldb.GetCandidatesForForgetting()
@ -215,23 +219,36 @@ func (cp *ConsolidationPipeline) mergeSimilar() (int, error) {
return 0, err
}
const maxMergePerBucket = 500
merged := 0
// 简单启发式:相同 category + 高内容重叠 → 合并
for i := 0; i < len(memories); i++ {
for j := i + 1; j < len(memories); j++ {
catI := strVal(memories[i]["category"])
catJ := strVal(memories[j]["category"])
if catI != catJ {
continue
}
contentI := strVal(memories[i]["content"])
contentJ := strVal(memories[j]["content"])
if overlap := contentOverlap(contentI, contentJ); overlap > 0.8 {
// 保留较新的
idI := strVal(memories[i]["id"])
idJ := strVal(memories[j]["id"])
_ = cp.ldb.SoftDelete(idJ, fmt.Sprintf("merged_into_%s", idI))
merged++
// 按 category 分桶
buckets := make(map[string][]map[string]interface{})
for _, m := range memories {
cat := strVal(m["category"])
buckets[cat] = append(buckets[cat], m)
}
// 桶内两两比较(同桶才可能合并,语义与原全量双层一致)
for _, mems := range buckets {
if len(mems) > maxMergePerBucket {
// 单桶爆炸防护:只处理最近 maxMergePerBucket 条created_at 倒序)
sort.SliceStable(mems, func(a, b int) bool {
return memoryInt64Val(mems[a], "created_at") > memoryInt64Val(mems[b], "created_at")
})
mems = mems[:maxMergePerBucket]
}
for i := 0; i < len(mems); i++ {
for j := i + 1; j < len(mems); j++ {
contentI := strVal(mems[i]["content"])
contentJ := strVal(mems[j]["content"])
if overlap := contentOverlap(contentI, contentJ); overlap > 0.8 {
// 保留较新的
idI := strVal(mems[i]["id"])
idJ := strVal(mems[j]["id"])
_ = cp.ldb.SoftDelete(idJ, fmt.Sprintf("merged_into_%s", idI))
merged++
}
}
}
}
@ -239,12 +256,21 @@ func (cp *ConsolidationPipeline) mergeSimilar() (int, error) {
}
// Step 2: 扫描所有待解决冲突
// 2026-09-06 P1 fix: 原注释"最近 100 条"但实际对全量做实体提取+对比 → 全表时 O(n·50)。
// 显式按 created_at 取最近 maxConflictScan 条再扫。
func (cp *ConsolidationPipeline) scanConflicts() (int, error) {
// 获取最近 100 条记忆
// 获取候选并按最近优先截断created_at 倒序)
memories, err := cp.ldb.GetCandidatesForForgetting()
if err != nil {
return 0, err
}
const maxConflictScan = 500
if len(memories) > maxConflictScan {
sort.SliceStable(memories, func(a, b int) bool {
return memoryInt64Val(memories[a], "created_at") > memoryInt64Val(memories[b], "created_at")
})
memories = memories[:maxConflictScan]
}
count := 0
for i := 0; i < len(memories); i++ {
@ -305,17 +331,28 @@ func (cp *ConsolidationPipeline) updateGraph() (int, error) {
return before - after, nil
}
// checkTimestampSanity 抽样检查记忆时间戳合理性(5% sampling,不阻塞)
// checkTimestampSanity 抽样检查记忆时间戳合理性(均匀抽样,不阻塞)
// 返回 (可疑数量, 抽样总数)。epoch-0< 2024-01-01视为可疑。
// 2026-09-06 P1 fix: 原实现全量拉取只为 5% 抽样 → 先截断最近 maxSanitySample
// 条再做均匀 step 抽样,避免全量内存/IO 与排序成本。
func (cp *ConsolidationPipeline) checkTimestampSanity() (suspicious, total int) {
const epochThreshold int64 = 1704067200 // 2024-01-01 00:00:00 UTC
const maxSanitySample = 1000
memories, err := cp.ldb.GetCandidatesForForgetting()
if err != nil || len(memories) == 0 {
return 0, 0
}
// 全量不超过 100 条
// 超过 maxSanitySample 时按 created_at 取最新,再均匀抽样
if len(memories) > maxSanitySample {
sort.SliceStable(memories, func(a, b int) bool {
return memoryInt64Val(memories[a], "created_at") > memoryInt64Val(memories[b], "created_at")
})
memories = memories[:maxSanitySample]
}
// 抽样最多 100 条
sampleSize := len(memories)
if sampleSize > 100 {
sampleSize = 100

View File

@ -323,6 +323,35 @@ func (a *API) Recall(w http.ResponseWriter, r *http.Request) {
}
}
// 2026-09-11: 笔记知识补充搜索。
// wiki_curator 把「笔记→知识」写进 wiki-curator-main而主召回只搜
// req.Namespace/shared ⇒ 33+ 条笔记知识长期在盲区(召回永远看不到)。
// 这里补搜:主结果不足 top_k 时,从额外命名空间补齐(去重 + 降权 0.45)。
if req.Namespace != "wiki-curator-main" {
if vec, encErr := a.Embedder.EncodeSingle(req.Query); encErr == nil {
for _, extra := range []string{"wiki-curator-main"} {
extraHits, _ := a.LanceDB.Search("memories", vec, 5, extra)
for _, m := range extraHits {
dup := false
for _, r := range results {
if r.ID == m.ID {
dup = true
break
}
}
if !dup {
results = append(results, models.RecallResult{
ID: m.ID,
Content: m.Content,
Category: m.Category,
Score: 0.45, // 笔记知识降权:不挤掉主记忆,但能被看到
})
}
}
}
}
}
// Record recall hit rate (has results = hit, empty = miss)
selfoptimize.Dash.RecordRecall(len(results) > 0)
// Gap auto-close: recall 命中后自动关闭该 query 对应的 open gap

View File

@ -2,6 +2,8 @@
package routes
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"log"
@ -11,7 +13,9 @@ import (
"strings"
"sync"
"time"
"unicode"
"github.com/xiaoxue/memoryweave/internal/redact"
"github.com/xiaoxue/memoryweave/internal/storage"
)
@ -52,14 +56,29 @@ func (s *ObsidianSyncer) PushToObsidian(memories []map[string]interface{}, folde
}
count := 0
used := make(map[string]int, len(memories))
for _, mem := range memories {
content := strVal(mem["content"])
// 2026-09-10 P0镜像正文同样脱敏——旧事故里正文与文件名都把
// tskey-auth-… 原样落盘(小唯/07-Wiki/织忆/未找到命令.md
content = redact.RedactSecrets(content)
if len(content) < 5 {
continue
}
filename := sanitizeFilename(content[:minz(len(content), 30)]) + ".md"
filepath := filepath.Join(targetDir, filename)
// 2026-09-10 修复:旧实现 content[:30] 按字节切 + sanitizeFilename 把非 ASCII
// 全替换成 '_',中文标题直接变 ____123.md07-Wiki/织忆 1628 个乱码文件名成因),
// 且不同记忆撞名会静默互相覆盖。改为 rune 级截断 + 保留 Unicode + 撞名加短哈希。
base := sanitizeFilename(truncateRunes(content, 30))
if base == "" {
base = "untitled"
}
if used[base] > 0 {
base = base + "-" + shortHash(content)
}
used[base]++
filename := base + ".md"
target := filepath.Join(targetDir, filename)
mdContent := fmt.Sprintf(`---
id: %s
@ -80,7 +99,7 @@ sync_time: %s
truncate(content, 60),
content)
if err := os.WriteFile(filepath, []byte(mdContent), 0644); err != nil {
if err := os.WriteFile(target, []byte(mdContent), 0644); err != nil {
log.Printf("[obsidian] 写入失败 %s: %v", filename, err)
continue
}
@ -206,29 +225,61 @@ func (s *ObsidianSyncer) StatusHandler(w http.ResponseWriter, r *http.Request) {
// ─── 辅助 ─────────────────────────────────────────────────
// sanitizeFilename 把标题转成文件系统安全的文件名片段,并保留 Unicode中文可读性。
//
// 2026-09-10 修复:旧实现把所有非 ASCII rune 逐个替换为 '_',于是中文标题全变成
// `____123.md`,在 Obsidian 里无法按名浏览07-Wiki/织忆 1628 个 legacy 文件的成因)。
// 现在只替换文件系统非法字符(/ \ : * ? " < > | 控制符等),连续的非法字符折叠成一个 '_'。
// 返回空串表示内容里没有任何可用的安全字符,由调用方兜底。
func sanitizeFilename(s string) string {
s = strings.Map(func(r rune) rune {
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') ||
(r >= '0' && r <= '9') || r == '_' || r == '-' || r == ' ' {
return r
// 2026-09-10 P0入参先脱敏防任何路径把凭证带进文件名
s = redact.RedactSecrets(s)
var b strings.Builder
prevIllegal := false
for _, r := range s {
if unicode.IsLetter(r) || unicode.IsDigit(r) ||
r == '-' || r == '_' || r == ' ' || r == '.' {
prevIllegal = false
b.WriteRune(r)
continue
}
return '_'
}, s)
return strings.TrimSpace(s)
// 文件系统非法字符(/ \ : * ? " < > | 控制符等)+ 其它符号 → 折叠成一个 '_'
if !prevIllegal {
b.WriteRune('_')
prevIllegal = true
}
}
return strings.Trim(b.String(), " ._-")
}
// shortHash 取内容摘要前 8 位十六进制,仅用于同名文件消歧,不做安全用途。
func shortHash(s string) string {
sum := sha256.Sum256([]byte(s))
return hex.EncodeToString(sum[:4])
}
// truncateRunes 按 rune而非字节截断避免把多字节字符切掉一半产生乱码。
func truncateRunes(s string, maxRunes int) string {
if maxRunes <= 0 {
return ""
}
n := 0
for i := range s {
if n == maxRunes {
return s[:i]
}
n++
}
return s
}
// truncate 展示用截断,按 rune 计数2026-09-10 从字节切改为 rune 切,避免乱码)。
func truncate(s string, maxLen int) string {
if len(s) <= maxLen {
t := truncateRunes(s, maxLen)
if len(t) == len(s) {
return s
}
return s[:maxLen] + "..."
}
func minz(a, b int) int {
if a < b {
return a
}
return b
return t + "..."
}
func floatVal(v interface{}) float64 {

View File

@ -0,0 +1,182 @@
// 织忆 MemoryWeave — Obsidian 同步文件名/slug 单元测试
// 2026-09-10 新增kanban t_5f9c56ef锁死「中文标题不许变下划线」这个回归。
package routes
import (
"os"
"path/filepath"
"strings"
"testing"
"unicode/utf8"
)
func TestSanitizeFilename_KeepsChinese(t *testing.T) {
cases := []struct {
in string
want string
}{
{"修复了respond重复声明问题", "修复了respond重复声明问题"},
{"110GB 磁盘占用排查", "110GB 磁盘占用排查"},
{"E5.3 Web UI 单文件React+静态文件中间件", "E5.3 Web UI 单文件React_静态文件中间件"},
{"a/b\\c:d*e?f\"g<h>i|j", "a_b_c_d_e_f_g_h_i_j"},
{" 前后空格 ", "前后空格"},
{"...隐藏点开头...", "隐藏点开头"},
{"///", ""},
{"", ""},
{"测试::连续非法字符::折叠", "测试_连续非法字符_折叠"},
}
for _, c := range cases {
if got := sanitizeFilename(c.in); got != c.want {
t.Errorf("sanitizeFilename(%q) = %q, want %q", c.in, got, c.want)
}
}
}
// 回归锁旧实现2026-09-10 前)会把中文全部换成 '_',这条用例就是当时的现场。
func TestSanitizeFilename_RegressionNoUnderscoreSoup(t *testing.T) {
got := sanitizeFilename("____112GB_____")
if strings.Contains(got, "____") {
t.Fatalf("中文标题不该出现连续下划线乱码: %q", got)
}
old := sanitizeFilename("修复了respond重复声明问题")
if strings.Contains(old, "_") {
t.Fatalf("中文/字母不该被替换成下划线: %q", old)
}
}
func TestTruncateRunes_NoBrokenUTF8(t *testing.T) {
s := "修复了respond重复声明问题这是很长的中文内容用来测试截断"
got := truncateRunes(s, 7)
if got != "修复了resp" {
t.Fatalf("truncateRunes = %q", got)
}
if !utf8.ValidString(got) {
t.Fatal("截断结果不是合法 UTF-8多字节字符被切半")
}
if truncateRunes(s, 0) != "" || truncateRunes(s, -1) != "" {
t.Fatal("maxRunes<=0 应返回空串")
}
if truncateRunes("短", 10) != "短" {
t.Fatal("短串应原样返回")
}
// 关键回归:旧实现 content[:30] 是字节切中文会切出半个字符utf8.RuneError
if got := truncateRunes("中文标题测试内容很长很长很长很长很长很长", 8); !utf8.ValidString(got) {
t.Fatalf("rune 截断后仍不是合法 UTF-8: %q", got)
}
}
func TestTruncate_RuneCounted(t *testing.T) {
s := strings.Repeat("记", 40) // 120 字节;旧实现 s[:10] 只能得到 3 个半截汉字
got := truncate(s, 10)
want := strings.Repeat("记", 10) + "..."
if got != want {
t.Fatalf("truncate = %q, want %q", got, want)
}
if !utf8.ValidString(got) {
t.Fatal("truncate 结果不是合法 UTF-8")
}
if truncate("短内容", 60) != "短内容" {
t.Fatal("未超长应原样返回")
}
}
func TestShortHash_Deterministic(t *testing.T) {
a, b := shortHash("同一段内容"), shortHash("同一段内容")
if a != b {
t.Fatal("shortHash 必须稳定")
}
if len(a) != 8 {
t.Fatalf("shortHash 长度应为 8得到 %d", len(a))
}
if shortHash("内容甲") == shortHash("内容乙") {
t.Fatal("不同内容不该同哈希")
}
}
// 端到端:真的落盘,检查文件名可读、无乱码、同名不互相覆盖。
func TestPushToObsidian_ReadableFilenames(t *testing.T) {
dir := t.TempDir()
s := NewObsidianSyncer(dir, nil)
dupPrefix := "同一段开头的内容 abcdefghijklmnopqrstuvwxyz"
mems := []map[string]interface{}{
{"id": "mem_1", "category": "distilled", "quality_score": 0.9,
"content": "修复了 respond 重复声明问题"},
{"id": "mem_2", "category": "distilled", "quality_score": 0.8,
"content": "LanceDB 已设天花板,不会再膨胀到 35G"},
// 前 30 个 rune 完全相同 → 必须加哈希后缀而不是覆盖
{"id": "mem_3", "category": "distilled", "content": dupPrefix + " 变体一"},
{"id": "mem_4", "category": "distilled", "content": dupPrefix + " 变体二"},
}
if err := s.PushToObsidian(mems, "mirror"); err != nil {
t.Fatalf("PushToObsidian 失败: %v", err)
}
entries, err := os.ReadDir(filepath.Join(dir, "mirror"))
if err != nil {
t.Fatal(err)
}
if len(entries) != len(mems) {
t.Fatalf("应写入 %d 个文件(含撞名消歧),实际 %d 个:%v", len(mems), len(entries), names(entries))
}
t.Logf("实际落盘文件名: %v", names(entries))
for _, e := range entries {
n := e.Name()
if strings.Contains(n, "____") {
t.Errorf("文件名出现下划线乱码: %q", n)
}
if !strings.ContainsAny(n, "修复了重复声明问题磁盘排查天花板膨胀同一段") {
t.Errorf("文件名丢失中文标题: %q", n)
}
if !utf8.ValidString(n) {
t.Errorf("文件名不是合法 UTF-8: %q", n)
}
data, err := os.ReadFile(filepath.Join(dir, "mirror", n))
if err != nil {
t.Fatal(err)
}
if !utf8.Valid(data) {
t.Errorf("文件内容不是合法 UTF-8: %q", n)
}
if !strings.Contains(string(data), "由织忆 MemoryWeave 同步") {
t.Errorf("缺少同步 footer: %q", n)
}
}
// 撞名消歧两条同前缀记忆都要落盘4 个文件已证明没互相覆盖)
var suffixed int
for _, e := range entries {
if strings.Contains(e.Name(), dupPrefix[:12]) && strings.Contains(e.Name(), "-") {
suffixed++
}
}
if suffixed != 1 {
t.Errorf("应恰好 1 个文件带哈希后缀消歧,实际 %d%v", suffixed, names(entries))
}
}
func names(entries []os.DirEntry) []string {
out := make([]string, 0, len(entries))
for _, e := range entries {
out = append(out, e.Name())
}
return out
}
// ─── 2026-09-10 P0文件名不得带凭证 ───────────────────────
func TestSanitizeFilename_RedactsSecret(t *testing.T) {
const fakeKey = "tskey-auth-kTESTONLY0000000000000000000000000000000000000"
got := sanitizeFilename("authkey " + fakeKey)
if got == "" {
t.Fatal("脱敏后不应为空(占位符仍可读)")
}
if strings.Contains(got, "tskey-auth-kTESTONLY") {
t.Errorf("文件名仍含凭证: %q", got)
}
if strings.Contains(got, "/") || strings.Contains(got, string(os.PathSeparator)) {
t.Errorf("文件名仍含路径分隔符: %q", got)
}
}

View File

@ -128,7 +128,7 @@ func TestConflictAPI_Resolve_Validation(t *testing.T) {
// ─── 知识缺口路由 ────────────────────────────────────────
func TestGapAPI_List(t *testing.T) {
gd := selfoptimize.NewGapDetector()
gd := selfoptimize.NewGapDetector(nil, nil)
ai := NewGapAPI(gd)
req := httptest.NewRequest("GET", "/api/v1/gaps", nil)
@ -141,7 +141,7 @@ func TestGapAPI_List(t *testing.T) {
}
func TestGapAPI_Detect(t *testing.T) {
gd := selfoptimize.NewGapDetector()
gd := selfoptimize.NewGapDetector(nil, nil)
ai := NewGapAPI(gd)
body, _ := json.Marshal(map[string]string{"topic": "test_gap"})
@ -199,8 +199,8 @@ func TestTriggers_List(t *testing.T) {
}
var resp map[string]interface{}
json.NewDecoder(w.Body).Decode(&resp)
if resp["count"].(float64) != 4 {
t.Errorf("expected 4 triggers, got %v", resp["count"])
if resp["count"].(float64) != 8 {
t.Errorf("expected 8 triggers, got %v", resp["count"])
}
}
@ -306,18 +306,19 @@ func TestAuxFunctions(t *testing.T) {
}
}
// ─── SSE 管理器 ───────────────────────────────────────────
// ─── WebSocket 事件推送 ───────────────────────────────────
func TestSSEManager_Push(t *testing.T) {
// SSEBus.Push should not panic with no clients
SSEBus.Push(SSEMessage{Type: "test", Payload: "hello"})
}
func TestSSEPushHelpers(t *testing.T) {
// All push helpers should not panic
PushMemoryCommitted("agent", "ns", "mem1")
PushConflictDetected("Docker", "facts conflict")
PushGapFound("topic", "A")
func TestWSPushHelpers_NoPanic(t *testing.T) {
// 所有推送 helper 在无订阅者时都不该 panic
// 2026-09-10 修复:原用例引用的 SSEBus/SSEMessage/PushMemoryCommitted/PushGapFound
// 已随 SSE→WebSocket 重构删除,改用当前 APIPushGapDetected/PushConflict* 新签名)。
PushPrefetch("agent", []interface{}{})
PushGapDetected("agent", "topic", "knowledge", "补一下")
PushGapFilled("agent", 1)
PushMemoryUpdated("mem1", 2, "test")
PushConflictDetected("agent", "c1", "Docker", map[string]string{"a": "b"})
PushConflictResolved("agent", "c1", "keep_left")
PushConsolidationDone("done")
PushConflictResolved("c1", "keep_left")
PushQualityDrop("agent", "mem1", 0.2, "补证据")
PushDistillationComplete(1, 0)
}

View File

@ -22,6 +22,7 @@ import (
"github.com/xiaoxue/memoryweave/internal/governance"
"github.com/xiaoxue/memoryweave/internal/metrics"
"github.com/xiaoxue/memoryweave/internal/models"
"github.com/xiaoxue/memoryweave/internal/redact"
"github.com/xiaoxue/memoryweave/internal/selfoptimize"
"github.com/xiaoxue/memoryweave/internal/storage"
)
@ -51,6 +52,11 @@ func NewServer() http.Handler {
// 存储后端
ldb := initStorageBackend(os.Getenv("STORAGE_BACKEND"), emb)
// 召回元数据「累积延迟更新」缓冲2026-09-12 优化B
// 读路径不再逐条写 LanceDBMVCC 每写一行一个版本 → _versions 膨胀根因),
// 改为窗口内合并 + 单事务批量提交(每批 1 个版本)。
storage.InitRecallWriteBuffer(ldb)
// 启动时数据目录一致性检查(防止路径混乱导致读取废弃数据)
runStartupChecks(os.Getenv("STORAGE_BACKEND"))
@ -139,7 +145,21 @@ func NewServer() http.Handler {
})
// 回写蒸馏产物到记忆库LanceDB
// 2026-09-07 方案B 质量门槛: 碎片(<20字)/噪声句(状态汇报/桥接/reflection)不入库,
// 与已有记忆完全重复跳过(蒸馏直写不走 commit exact_dup, 需本层查重)。
for _, fact := range result.Facts {
// 2026-09-10 P0直写路径同样脱敏这里绕过 FilterQualityFacts
// 事故里泄漏进镜像的记忆正是从这条路径入库的)。
fact = redact.RedactSecrets(fact)
if !distill.IsQualityFact(fact) {
continue // 碎片/噪声过滤
}
// 查重: 同 namespace 已存在完全相同 content → 跳过
if qvec, eErr := emb.EncodeSingle(fact); eErr == nil {
if sim, _ := ldb.Search("memories", qvec, 1, input.Namespace); len(sim) > 0 && sim[0].Content == fact {
continue
}
}
mem := models.MemoryRecord{
ID: fmt.Sprintf("mem_%d", time.Now().UnixNano()),
AgentID: input.AgentID,
@ -153,6 +173,7 @@ func NewServer() http.Handler {
UpdatedAt: time.Now(),
DerivedFrom: input.EpisodeID,
Freshness: "fresh",
Source: "llm_distill", // 2026-09-07 方案C provenance: LLM蒸馏推断来源
}
if err := ldb.InsertMemory(mem); err != nil {
log.Printf("[distill] commit fact failed: %v", err)
@ -572,6 +593,51 @@ func NewServer() http.Handler {
})
})
// 图谱多 agent 测试污染清理namespace + 名称子串圈定,不碰全表编码规则)
// 2026-09-08 卫安/A03/A04 371 节点openclaw-main/a06-main/hermes-main 中测试 agent 产生的概念/事实节点
// GET/POST /api/v1/graph/cleanup/scoped?dry_run=true&namespace=a,b,c&name=a03,a04,卫安
mux.HandleFunc("/api/v1/graph/cleanup/scoped", func(w http.ResponseWriter, r *http.Request) {
if r.Method != "POST" && r.Method != "GET" {
http.Error(w, "GET or POST only", http.StatusMethodNotAllowed)
return
}
dryRun := r.URL.Query().Get("dry_run") != "false" // 默认为 dryRun
nsParam := r.URL.Query().Get("namespace")
nameParam := r.URL.Query().Get("name")
if nsParam == "" || nameParam == "" {
http.Error(w, "namespace and name (comma separated) are required", http.StatusBadRequest)
return
}
namespaces := strings.Split(nsParam, ",")
nameContains := strings.Split(nameParam, ",")
type scopedCleaner interface {
CleanupScopedNodes(dryRun bool, namespaces, nameContains []string) (int, []string, error)
}
cleaner, ok := graphStore.(scopedCleaner)
if !ok {
http.Error(w, "scoped cleanup not supported by current graph backend", http.StatusNotImplemented)
return
}
count, ids, err := cleaner.CleanupScopedNodes(dryRun, namespaces, nameContains)
if err != nil {
http.Error(w, "cleanup failed: "+err.Error(), http.StatusInternalServerError)
return
}
// 实际执行后失效缓存dry_run=true 时不执行,无需失效)
if !dryRun && cachedGraphStoreRef != nil && count > 0 {
cachedGraphStoreRef.InvalidateAll()
}
respondJSON(w, 200, map[string]interface{}{
"dry_run": dryRun,
"removed": count,
"node_ids": ids,
"namespaces": namespaces,
"name_contains": nameContains,
"message": fmt.Sprintf("Found %d scoped noise nodes", count),
})
})
// 静态文件服务(知识图谱可视化 HTML
staticDir := os.Getenv("STATIC_DIR")
if staticDir == "" {
@ -1245,9 +1311,36 @@ func NewServer() http.Handler {
graphStore.Prune(0.15)
case "decay":
// 扫描所有记忆,按衰减率计算 freshness
// 2026-09-06 P1 fix: 每 tick 限量评估maxDecayBatch优先处理
// 最久未访问的last_recalled_at 最早 → 最该遗忘),防全表单次拉取
// 内存翻倍 + 长时间阻塞触发器循环。
memories, e := ldb.GetCandidatesForForgetting()
decayScanned := 0
decayForgotten := 0
if e == nil {
const maxDecayBatch = 2000
if len(memories) > maxDecayBatch {
sort.SliceStable(memories, func(a, b int) bool {
return memoryTimeVal(memories[a], "last_recalled_at", "created_at") <
memoryTimeVal(memories[b], "last_recalled_at", "created_at")
})
memories = memories[:maxDecayBatch]
}
for _, m := range memories {
decayScanned++
// 🔒 2026-09-07 修复(误删事故): episodes(原始对话)不参与遗忘——
// 它们是蒸馏原料/审计历史, 不是可遗忘的记忆。曾因未过滤 category
// 导致 67 条 8月对话被 auto_forget 清掉(已恢复)。仅 distilled/general 等记忆可遗忘。
if cat, ok := m["category"].(string); ok && cat == "episodes" {
continue
}
// 🔒 长内容保护: distilled >200 字(有实质信息)不参与 auto 遗忘
// (曾误删 405 字 CNB 修复经验; 长记忆应由蒸馏/整合管理, 非时间遗忘)
if cat, _ := m["category"].(string); cat != "episodes" {
if cs, ok := m["content"].(string); ok && len([]rune(cs)) > 200 {
continue
}
}
// 解析 last_recalled_at
lastAccessed := time.Now().Add(-30 * 24 * time.Hour) // 默认30天前
if t, ok := m["last_recalled_at"].(string); ok && t != "" {
@ -1255,6 +1348,20 @@ func NewServer() http.Handler {
lastAccessed = parsed
}
}
// 2026-09-07 方案A 碎片快速道: <30 字且 >20 天未访问 → 直接遗忘
// (绕过 ShouldForget 的 recallCount 保命——碎片 recall_count 可能虚高)。
// 碎片无记忆价值(审计: 30.9% <20字 + 43% <50字 = 蒸馏无门槛产物)。
contentStr := ""
if cs, ok := m["content"].(string); ok {
contentStr = cs
}
if len([]rune(contentStr)) < 30 && time.Since(lastAccessed).Hours()/24 > 20 {
if id, ok := m["id"].(string); ok {
_ = ldb.SoftDelete(id, "auto_forget_fragment")
decayForgotten++
}
continue
}
recallCount := 0
if rc, ok := m["recall_count"].(int); ok {
recallCount = rc
@ -1277,9 +1384,14 @@ func NewServer() http.Handler {
forgetter.ScanAndForget(id)
// 实际执行软删除(持久化到 LanceDB
_ = ldb.SoftDelete(id, "auto_forget")
decayForgotten++
}
}
}
}
}
// AC-1 验收日志P1 修复 type 断言后 last_recalled_at 可解析batch<=2000
if decayScanned > 0 {
log.Printf("[decay] scanned=%d forgotten=%d (batch<=2000, 最久未访问优先)", decayScanned, decayForgotten)
}
case "gap_scan":
// 检查已有缺口:过期 7 天的自动关闭
@ -1422,6 +1534,40 @@ func AgentTypeDecayOrDefault(agentType string) float64 {
return 0.015
}
// memoryTimeVal 取记忆时间戳用于排序(主 key 优先,空则回退副 key均空返回 0
// 2026-09-06 P1: decay 分批按"最久未访问优先"排序需要;兼容 RFC3339 string / time.Time / int。
func memoryTimeVal(m map[string]interface{}, primary, fallback string) int64 {
for _, key := range []string{primary, fallback} {
v, ok := m[key]
if !ok || v == nil {
continue
}
switch val := v.(type) {
case string:
if val == "" {
continue
}
if t, err := time.Parse(time.RFC3339, val); err == nil {
return t.Unix()
}
if n, err := strconv.ParseInt(val, 10, 64); err == nil {
return n
}
case time.Time:
if !val.IsZero() {
return val.Unix()
}
case float64:
return int64(val)
case int64:
return val
case int:
return int64(val)
}
}
return 0
}
// runStartupChecks 启动时数据目录一致性检查
// 检测废弃路径(如 /home/muc/data并警告防止数据源混乱
func runStartupChecks(backend string) {

View File

@ -273,6 +273,16 @@ func (e *Engine) distillOne(input DistillInput) DistillResult {
facts = heuristicFacts
}
// 2026-09-07 质量门槛(t_cc07ef4b): LLM 提炼 fact 入库前过滤
// <20字/纯疑问句/状态汇报型 拒绝; 20-50字需含实体或动词短语
// 在源头过滤, 使图谱/AAAK/记忆库一律只看到质量事实
if before := len(facts); before > 0 {
facts = FilterQualityFacts(facts)
if dropped := before - len(facts); dropped > 0 {
log.Printf("[distill] quality gate dropped %d/%d facts for %s", dropped, before, input.EpisodeID)
}
}
// P4: 生成 AAAK 压缩索引(每条事实的紧凑摘要,供召回快速定位)
index := buildAAAKIndex(facts, entities, overall)
if len(index) > 0 {
@ -612,6 +622,8 @@ func fallbackSingle(input DistillInput) DistillResult {
if len(input.Content) > 20 {
facts = append(facts, truncate(input.Content, 100))
}
// 质量门槛同样作用于 fallback 路径(t_cc07ef4b)
facts = FilterQualityFacts(facts)
return DistillResult{Facts: facts}
}

View File

@ -0,0 +1,247 @@
// 织忆 MemoryWeave — 蒸馏质量门槛方案B 严格版, kanban t_cc07ef4b
//
// 2026-09-07 t_cc07ef4b 继承 t_9316bf56 AC-1: LLM 提炼 fact 入库前过滤
//
// 规则(按卡面 spec, 覆盖原 946f0c9 宽松版):
// - <20字 → 拒绝 (碎片; 与质量看门狗 FRAG_LEN=20 对齐, 无用户信号豁免)
// - 纯疑问句 → 拒绝 (OpenQuestions 合流 + LLM 直接输出问句)
// - 状态汇报型 → 拒绝 (健康检查/重启/验证/桥接/daemon reflection 类进程噪声)
// - 20-50字 → 需含实体或动词短语才入库
// - >50字 → 通过 (除非命中上面 纯疑问/状态汇报 规则)
//
// 与 LightMem"保留小细节"哲学差异: 原版允许 8-20 字带用户信号(如"用户喜欢喝咖啡")
//
// 保留——但看门狗把 <20 字全部计为碎片, AC-2(新增碎片率<10%)要求一律拒绝,
// 故此处 <20 字无豁免。短对话原始内容仍完整保留在 episodes 层, 不丢失信息。
package distill
import (
"strings"
"unicode"
"unicode/utf8"
"github.com/xiaoxue/memoryweave/internal/redact"
)
// IsQualityFact 判断蒸馏事实是否有入库价值(严格质量门槛)
//
// 2026-09-10 P0 追加:先做密钥脱敏再判定。含凭证的事实不再把原文带进图谱
// /镜像(事故见 internal/redact 包注释)。脱敏后若只剩占位符(<20 字)自然被拒。
func IsQualityFact(fact string) bool {
s := strings.TrimSpace(redact.RedactSecrets(fact))
if s == "" {
return false
}
// 纯疑问句 → 拒绝(任何长度)
if isPureQuestion(s) {
return false
}
// 状态汇报型/进程噪声 → 拒绝(任何长度)
if isStatusReportLike(s) {
return false
}
runeLen := utf8.RuneCountInString(s)
// <20字 → 拒绝(碎片)
if runeLen < 20 {
return false
}
// 20-50字 → 需含实体或动词短语
if runeLen <= 50 {
if !(hasEntity(s) || hasVerbPhrase(s)) {
return false
}
}
return true
}
// FilterQualityFacts 过滤事实列表, 返回通过质量门槛的子集(引擎入队后统一调用)
//
// 2026-09-10 P0返回值已做密钥脱敏调用方沿用返回值即可勿再用原始 slice
func FilterQualityFacts(facts []string) []string {
out := make([]string, 0, len(facts))
for _, f := range facts {
f = redact.RedactSecrets(f)
if IsQualityFact(f) {
out = append(out, f)
}
}
return out
}
// ─── 纯疑问句判定 ──────────────────────────────────────────
var questionSuffixes = []string{"", "?", "吗?", "呢?", "么?", "嘛?", "吗?", "呢?"}
var questionPrefixes = []string{
"为什么", "怎么", "如何", "是否", "能不能", "可不可以", "要不要",
"有没有", "什么", "哪些", "哪个", "哪里", "几时", "多少", "多久",
"谁", "何时", "为何", "请问", "能否", "咋", "啥",
}
// isPureQuestion 判断整句是否为疑问句(不是转述的疑问)
func isPureQuestion(s string) bool {
t := strings.TrimSpace(s)
if t == "" {
return false
}
// 以问号结尾 → 强信号
for _, q := range []string{"", "?"} {
if strings.HasSuffix(t, q) {
// 排除 "用户问是否升级?" 这类转述?——结尾问号仍视为疑问句本体,
// 蒸馏 facts 里不应出现任何问句; 转述型应改写为陈述("用户询问了...")
return true
}
}
// 疑问前缀 + 以 吗/呢/么 等结尾(无问号的口语问句)
for _, p := range questionPrefixes {
if strings.HasPrefix(t, p) {
for _, suf := range []string{"吗", "呢", "么", "嘛", "啊"} {
if strings.HasSuffix(t, suf) {
return true
}
}
// 疑问前缀且整句较短(<30)且无陈述主语 → 判为问句
if utf8.RuneCountInString(t) <= 30 && !hasStatementSubject(t) {
return true
}
}
}
return false
}
// hasStatementSubject 粗略判断是否带陈述主语(用户/小唯/牧尘/他/她/系统等)
func hasStatementSubject(s string) bool {
subjects := []string{"用户", "小唯", "牧尘", "他", "她", "它", "我", "我们", "系统", "服务器", "对方", "同事", "老板"}
for _, sub := range subjects {
if strings.HasPrefix(s, sub) {
return true
}
}
return false
}
// ─── 状态汇报型 / 进程噪声判定 ─────────────────────────────
// statusExactPhrases 无主状态句(完整或前缀命中即拒 —— 进程/守护噪声不是用户事实)
var statusExactPhrases = []string{
"健康检查完成", "健康检查通过", "系统健康检查", "所有进程运行中",
"当前系统状态完全正常", "状态同步", "正在评估上次决策", "上次修复了",
"心跳正常", "运行状态正常", "一切正常", "无异常", "检查完毕",
"自检完成", "任务已完成", "处理完成", "收到指令", "开始执行",
"执行完毕", "正在执行任务", "测试通过", "验证通过",
"小唯a06主profile", "小唯a06使用的模型", "小唯a06的织忆数据",
"小唯a06的cron", "小唯a06的daemon", "小唯a06的",
}
// statusNoiseMarkers 强噪声子串(含"健康检查"级信号, 无主句才拒)
var statusNoiseMarkers = []string{
"健康检查", "桥接", "reflection", "reflecting", "heartbeat",
"cron运行", "cron任务", "状态同步",
}
// completionMarkers 完成/重启/恢复类动词短语(无主句时才视为状态汇报)
var completionMarkers = []string{
"重启完成", "重启成功", "已重启", "验证通过", "验证完成", "测试通过",
"测试完成", "恢复完成", "已恢复", "已启动", "已停止", "已完成清理",
"清理完成", "备份完成", "同步完成", "重启系统", "执行了重启",
}
// isStatusReportLike 判断是否为状态汇报/进程噪声句。
// 判定原则: 只有"无主语"的状态汇报才拒 —— 带用户/系统主体的真实事实(如
// "用户重启了服务器")不在此列, 交给 <20字/20-50字 规则约束。
func isStatusReportLike(s string) bool {
t := strings.TrimSpace(s)
lower := strings.ToLower(t)
// 1) 完整/前缀命中已知无主状态句 → 拒
for _, p := range statusExactPhrases {
if strings.HasPrefix(lower, strings.ToLower(p)) || lower == strings.ToLower(p) {
return true
}
}
// 2) 无主句 + 强噪声子串 → 拒
if !hasStatementSubject(t) {
for _, m := range statusNoiseMarkers {
if strings.Contains(lower, strings.ToLower(m)) {
return true
}
}
for _, c := range completionMarkers {
if strings.Contains(lower, strings.ToLower(c)) {
return true
}
}
}
return false
}
// ─── 实体 / 动词短语 判定20-50字 门槛)───────────────────
// strongEntities 常见强实体词(技术名/系统名/专有名词; 命中 = 有实体)
var strongEntities = []string{
"牧尘", "小唯", "织忆", "KOCR", "Hermes", "hermes", "NewAPI", "newapi",
"飞书", "服务器", "笔记本", "zhiyid", "daemon", "llama", "qwen", "glm",
"deepseek", "agnes", "Gitea", "gitea", "Redis", "redis", "LanceDB",
"ComfyUI", "Windows", "Linux", "Deepin", "Vulkan", "CUDA", "ONNX",
"Python", "python", "Go语言", "golang", "Rust", "MySQL", "SQLite",
"ffmpeg", "docker", "nginx", "frpc", "frps",
}
// strongVerbs 常见动词短语(动作/状态变化词; 命中 = 有动词)
var strongVerbs = []string{
"安装", "部署", "修复", "升级", "使用", "需要", "决定", "选择", "计划",
"希望", "认为", "购买", "买了", "完成", "重启", "切换", "迁移", "创建",
"删除", "下载", "上传", "配置", "编写", "开发", "测试", "验证", "检查",
"发现", "解决", "提交", "推送", "更新", "设置", "启动", "停止", "运行",
"增加", "减少", "更换", "替换", "调整", "开始", "结束", "参加", "申请",
"同意", "拒绝", "喜欢", "讨厌", "学习", "研究", "阅读", "访问", "连接",
"发送", "接收", "保存", "修改", "整理", "清理", "备份", "恢复", "管理",
"维护", "分析", "讨论", "提出", "回答", "询问", "告知", "邀请", "约定",
"到达", "离开", "回来", "前往", "居住", "工作", "出生", "结婚", "毕业",
"入职", "离职", "订购", "预约", "取消", "返回", "打开", "关闭", "进入",
"退出", "登录", "注册", "写", "读", "做", "换", "买", "改", "拆", "装",
"尝试", "测试过", "调研", "评估", "对比", "选择用", "改用", "转用",
"处理", "解决掉", "搞定", "看过", "打开过", "拆过",
}
// hasEntity 是否含实体(专有名词/技术名/大写英文词/数字)
func hasEntity(s string) bool {
// 强实体词
for _, e := range strongEntities {
if strings.Contains(s, e) {
return true
}
}
// 大写英文单词Hermes/ComfyUI/NewAPI 等)
hasUpperWord := false
words := strings.Fields(s)
for _, w := range words {
runes := []rune(w)
if len(runes) >= 2 && unicode.IsUpper(runes[0]) {
// 排除句首大写的中文拼音误判?中文在 Go range 下不是 IsUpper,
// 只捕获 ASCII 大写开头 → 安全
hasUpperWord = true
break
}
}
if hasUpperWord {
return true
}
// 含数字(端口/型号/年份/数量)
for _, r := range s {
if r >= '0' && r <= '9' {
return true
}
}
return false
}
// hasVerbPhrase 是否含动词短语
func hasVerbPhrase(s string) bool {
for _, v := range strongVerbs {
if strings.Contains(s, v) {
return true
}
}
return false
}

View File

@ -0,0 +1,76 @@
package distill
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
// AC-1 dev 验证 (kanban t_cc07ef4b): 构造含短对话的 episodes → 蒸馏后碎片不入库。
// 用 mock LLM 模拟真实蒸馏调用返回混合 facts(碎片+疑问句+状态汇报+合格事实),
// 断言 distillOne 返回的 facts 已按质量门槛过滤, 不会进入入库回调。
func TestDistillOne_QualityGateBlocksFragments(t *testing.T) {
// mock LLM 返回: 短对话蒸馏产物——大部分是 <20字碎片/疑问句/状态汇报
mockResp := LLMResponse{
Facts: []string{
"重启系统", // <20 碎片 → 拒
"健康检查完成", // 状态汇报 → 拒
"验证通过", // 状态汇报 → 拒
"为什么系统会重启?", // 纯疑问句 → 拒
"用户喜欢喝咖啡", // <20 碎片 → 拒
"用户对织忆系统的蒸馏质量提出了改进要求并希望尽快处理", // 24字 含实体 → 通过
},
Entities: []string{"织忆"},
IS: 0.9, SU: 0.8, PA: 0.8, VD: 0.9, RU: 0.8,
}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
payload := map[string]interface{}{
"choices": []map[string]interface{}{
{"message": map[string]interface{}{"content": mustJSON(t, mockResp)}},
},
}
_ = json.NewEncoder(w).Encode(payload)
}))
defer srv.Close()
e := NewEngine(srv.URL, "mock-model", "test-key")
// 构造短对话 episode: 大量碎片式短句 (类似 openclaw bridge/闲聊/状态行)
episode := DistillInput{
EpisodeID: "ep_quality_gate_test",
Content: "[2026-09-07] user: 重启系统\n[2026-09-07] assistant: 好的\n[2026-09-07] user: 健康检查完成\n[2026-09-07] user: 为什么系统会重启?\n[2026-09-07] user: 用户喜欢喝咖啡\n[2026-09-07] user: 用户对织忆系统的蒸馏质量提出了改进要求并希望尽快处理",
Category: CatSystemFact,
Namespace: "dev-quality-gate",
AgentID: "a06",
}
res := e.distillOne(episode)
t.Logf("distillOne facts kept = %d: %v", len(res.Facts), res.Facts)
// 断言: 只有 1 条合格事实通过; 碎片/疑问/状态汇报全部被拒
if len(res.Facts) != 1 {
t.Fatalf("expected exactly 1 quality fact to survive gate, got %d: %v", len(res.Facts), res.Facts)
}
if res.Facts[0] != "用户对织忆系统的蒸馏质量提出了改进要求并希望尽快处理" {
t.Errorf("unexpected surviving fact: %q", res.Facts[0])
}
}
// TestFilterQualityFacts_AllNoise 全噪声 episode: 蒸馏产物全碎片 → 0 条入库
func TestFilterQualityFacts_AllNoise(t *testing.T) {
in := []string{"重启系统", "验证通过", "健康检查完成", "为什么?", "好的"}
got := FilterQualityFacts(in)
if len(got) != 0 {
t.Fatalf("expected 0 facts for all-noise input, got %v", got)
}
}
func mustJSON(t *testing.T, v interface{}) string {
t.Helper()
b, err := json.Marshal(v)
if err != nil {
t.Fatalf("marshal: %v", err)
}
return string(b)
}

View File

@ -0,0 +1,169 @@
package distill
import (
"strings"
"testing"
)
// 质量门槛单测 (kanban t_cc07ef4b 继承 t_9316bf56 AC-1)
// 规则:
// - <20字 → 拒绝
// - 纯疑问句 → 拒绝
// - 状态汇报型(无主句) → 拒绝
// - 20-50字 → 需含实体或动词短语才入库
// - >50字 → 通过(除非疑问/状态汇报)
func TestIsQualityFact_ShortFragmentsRejected(t *testing.T) {
cases := []struct {
name string
fact string
want bool
}{
// AC-1 审计样例: <20字碎片 一律拒绝(无用户信号豁免)
{"重启系统", "重启系统", false},
{"健康检查完成", "健康检查完成", false},
{"所有进程运行中", "所有进程运行中", false},
{"验证通过", "验证通过", false},
{"用户喜欢喝咖啡(短)", "用户喜欢喝咖啡", false},
{"用户重启了系统", "用户重启了系统", false},
{"19字边界", "用户今天完成了一次系统检查工作", false}, // 14字
{"空串", "", false},
{"纯空白", " ", false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := IsQualityFact(tc.fact); got != tc.want {
t.Errorf("IsQualityFact(%q) = %v, want %v", tc.fact, got, tc.want)
}
})
}
}
func TestIsQualityFact_PureQuestionRejected(t *testing.T) {
cases := []struct {
name string
fact string
want bool
}{
{"带问号", "为什么系统会重启?", false},
{"英文问号", "how to install hermes?", false},
{"疑问词+吗", "用户要不要升级系统吗", false},
{"怎么开头", "怎么解决这个报错问题呢", false},
{"是否开头无标点", "是否应该把模型切换到本地", false},
{"转述疑问句(带主语,陈述)", "用户询问了系统是否可以升级到最新稳定版本", true}, // 20字 主语转述非纯问句
{"正常陈述带吗字(句中)", "用户说这个报错不用再管它了因为已经处理过了", true}, // 21字, 有主语+动词
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := IsQualityFact(tc.fact); got != tc.want {
t.Errorf("IsQualityFact(%q) = %v, want %v", tc.fact, got, tc.want)
}
})
}
}
func TestIsQualityFact_StatusReportRejected(t *testing.T) {
cases := []struct {
name string
fact string
want bool
}{
{"无主健康检查", "健康检查完成,所有服务运行正常,无异常", false}, // 无主句+健康检查
{"前缀状态句", "当前系统状态完全正常,无需处理", false},
{"桥接噪声", "小唯a06与all桥接状态同步完成", false},
{"重启完成无主", "重启完成,服务已恢复运行", false},
{"带主语重启(真事实)", "用户昨天重启了家里的服务器并恢复了所有服务", true}, // 21字 有主语+实体服务器
{"带主语升级", "小唯昨天将织忆系统升级到了新版本并验证通过", true}, // 有主语
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := IsQualityFact(tc.fact); got != tc.want {
t.Errorf("IsQualityFact(%q) = %v, want %v", tc.fact, got, tc.want)
}
})
}
}
func TestIsQualityFact_MidLengthNeedsEntityOrVerb(t *testing.T) {
cases := []struct {
name string
fact string
want bool
}{
// 20-50字 无实体无动词 → 拒绝
{"纯形容词堆叠", "很好很好很好很好很好很好很好很好很好", false}, // 20字 无实体无动词
{"无实义名词串", "关于这个系统的一些非常详细的说明文档内容汇总介绍", false}, // 24字 无实体无动词
// 20-50字 含实体 → 通过
{"含织忆实体", "用户对织忆系统的蒸馏质量提出了改进要求并希望尽快处理", true},
{"含hermes实体", "用户说hermes升级后速度明显变快了体验很好", true},
{"含数字实体", "用户提到显卡是rtx3050只有4gb显存不够用", true},
// 20-50字 含动词 → 通过
{"含动词", "用户打算下个月把家里的网络设备全部更换一遍", true},
{"含完成动词", "用户今天完成了对全部旧脚本的清理和归档工作", true},
// >50字 → 通过
{"超50字正常陈述", "用户详细讲述了昨天在整理旧项目时发现的问题以及处理思路并且记录了完整的解决方案方便以后遇到类似情况时参考", true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := IsQualityFact(tc.fact); got != tc.want {
t.Errorf("IsQualityFact(%q) = %v, want %v", tc.fact, got, tc.want)
}
})
}
}
func TestFilterQualityFacts(t *testing.T) {
in := []string{
"重启系统", // <20 拒
"为什么系统会重启?", // 疑问 拒
"健康检查完成", // 状态 拒
"用户对织忆系统提出了新的功能需求希望尽快实现", // 22字 实体 通过
}
got := FilterQualityFacts(in)
if len(got) != 1 {
t.Fatalf("FilterQualityFacts(%v) len = %d, want 1 (got %v)", in, len(got), got)
}
if got[0] != in[3] {
t.Errorf("FilterQualityFacts kept %q, want %q", got[0], in[3])
}
}
// ─── 2026-09-10 P0凭证不得原样进入记忆/图谱 ───────────────
const fakeSecretFact = "tskey-auth-kTESTONLY0000000000000000000000000000000000000000"
func TestFilterQualityFacts_RedactsSecrets(t *testing.T) {
facts := []string{
"tailscale 预授权 key " + fakeSecretFact + " 已写进启动脚本用于远程接入",
"牧尘偏好结论先行,不喜欢废话科普与背景铺垫",
}
out := FilterQualityFacts(facts)
if len(out) != 2 {
t.Fatalf("期望 2 条通过,得到 %d: %v", len(out), out)
}
for _, f := range out {
if strings.Contains(f, fakeSecretFact) {
t.Errorf("输出仍含完整密钥: %q", f)
}
if strings.Contains(f, "tskey-auth-kTESTONLY") {
t.Errorf("输出仍含凭证前缀: %q", f)
}
}
if !strings.Contains(out[0], "<REDACTED-secret>") {
t.Errorf("期望占位符,得到 %q", out[0])
}
}
func TestIsQualityFact_SecretOnlyRejected(t *testing.T) {
if IsQualityFact(fakeSecretFact) {
t.Error("纯凭证事实脱敏后不足 20 字,应被拒绝")
}
}
func TestFilterQualityFacts_KeepsURLUntouched(t *testing.T) {
in := "参考 https://yoheinakajima.com/task-driven-autonomous-agents 项目的 Agent 编排设计思路"
out := FilterQualityFacts([]string{in})
if len(out) != 1 || out[0] != in {
t.Errorf("URL 文本被误改: %v", out)
}
}

View File

@ -110,7 +110,7 @@ func BenchmarkGraph_Navigate_5Hops(b *testing.B) {
g := buildScaleGraph(200, 3)
b.ResetTimer()
for i := 0; i < b.N; i++ {
g.Navigate(fmt.Sprintf("n-%d", i%200), 5, "shared")
g.Navigate(fmt.Sprintf("n-%d", i%200), 5, "shared", nil)
}
}
@ -118,7 +118,7 @@ func BenchmarkGraph_Navigate_Deep(b *testing.B) {
g := buildScaleGraph(500, 2)
b.ResetTimer()
for i := 0; i < b.N; i++ {
g.Navigate("n-0", 10, "shared")
g.Navigate("n-0", 10, "shared", nil)
}
}
@ -126,7 +126,7 @@ func BenchmarkGraph_Navigate_LargeScale(b *testing.B) {
g := buildScaleGraph(2000, 3)
b.ResetTimer()
for i := 0; i < b.N; i++ {
g.Navigate(fmt.Sprintf("n-%d", i%2000), 3, "shared")
g.Navigate(fmt.Sprintf("n-%d", i%2000), 3, "shared", nil)
}
}

View File

@ -220,6 +220,8 @@ type Forgetter struct {
}
func NewForgetter() *Forgetter {
// 2026-09-07 方案A: 0.015→0.03 曾致误删(36天库龄下 0-recall 长记忆全被清, 405字CNB经验被删),
// 回滚 0.015。碎片清除由碎片快速道(server.go <30字&>20天)负责, 普通记忆 53 天老化合理。
return &Forgetter{agentType: "default", decayRate: 0.015}
}
@ -259,7 +261,10 @@ func (f *Forgetter) ShouldForget(lastAccessed time.Time, recallCount int, tier s
score = 0.1
}
// recallCount > 0 减缓衰减
score += float64(recallCount) * 0.05
// 2026-09-07 方案A: 0.05→0.005。原 0.05/次 + 无 cap → recall_count 457-540 时 +25 分,
// 永不遗忘 (日志大量 recall_count 500+ = 每次搜索命中都 ++, 虚高保命)。
// 现 0.005/次, cap 30 次后贡献 ≤0.15 分 ≈ 5 天保护, 合理。
score += float64(recallCount) * 0.005
// 图谱节点度 > 5 时,每超过 1 度 + 0.03 保留分E4.3: 图谱推理参与遗忘决策)
if len(graphDegree) > 0 && graphDegree[0] > 5 {
score += float64(graphDegree[0]-5) * 0.03

View File

@ -135,7 +135,7 @@ func TestInMemoryGraph_Navigate(t *testing.T) {
g.AddEdge("e1", "n1", "n2", "used_by", "shared", 1.0)
g.AddEdge("e2", "n2", "n3", "connects_to", "shared", 0.8)
paths, err := g.Navigate("n1", 2, "shared")
paths, err := g.Navigate("n1", 2, "shared", nil)
if err != nil {
t.Fatalf("navigate failed: %v", err)
}
@ -170,12 +170,12 @@ func TestInMemoryGraph_NamespaceIsolation(t *testing.T) {
g.AddNode("n2", "B", "type", "hermes")
g.AddEdge("e1", "n1", "n2", "r", "hermes", 1.0)
paths, _ := g.Navigate("n1", 2, "shared")
paths, _ := g.Navigate("n1", 2, "shared", nil)
if len(paths) > 0 {
t.Error("shared namespace should NOT see hermes-only edges")
}
paths2, _ := g.Navigate("n1", 2, "hermes")
paths2, _ := g.Navigate("n1", 2, "hermes", nil)
if len(paths2) == 0 {
t.Error("hermes namespace should see its edges")
}
@ -203,7 +203,7 @@ func Test_isContradiction(t *testing.T) {
{"config updated", "config not updated", true},
}
for _, tc := range tests {
got := isContradiction(tc.a, tc.b)
got := IsContradiction(tc.a, tc.b)
if got != tc.expect {
t.Errorf("isContradiction(%q, %q) = %v, want %v", tc.a, tc.b, got, tc.expect)
}
@ -231,6 +231,6 @@ func BenchmarkGraphNavigate(b *testing.B) {
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
g.Navigate("na", 2, "s")
g.Navigate("na", 2, "s", nil)
}
}

View File

@ -6,6 +6,8 @@ import (
"strings"
"time"
"unicode"
"github.com/xiaoxue/memoryweave/internal/redact"
)
// AutoGraphUpdater 自动维护知识图谱
@ -19,23 +21,34 @@ func NewAutoGraphUpdater(g GraphStore) *AutoGraphUpdater {
// UpdateFromDistill 从蒸馏产物自动更新图谱§3.4
func (agu *AutoGraphUpdater) UpdateFromDistill(distilled *DistillInput) {
// 1. 提取实体并创建节点ID 和 name 都清洗)
// 0. 密钥过滤2026-09-10 P0见 internal/redact含凭证形状的实体不进图
// 否则实体名会被 ObsidianSync 当文件名,把密钥落成可读镜像文件
// (实例:小唯/07-Wiki/织忆/未找到命令.md 的文件名与正文含 tskey-auth-…)。
validEnts := make([]string, 0, len(distilled.Entities))
for _, entity := range distilled.Entities {
if redact.ContainsSecret(entity) || cleanEntityName(entity) == "" {
continue
}
validEnts = append(validEnts, entity)
}
// 1. 提取实体并创建节点ID 和 name 都清洗)
for _, entity := range validEnts {
nodeID := entityID(entity)
cleanName := cleanEntityName(entity)
agu.graph.AddNode(nodeID, cleanName, detectNodeType(entity), distilled.Namespace)
}
// 2. 创建实体间关系 + CO_OCCURS 共访边
entityCount := len(distilled.Entities)
entityCount := len(validEnts)
for i := 0; i < entityCount; i++ {
for j := i + 1; j < entityCount; j++ {
eidI := entityID(distilled.Entities[i])
eidJ := entityID(distilled.Entities[j])
eidI := entityID(validEnts[i])
eidJ := entityID(validEnts[j])
// 2a. 语义关系边inferRelation
edgeID := fmt.Sprintf("e_%s_%s_%d", eidI, eidJ, time.Now().UnixNano())
relation := inferRelation(distilled.Entities[i], distilled.Entities[j], distilled.Content)
relation := inferRelation(validEnts[i], validEnts[j], distilled.Content)
agu.graph.AddEdge(edgeID, eidI, eidJ,
relation, distilled.Namespace, 0.5)
@ -69,13 +82,16 @@ func (agu *AutoGraphUpdater) UpdateFromDistill(distilled *DistillInput) {
// 4. DERIVED_FROM 边:蒸馏产物 → 原始 episode
if distilled.EpisodeID != "" {
for _, entity := range distilled.Entities {
for _, entity := range validEnts {
eid := entityID(entity)
derivedEdgeID := fmt.Sprintf("df_%s_%s_%d", eid, distilled.EpisodeID, time.Now().UnixNano())
agu.graph.AddEdge(derivedEdgeID, eid, distilled.EpisodeID,
"DERIVED_FROM", distilled.Namespace, 0.9)
}
for _, fact := range distilled.Facts {
if redact.ContainsSecret(fact) {
continue // 含凭证的事实不建节点2026-09-10 P0
}
factNodeID := "f_" + strings.ReplaceAll(entityID(fact), "n_", "")
derivedEdgeID := fmt.Sprintf("df_%s_%s_%d", factNodeID, distilled.EpisodeID, time.Now().UnixNano())
agu.graph.AddEdge(derivedEdgeID, factNodeID, distilled.EpisodeID,
@ -146,6 +162,10 @@ func cleanEntityName(raw string) string {
if clean == "" {
clean = "unknown"
}
// 含凭证形状 → 返回空串,调用方 skip2026-09-10 P0
if redact.ContainsSecret(clean) {
return ""
}
return clean
}
@ -181,6 +201,9 @@ func extractEntitiesFromText(text string) []string {
var entities []string
for _, w := range words {
if len(w) > 1 && (w[0] >= 'A' && w[0] <= 'Z') {
if redact.ContainsSecret(w) {
continue // 2026-09-10 P0凭证形状的 token 不当实体
}
entities = append(entities, w)
}
}
@ -188,7 +211,9 @@ func extractEntitiesFromText(text string) []string {
}
func minz(a, b int) int {
if a < b { return a }
if a < b {
return a
}
return b
}

View File

@ -0,0 +1,82 @@
// 织忆 MemoryWeave — 图谱写入侧密钥过滤单元测试
// 2026-09-10 新增mc P0 事故):实体名含凭证时不得建节点/建边,
// 否则 ObsidianSync 会拿实体标题当文件名把密钥落成镜像。
package governance
import (
"strings"
"testing"
)
// 测试用假密钥(非真实凭证)
const fakeSecretEntity = "tskey-auth-kTESTONLY0000000000000000000000000000000000000000"
func TestCleanEntityName_DropsSecretShaped(t *testing.T) {
if got := cleanEntityName(fakeSecretEntity); got != "" {
t.Errorf("含凭证的实体名应返回空串,得到 %q", got)
}
if got := cleanEntityName("key=sk-TESTONLY000000000000000000000000000"); got != "" {
t.Errorf("含 sk- 凭证的实体名应返回空串,得到 %q", got)
}
}
func TestCleanEntityName_KeepsNormalUnicode(t *testing.T) {
for _, in := range []string{"牧尘", "织忆 MemoryWeave", "PostgreSQL", "192.168.123.11"} {
got := cleanEntityName(in)
if got == "" {
t.Errorf("正常实体 %q 不应被丢弃", in)
}
if strings.Contains(got, "REDACTED") {
t.Errorf("正常实体 %q 被误脱敏为 %q", in, got)
}
}
}
func TestExtractEntitiesFromText_SkipsSecret(t *testing.T) {
// extractEntitiesFromText 只收大写开头的 token凭证行需以大写引入才可能被收
text := "TsKey " + fakeSecretEntity + " Docker Nginx"
got := extractEntitiesFromText(text)
for _, e := range got {
if strings.Contains(e, "tskey-auth") {
t.Errorf("凭证形状的 token 不应成为实体: %q", e)
}
}
}
func TestUpdateFromDistill_NoNodeForSecretEntity(t *testing.T) {
g := NewInMemoryGraph()
up := NewAutoGraphUpdater(g)
up.UpdateFromDistill(&DistillInput{
EpisodeID: "ep_test_1",
Content: "配置 tailscale",
Facts: []string{"Tailscale 预授权 key 已写入配置文件并完成节点加入"},
Entities: []string{"Docker", fakeSecretEntity},
Namespace: "ns_test",
})
nodes := g.ListNodes("ns_test")
if len(nodes) == 0 {
t.Fatal("正常实体 Docker 应建出节点,实际 0 个节点")
}
for _, n := range nodes {
name, _ := n["name"].(string)
if strings.Contains(name, "tskey-auth") {
t.Errorf("凭证实体不该建节点,却出现: %q", name)
}
}
found := false
for _, n := range nodes {
if n["name"] == "Docker" {
found = true
}
}
if !found {
t.Errorf("正常实体 Docker 应保留,实际节点: %v", nodes)
}
// 边也不该引用凭证实体
for _, e := range g.ListNodes("") {
if strings.Contains(e["id"].(string), "tskey-auth") {
t.Errorf("凭证实体不该出现在边端点: %v", e)
}
}
}

View File

@ -604,8 +604,13 @@ func (gs *SQLiteGraphStore) CleanupNoiseNodes(dryRun bool) (int, []string, error
rows := queryRows(gs.db, "SELECT id, name FROM graph_nodes")
var noiseIDs []string
for _, row := range rows {
id := row["id"].(string)
name := row["name"].(string)
// 防 panicid 为 NULL如 daemon-distill 的 pattern 模板行id 无值)时
// row["id"] 是 nil interface直接 .(string) 会 panic → 跳过(无 id 也无法删除)
id, _ := row["id"].(string)
if id == "" {
continue
}
name, _ := row["name"].(string)
if findNoise(name) {
noiseIDs = append(noiseIDs, id)
}
@ -623,6 +628,67 @@ func (gs *SQLiteGraphStore) CleanupNoiseNodes(dryRun bool) (int, []string, error
return len(noiseIDs), noiseIDs, nil
}
// CleanupScopedNodes 删除指定 namespace 中名称含指定子串的噪音节点。
// 与 CleanupNoiseNodes编码噪音fts=/括号不匹配等)不同,这里按 namespace + 名称子串精确圈定,
// 用于清理多 agent 测试污染(如 openclaw-main/a06-main/hermes-main 中 2026-06 测试 agent
// A03/A04/卫安 产生的概念/事实节点),避免全表编码规则误删真实节点。
// namespace/nameContains 为空列表表示不限制谨慎使用nameContains 大小写不敏感。
// dryRun=true 时只检查不删除,返回预检结果。
func (gs *SQLiteGraphStore) CleanupScopedNodes(dryRun bool, namespaces, nameContains []string) (int, []string, error) {
gs.mu.Lock()
defer gs.mu.Unlock()
nsSet := make(map[string]struct{}, len(namespaces))
for _, ns := range namespaces {
if ns != "" {
nsSet[ns] = struct{}{}
}
}
patterns := make([]string, 0, len(nameContains))
for _, p := range nameContains {
if p != "" {
patterns = append(patterns, strings.ToLower(p))
}
}
rows := queryRows(gs.db, "SELECT id, name, namespace FROM graph_nodes")
var noiseIDs []string
for _, row := range rows {
id, _ := row["id"].(string)
if id == "" {
continue // NULL id 行pattern 模板)无法按 id 删除,跳过
}
name, _ := row["name"].(string)
ns, _ := row["namespace"].(string)
if len(nsSet) > 0 {
if _, ok := nsSet[ns]; !ok {
continue
}
}
lower := strings.ToLower(name)
matched := false
for _, p := range patterns {
if strings.Contains(lower, p) {
matched = true
break
}
}
if matched {
noiseIDs = append(noiseIDs, id)
}
}
if dryRun || len(noiseIDs) == 0 {
return len(noiseIDs), noiseIDs, nil
}
for _, nid := range noiseIDs {
execSQL(gs.db, fmt.Sprintf("DELETE FROM graph_edges WHERE source = '%s' OR target = '%s'", escape(nid), escape(nid)))
execSQL(gs.db, fmt.Sprintf("DELETE FROM graph_nodes WHERE id = '%s'", escape(nid)))
}
return len(noiseIDs), noiseIDs, nil
}
func (gs *SQLiteGraphStore) GetGraph(namespace string, limit int) ([]map[string]interface{}, []map[string]interface{}) {
nodes := []map[string]interface{}{}
edges := []map[string]interface{}{}
@ -941,21 +1007,26 @@ func (gs *SQLiteGraphStore) PageRank(damping float64, iterations int) map[string
}
for iter := 0; iter < iterations; iter++ {
newRanks := make(map[string]float64)
for _, node := range nodes {
rank := base
for src, edges := range outEdges {
totalWt := 0.0
for _, e := range edges {
totalWt += e.weight
}
for _, e := range edges {
if e.target == node && totalWt > 0 {
rank += damping * ranks[src] * e.weight / totalWt
}
}
// 正确 PageRank 实现 O(V+E):先归一化每个源节点总权重,
// 再沿出边把贡献直接累加到目标节点(旧实现每目标遍历全源 O(V²)
// 17628 节点 × 20 迭代 = 6.2 亿次 → 3 分钟;现在亚秒级)
contrib := make(map[string]float64, len(outEdges))
for src, edges := range outEdges {
totalWt := 0.0
for _, e := range edges {
totalWt += e.weight
}
newRanks[node] = rank
if totalWt <= 0 {
continue
}
share := damping * ranks[src] / totalWt
for _, e := range edges {
contrib[e.target] += share * e.weight
}
}
newRanks := make(map[string]float64, len(nodes))
for _, node := range nodes {
newRanks[node] = base + contrib[node]
}
ranks = newRanks
}

View File

@ -0,0 +1,74 @@
// Package redact —— 密钥脱敏唯一真源2026-09-10 P0 事故后固化)
//
// 事故背景:某轮蒸馏把含真密钥的原始对话当事实入库 → 知识图谱按该文本建实体
// → ObsidianSyncer.sanitizeFilename 用实体标题当文件名,于是
// `tskey-auth-kzu…` 被写进镜像文件名与正文(实例:小唯/07-Wiki/织忆/未找到命令.md
// 该 vault 又被 git 跟踪并推送 Gitea形成凭证泄露面。
//
// 本包提供内容侧唯一真源,供三处共用:
// - distill 质量门槛(事实入库前)
// - governance 图谱实体名清洗(建节点/建边前)
// - api/routes Obsidian 文件名 sanitize落盘前
package redact
import (
"regexp"
"strings"
)
// SecretPattern 匹配常见凭证形状。
//
// 注意Go RE2 不支持 lookbehind故"前一字符是否属于更长标识符"的判定
// 在 RedactSecrets/ContainsSecret 里手工做(见 isTokenByte
// `sk-` 段故意只吃 [A-Za-z0-9],避免误伤 URL 片段
// (如 https://a.com/task-driven-autonomous-agents 里的 "sk-driven…")。
var SecretPattern = regexp.MustCompile(
`sk-[A-Za-z0-9]{24,}` + // OpenAI / DeepSeek / Agnes / NewAPI 系
`|tskey-[A-Za-z0-9_-]{20,}` + // Tailscale 预授权 key
`|github_pat_[A-Za-z0-9_]{20,}` + // GitHub fine-grained PAT
`|ghp_[A-Za-z0-9]{30,}` + // GitHub classic PAT
`|nvapi-[A-Za-z0-9_-]{20,}` + // NVIDIA NIM
`|AKIA[0-9A-Z]{16}`) // AWS access key id
// Placeholder 脱敏后的替换文本。
const Placeholder = "<REDACTED-secret>"
// isTokenByte 判断字节是否可能属于一个更长 token用于剔除嵌在标识符里的假命中
func isTokenByte(c byte) bool {
return c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' ||
c >= '0' && c <= '9' || c == '_' || c == '-'
}
// RedactSecrets 把文本里所有凭证形状替换为 Placeholder无命中时原样返回。
func RedactSecrets(s string) string {
idx := SecretPattern.FindAllStringIndex(s, -1)
if len(idx) == 0 {
return s
}
var b strings.Builder
last := 0
for _, m := range idx {
start, end := m[0], m[1]
if start > 0 && isTokenByte(s[start-1]) {
continue // 前一位是单词字符 → 属于更长标识符,不替换
}
b.WriteString(s[last:start])
b.WriteString(Placeholder)
last = end
}
if last == 0 {
return s // 全部命中都被判为假阳性
}
b.WriteString(s[last:])
return b.String()
}
// ContainsSecret 判断文本是否含真凭证形状(用于实体名/标题直接丢弃的场景)。
func ContainsSecret(s string) bool {
for _, m := range SecretPattern.FindAllStringIndex(s, -1) {
if m[0] == 0 || !isTokenByte(s[m[0]-1]) {
return true
}
}
return false
}

View File

@ -0,0 +1,80 @@
// 织忆 MemoryWeave — 凭证脱敏单元测试
// 2026-09-10 新增mc P0 事故):锁死「密钥不得进记忆/图谱/Obsidian 镜像」这个回归。
package redact
import (
"strings"
"testing"
)
// 测试用假密钥(非真实凭证,按各平台形状构造)
const (
fakeTailscale = "tskey-auth-kTESTONLY0000000000000000000000000000000000000000000000000" // 61 位
fakeOpenAI = "sk-TESTONLY000000000000000000000000000" // 48 位
fakeGithubPAT = "github_pat_11TESTONLY0000000000000000000000000000000000000000000000000000000000000000"
fakeNvapi = "nvapi-TESTONLY0000000000000000000000000000"
)
func TestRedactSecrets_KnownShapes(t *testing.T) {
cases := []struct {
name string
in string
}{
{"tailscale", "配置 tailscale up --authkey " + fakeTailscale + " 完成"},
{"openai", "OPENAI_API_KEY=" + fakeOpenAI},
{"github_pat", "GITHUB_PERSONAL_ACCESS_TOKEN: " + fakeGithubPAT},
{"nvapi", "key=" + fakeNvapi},
{"aws", "AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE"},
}
for _, tc := range cases {
got := RedactSecrets(tc.in)
if strings.Contains(got, "<REDACTED-secret>") != true {
t.Errorf("%s: 未脱敏: %q", tc.name, got)
}
}
// 脱敏后不得再残留任何形状
for _, tc := range cases {
if ContainsSecret(RedactSecrets(tc.in)) {
t.Errorf("%s: 脱敏后仍能匹配到凭证形状", tc.name)
}
}
}
// URL / 普通连字符词里的 "sk-" 片段不得误伤(事故里 false positive 的来源)
func TestRedactSecrets_NoFalsePositiveOnURL(t *testing.T) {
in := "参考 https://yoheinakajima.com/task-driven-autonomous-agents 与 babybeeagi-task-xxx 的说明"
if got := RedactSecrets(in); got != in {
t.Errorf("URL 片段被误脱敏:\n got=%q\nwant=%q", got, in)
}
if ContainsSecret(in) {
t.Error("ContainsSecret 对 URL 片段误报")
}
}
// 嵌在更长标识符里的假命中不替换(前一位是单词字符)
func TestRedactSecrets_EmbeddedTokenIgnored(t *testing.T) {
in := "字段名 xxxkey_sk-TESTONLY0000000000000000000000000001234 不是凭证"
if got := RedactSecrets(in); got != in {
t.Errorf("嵌在标识符里的 token 被误脱敏: %q", got)
}
}
func TestRedactSecrets_IdempotentAndPlainTextUntouched(t *testing.T) {
plain := "牧尘喜欢结论先行,不要废话科普。"
if got := RedactSecrets(plain); got != plain {
t.Errorf("纯文本被改动: %q", got)
}
once := RedactSecrets("key=" + fakeOpenAI)
if twice := RedactSecrets(once); twice != once {
t.Errorf("脱敏不幂等: %q → %q", once, twice)
}
}
func TestContainsSecret(t *testing.T) {
if !ContainsSecret("authkey " + fakeTailscale) {
t.Error("真实形状应被判为含凭证")
}
if ContainsSecret("普通的记忆内容,没有凭证") {
t.Error("普通内容误报")
}
}

View File

@ -52,7 +52,7 @@ func BenchmarkDashboard_RecordFeedback_Parallel(b *testing.B) {
// ─── 缺口检测 ────────────────────────────────────────────
func BenchmarkGapDetector_RecordMiss(b *testing.B) {
gd := NewGapDetector()
gd := NewGapDetector(nil, nil)
topics := []string{"config", "memory", "system", "docker", "nginx"}
b.ResetTimer()
for i := 0; i < b.N; i++ {
@ -61,7 +61,7 @@ func BenchmarkGapDetector_RecordMiss(b *testing.B) {
}
func BenchmarkGapDetector_List(b *testing.B) {
gd := NewGapDetector()
gd := NewGapDetector(nil, nil)
// 预填充 100 个缺口
for i := 0; i < 100; i++ {
topic := fmt.Sprintf("gap-%d", i)
@ -76,7 +76,7 @@ func BenchmarkGapDetector_List(b *testing.B) {
}
func BenchmarkGapDetector_Close(b *testing.B) {
gd := NewGapDetector()
gd := NewGapDetector(nil, nil)
for i := 0; i < 100; i++ {
topic := fmt.Sprintf("close-%d", i)
for j := 0; j < 3; j++ {

View File

@ -94,7 +94,7 @@ func TestDashboard_RecordConflictResolved(t *testing.T) {
// ─── 知识缺口检测 ────────────────────────────────────────
func TestGapDetector_NotTriggeredBeforeThreshold(t *testing.T) {
gd := NewGapDetector()
gd := NewGapDetector(nil, nil)
gap := gd.RecordMiss("kubernetes")
if gap != nil {
t.Error("should not trigger gap after 1 miss")
@ -106,7 +106,7 @@ func TestGapDetector_NotTriggeredBeforeThreshold(t *testing.T) {
}
func TestGapDetector_DetectsGapAtThreshold(t *testing.T) {
gd := NewGapDetector()
gd := NewGapDetector(nil, nil)
gd.RecordMiss("topicX")
gd.RecordMiss("topicX")
gap := gd.RecordMiss("topicX")
@ -123,7 +123,7 @@ func TestGapDetector_DetectsGapAtThreshold(t *testing.T) {
}
func TestGapDetector_CloseGap(t *testing.T) {
gd := NewGapDetector()
gd := NewGapDetector(nil, nil)
gd.RecordMiss("topic")
gd.RecordMiss("topic")
gd.RecordMiss("topic")
@ -136,7 +136,7 @@ func TestGapDetector_CloseGap(t *testing.T) {
}
func TestGapDetector_ClassifySynonym(t *testing.T) {
gd := NewGapDetector()
gd := NewGapDetector(nil, nil)
gd.RecordMiss("API")
gd.RecordMiss("API")
gd.RecordMiss("API")
@ -277,7 +277,7 @@ func BenchmarkDashboardMetrics(b *testing.B) {
}
func BenchmarkGapDetection(b *testing.B) {
gd := NewGapDetector()
gd := NewGapDetector(nil, nil)
b.ResetTimer()
for i := 0; i < b.N; i++ {
gd.RecordMiss("benchmark_topic")

View File

@ -0,0 +1,200 @@
package storage
import (
"encoding/binary"
"encoding/json"
"fmt"
"io"
"net"
"os"
"testing"
"time"
)
// 端到端验证 lancedb_update_batch IPC需要真实 sidecar socket
// 用法:
//
// ZHIYI_TEST_IPC_SOCK=/tmp/ipc-batch-test.sock go test ./internal/storage/ -run TestBatchIPC -v
//
// 验证两点:① 批量调用按 delta 正确累加 recall_count② 整批只产生 1 个 LanceDB 版本(由脚本比对
// _versions 目录文件数确认,见 AC 记录)。
func ipcCall(t *testing.T, sock string, req map[string]any) map[string]any {
t.Helper()
conn, err := net.DialTimeout("unix", sock, 5*time.Second)
if err != nil {
t.Fatalf("dial %s: %v", sock, err)
}
defer conn.Close()
body, _ := json.Marshal(req)
var lenBuf [4]byte
binary.BigEndian.PutUint32(lenBuf[:], uint32(len(body)))
if _, err := conn.Write(lenBuf[:]); err != nil {
t.Fatalf("write len: %v", err)
}
if _, err := conn.Write(body); err != nil {
t.Fatalf("write body: %v", err)
}
if _, err := io.ReadFull(conn, lenBuf[:]); err != nil {
t.Fatalf("read len: %v", err)
}
respBuf := make([]byte, binary.BigEndian.Uint32(lenBuf[:]))
if _, err := io.ReadFull(conn, respBuf); err != nil {
t.Fatalf("read body: %v", err)
}
var resp map[string]any
if err := json.Unmarshal(respBuf, &resp); err != nil {
t.Fatalf("unmarshal resp: %v", err)
}
if resp["status"] != "ok" {
t.Fatalf("IPC status=%v detail=%v", resp["status"], resp["error_detail"])
}
return resp
}
func scanRecallCount(t *testing.T, sock, wantID string) int64 {
t.Helper()
resp := ipcCall(t, sock, map[string]any{"type": "lancedb_scan", "limit": 2000})
var records []struct {
ID string `json:"id"`
RecallCount int64 `json:"recall_count"`
}
if err := json.Unmarshal([]byte(resp["report_json"].(string)), &records); err != nil {
t.Fatalf("unmarshal scan: %v", err)
}
for _, r := range records {
if r.ID == wantID {
return r.RecallCount
}
}
t.Fatalf("id %s 不在 scan 结果中(%d 条)", wantID, len(records))
return 0
}
func TestBatchIPCVersionAccounting(t *testing.T) {
sock := os.Getenv("ZHIYI_TEST_IPC_SOCK")
dir := os.Getenv("ZHIYI_TEST_DATA_DIR")
if sock == "" || dir == "" {
t.Skip("ZHIYI_TEST_IPC_SOCK / ZHIYI_TEST_DATA_DIR 未设置,跳过版本计数对比")
}
countVersions := func() int {
entries, err := os.ReadDir(dir + "/memories.lance/_versions")
if err != nil {
t.Fatalf("read versions dir: %v", err)
}
n := 0
for _, e := range entries {
if !e.IsDir() && len(e.Name()) > 9 && e.Name()[len(e.Name())-9:] == ".manifest" {
n++
}
}
return n
}
resp := ipcCall(t, sock, map[string]any{"type": "lancedb_scan", "limit": 20})
var recs []struct {
ID string `json:"id"`
}
if err := json.Unmarshal([]byte(resp["report_json"].(string)), &recs); err != nil || len(recs) < 6 {
t.Fatalf("样本不足: err=%v n=%d", err, len(recs))
}
ts := time.Now().Format(time.RFC3339)
// A) 新路径一次批量3 条不同记忆)
v0 := countVersions()
items, _ := json.Marshal([]RecallWriteItem{{ID: recs[0].ID, Delta: 1}, {ID: recs[1].ID, Delta: 1}, {ID: recs[2].ID, Delta: 1}})
ipcCall(t, sock, map[string]any{"type": "lancedb_update_batch", "items": string(items), "ts": ts})
v1 := countVersions()
batchVersions := v1 - v0
// B) 旧路径3 次单条 lancedb_updaterecall 读路径原行为)
fields, _ := json.Marshal([]map[string]string{
{"column": "recall_count", "value": "1"},
{"column": "last_recalled_at", "value": "'" + ts + "'"},
{"column": "freshness", "value": "'verified'"},
})
for i := 3; i < 6; i++ {
ipcCall(t, sock, map[string]any{"type": "lancedb_update", "table": "memories", "id": recs[i].ID, "fields": string(fields)})
}
v2 := countVersions()
perItemVersions := v2 - v1
fmt.Printf("VERSION_ACCOUNTING: 批量3条 → %d 个版本 | 逐条3次 → %d 个版本\n", batchVersions, perItemVersions)
if perItemVersions <= batchVersions {
t.Fatalf("逐条(%d) 应比批量(%d) 产生更多版本", perItemVersions, batchVersions)
}
}
// 只读校验:扫生产 socket确认批量落盘真的写进了 recall_count不产生任何写操作
//
// ZHIYI_TEST_IPC_SOCK=/tmp/zhiyi-ipc.sock go test ./internal/storage/ -run TestBatchIPCScanReadOnly -v
func TestBatchIPCScanReadOnly(t *testing.T) {
sock := os.Getenv("ZHIYI_TEST_IPC_SOCK")
if sock == "" {
t.Skip("ZHIYI_TEST_IPC_SOCK 未设置")
}
resp := ipcCall(t, sock, map[string]any{"type": "lancedb_scan", "limit": 2000})
var recs []struct {
ID string `json:"id"`
RecallCount int64 `json:"recall_count"`
LastRecalledAt string `json:"last_recalled_at"`
}
if err := json.Unmarshal([]byte(resp["report_json"].(string)), &recs); err != nil {
t.Fatalf("unmarshal: %v", err)
}
withCount, maxCount := 0, int64(0)
var sampleID, sampleTS string
for _, r := range recs {
if r.RecallCount > 0 {
withCount++
if r.RecallCount > maxCount {
maxCount = r.RecallCount
sampleID, sampleTS = r.ID, r.LastRecalledAt
}
}
}
fmt.Printf("SCAN_READONLY: 总 %d 条, 其中 recall_count>0 的 %d 条, 最大 recall_count=%d (id=%s last_recalled_at=%s)\n",
len(recs), withCount, maxCount, sampleID, sampleTS)
if withCount == 0 {
t.Fatalf("scan 里没有任何 recall_count>0 —— 批量落盘可能没生效")
}
}
func TestBatchIPCUpdateRecallBatch(t *testing.T) {
sock := os.Getenv("ZHIYI_TEST_IPC_SOCK")
if sock == "" {
t.Skip("ZHIYI_TEST_IPC_SOCK 未设置,跳过 IPC 端到端验证")
}
// 取一条真实记忆做样本
resp := ipcCall(t, sock, map[string]any{"type": "lancedb_scan", "limit": 50})
var recs []struct {
ID string `json:"id"`
RecallCount int64 `json:"recall_count"`
}
if err := json.Unmarshal([]byte(resp["report_json"].(string)), &recs); err != nil || len(recs) == 0 {
t.Fatalf("scan 无样本: err=%v n=%d", err, len(recs))
}
target := recs[0].ID
before := scanRecallCount(t, sock, target)
// 批量:同一条 delta=3 + 另一条 delta=1验证合并语义与多行单事务
second := target
if len(recs) > 1 {
second = recs[1].ID
}
items, _ := json.Marshal([]RecallWriteItem{{ID: target, Delta: 3}, {ID: second, Delta: 1}})
ts := time.Now().Format(time.RFC3339)
out := ipcCall(t, sock, map[string]any{
"type": "lancedb_update_batch",
"table": "memories",
"items": string(items),
"ts": ts,
})
t.Logf("batch resp: %v", out)
after := scanRecallCount(t, sock, target)
if after != before+3 {
t.Fatalf("recall_count 期望 %d实际 %ddelta 未正确累加)", before+3, after)
}
fmt.Printf("BATCH_IPC_OK id=%s recall_count %d → %d\n", target, before, after)
}

View File

@ -46,10 +46,11 @@ func BenchmarkEmbedder_Batch50(b *testing.B) {
// ─── Recall Pipeline 基准 ─────────────────────────────────
func BenchmarkRecallPipeline_10Docs(b *testing.B) {
emb := &Embedder{endpoint: "http://localhost:8000/v1/embeddings"}
p := NewRecallPipeline(
&Embedder{endpoint: "http://localhost:8000/v1/embeddings"},
emb,
NewMemLanceClient(nil),
NewReranker("http://localhost:8001/rerank"),
NewReranker("http://localhost:8001/rerank", emb),
)
b.ResetTimer()
for i := 0; i < b.N; i++ {
@ -58,10 +59,11 @@ func BenchmarkRecallPipeline_10Docs(b *testing.B) {
}
func BenchmarkRecallPipeline_50Docs(b *testing.B) {
emb := &Embedder{endpoint: "http://localhost:8000/v1/embeddings"}
p := NewRecallPipeline(
&Embedder{endpoint: "http://localhost:8000/v1/embeddings"},
emb,
NewMemLanceClient(nil),
NewReranker("http://localhost:8001/rerank"),
NewReranker("http://localhost:8001/rerank", emb),
)
b.ResetTimer()
for i := 0; i < b.N; i++ {

View File

@ -9,6 +9,7 @@ import (
"net/http"
"os"
"sync"
"time"
)
// Embedder bge-m3 编码客户端。优先使用本地 vLLM端口 8000fallback 到模力方舟 API。
@ -33,7 +34,7 @@ func NewEmbedder(endpoint string) *Embedder {
endpoint: endpoint,
modelName: getModelName(),
apiKey: os.Getenv("MOLIFANG_API_KEY"),
httpClient: &http.Client{},
httpClient: &http.Client{Timeout: 8 * time.Second},
dim: 1024,
}
}

View File

@ -161,6 +161,17 @@ func (c *cachedGraphStore) CleanupNoiseNodes(dryRun bool) (int, []string, error)
return c.inner.CleanupNoiseNodes(dryRun)
}
// CleanupScopedNodes 透传inner 不支持该能力时静默返回 0生产后端 SQLiteGraphStore 已实现)
func (c *cachedGraphStore) CleanupScopedNodes(dryRun bool, namespaces, nameContains []string) (int, []string, error) {
type scoped interface {
CleanupScopedNodes(dryRun bool, namespaces, nameContains []string) (int, []string, error)
}
if s, ok := c.inner.(scoped); ok {
return s.CleanupScopedNodes(dryRun, namespaces, nameContains)
}
return 0, nil, nil
}
// P0: FallbackTextSearch 透传到内层
func (c *cachedGraphStore) FallbackTextSearch(query, namespace string, limit int) []map[string]interface{} {
return c.inner.FallbackTextSearch(query, namespace, limit)

View File

@ -44,6 +44,8 @@ type ipcReq struct {
Fields string `json:"fields,omitempty"`
MinRecall int `json:"min_recall,omitempty"`
QueryLimit int `json:"limit,omitempty"`
Items string `json:"items,omitempty"` // lancedb_update_batch: [{"id":..,"delta":N}]
TS string `json:"ts,omitempty"` // lancedb_update_batch: last_recalled_at
}
type ipcResp struct {
@ -450,6 +452,38 @@ func (rc *RustLanceDBClient) Update(table, id string, fields map[string]any) err
return nil
}
// UpdateRecallBatch 单事务批量更新召回元数据2026-09-12 优化B
// Rust 侧把整批按 delta 分组,每组一次 update`recall_count + delta` + id IN (...))提交
// lance 不支持 CASE WHEN→ 整批版本数 = 不同 delta 的个数,通常 1 个(旧实现:每行 1 个)。
// 注Rust 侧固定操作 memories 表table 参数暂未使用。
func (rc *RustLanceDBClient) UpdateRecallBatch(table string, items []RecallWriteItem, lastRecalledAt string) (int64, error) {
if len(items) == 0 {
return 0, nil
}
fj, err := json.Marshal(items)
if err != nil {
return 0, err
}
resp, err := rc.rpc(ipcReq{
Type: "lancedb_update_batch",
Table: table,
Items: string(fj),
TS: lastRecalledAt,
})
if err != nil {
return 0, err
}
var out struct {
Updated int64 `json:"updated"`
Items int `json:"items"`
}
if resp.ReportJSON != "" {
_ = json.Unmarshal([]byte(resp.ReportJSON), &out)
}
log.Printf("[ipc] UpdateRecallBatch: 合并 %d 条记忆 → 落盘 %d 行(单事务)", len(items), out.Updated)
return out.Updated, nil
}
func (rc *RustLanceDBClient) GetTopByQuality(agentID string, limit int) ([]models.MemoryRecord, error) {
_local.mu.RLock()
defer _local.mu.RUnlock()
@ -488,6 +522,47 @@ func (rc *RustLanceDBClient) SoftDelete(id, reason string) error {
func (rc *RustLanceDBClient) GetVersionHistory(id string) ([]map[string]interface{}, error) { return nil, nil }
func (rc *RustLanceDBClient) GetCandidatesForForgetting() ([]map[string]interface{}, error) {
// P2 2026-09-06: 走 Rust 全表扫描(安全版: limit 2000 + 跳 vector + 真读 last_recalled_at)。
// 替代纯缓存遍历 — 缓存只含近期 commit/recall/search 触碰的记忆, 到期旧记忆扫不到 → 9-25 遗忘闭环断。
// IPC 失败/空 → fallback 缓存遍历(保持可用)。
resp, err := rc.rpc(ipcReq{Type: "lancedb_scan", QueryLimit: 2000})
if err == nil && resp.ReportJSON != "" {
var raw []struct {
ID string `json:"id"`
Content string `json:"content"`
Namespace string `json:"namespace"`
Category string `json:"category"`
Tier string `json:"tier"`
IsDeleted bool `json:"is_deleted"`
Importance float64 `json:"importance"`
RecallCount int64 `json:"recall_count"`
LastRecalledAt string `json:"last_recalled_at"`
CreatedAt string `json:"created_at"`
}
uerr := json.Unmarshal([]byte(resp.ReportJSON), &raw)
if uerr == nil && len(raw) > 0 {
log.Printf("[ipc] GetCandidatesForForgetting: Rust scan → %d candidates", len(raw))
out := make([]map[string]interface{}, 0, len(raw))
for _, r := range raw {
if r.IsDeleted || r.Tier == "core" {
continue
}
out = append(out, map[string]interface{}{
"id": r.ID,
"content": r.Content,
"namespace": r.Namespace,
"category": r.Category,
"tier": r.Tier,
"importance": r.Importance,
"recall_count": int(r.RecallCount), // decay 侧 .(int) 断言 — int64 会恒 0
"last_recalled_at": r.LastRecalledAt,
"created_at": r.CreatedAt,
})
}
return out, nil
}
log.Printf("[ipc] GetCandidatesForForgetting: scan unmarshal empty (rpc=%v err=%v raw=%d) → fallback 缓存", err, uerr, len(raw))
}
_local.mu.RLock()
defer _local.mu.RUnlock()
var out []map[string]interface{}
@ -497,7 +572,9 @@ func (rc *RustLanceDBClient) GetCandidatesForForgetting() ([]map[string]interfac
"id": id,
"content": m.Content,
"namespace": m.Namespace,
"last_recalled_at": m.LastRecalledAt,
"category": m.Category,
"last_recalled_at": fmtTimeRFC3339(m.LastRecalledAt),
"created_at": fmtTimeRFC3339(m.CreatedAt),
"recall_count": m.RecallCount,
"tier": m.Tier,
"importance": m.Importance,
@ -507,6 +584,17 @@ func (rc *RustLanceDBClient) GetCandidatesForForgetting() ([]map[string]interfac
return out, nil
}
// fmtTimeRFC3339 统一时间输出契约RFC3339 字符串;零值输出 ""。
// 2026-09-06 P1 修复:此前直接放 time.Time → server.go decay 的 .(string) 断言
// 恒失败 → lastAccessed 永远"默认30天前" → 遗忘判定失真(叠加数据年龄未到)。
// 调用方一律按 string 解析parseTimeStr / time.Parse RFC3339
func fmtTimeRFC3339(t time.Time) string {
if t.IsZero() {
return ""
}
return t.Format(time.RFC3339)
}
func mapStr(m map[string]interface{}, key string) string {
if v, ok := m[key]; ok {
if s, ok := v.(string); ok {

View File

@ -73,7 +73,7 @@ func (p *RecallPipeline) Recall(query, namespace string, topK int, diversity flo
if len(filtered) == 0 && len(results) > 0 {
filtered = results
}
go p.incrementRecallCount(results)
p.recordRecallWrites(filtered)
return filtered, nil
}
}
@ -258,26 +258,11 @@ func (p *RecallPipeline) Recall(query, namespace string, topK int, diversity flo
}
}
// 将召回结果写入本地缓存goroutine 依赖此 cache 做 increment
// cache key = id与 Update() 中 lookup key 一致
for _, r := range results {
if r.ID == "" {
continue
}
_local.mu.Lock()
if existing, ok := _local.memories[r.ID]; ok {
existing.RecallCount++
} else {
_local.memories[r.ID] = &models.MemoryRecord{
ID: r.ID,
RecallCount: 1,
}
}
_local.mu.Unlock()
}
// Async increment recall_count + update last_recalled_at + importance
go p.incrementRecallCount(results)
// 召回元数据写2026-09-12 优化B累积延迟更新
// 旧实现:对每条结果同步 lancedb.Update → LanceDB MVCC 每写一行一个版本
// (实测 15 版本/分钟 → _versions 膨胀到 17.7G,根因所在)
// 现在:只把命中 ID 放进 RecallWriteBuffer后台按窗口批量单事务提交每批 1 个版本)
p.recordRecallWrites(results)
// 共访追踪(即使无 prefetch pusher 也运行,用于持久化)
go func() {
@ -410,6 +395,28 @@ func cosineSimilarity(a, b []float32) float64 {
return sum
}
// recordRecallWrites 记录召回元数据写。
// 主路径累积延迟更新缓冲RecallWriteBufferInstance——读路径零同步写、批量单事务落盘。
// 兜底:缓冲未初始化(如单测/独立调用)时退回旧的逐条异步 Update。
func (p *RecallPipeline) recordRecallWrites(results []models.RecallResult) {
ids := make([]string, 0, len(results))
for _, r := range results {
if r.ID != "" {
ids = append(ids, r.ID)
}
}
if len(ids) == 0 {
return
}
if RecallWriteBufferInstance != nil {
RecallWriteBufferInstance.Record(ids)
return
}
go p.incrementRecallCount(results)
}
// incrementRecallCount 旧版逐条同步写(每条结果 = 1 个 LanceDB 版本)。
// 已不作为主路径,仅保留为缓冲不可用时的兜底。
func (p *RecallPipeline) incrementRecallCount(results []models.RecallResult) {
now := time.Now().Format(time.RFC3339)
for _, r := range results {

View File

@ -0,0 +1,281 @@
package storage
import (
"log"
"os"
"sort"
"strconv"
"sync"
"sync/atomic"
"time"
)
// ─── 召回元数据「累积延迟更新」缓冲2026-09-12 优化B─────────────────────────
//
// 问题根因recall 读路径原先对**每条结果**同步调用 lancedb.Update
// recall_count / last_recalled_at / freshness。LanceDB 是 MVCC 存储——
// **每次 update 提交产生一个版本**。实测生产15 版本/分钟、_versions 目录
// 膨胀到 17.72G(真实数据仅 22M75,327 个 .manifest放大 800 倍)。
//
// 方案:
//
// Record() 读路径只把增量写进内存缓冲;**同一记忆在一个窗口内多次命中合并成 1 条 delta**
// Flush() 后台按窗口(默认 300s+ 阈值(默认 256 条不同记忆)批量提交一次;
// Rust 侧 lancedb_update_batch 把整批**按 delta 分组**,每组一次
// update(`recall_count + delta` + id IN (...)) 提交lance 不支持 CASE WHEN
// → 每批版本数 = 不同 delta 的个数delta 绝大多数为 1故通常 1 个版本;
// 旧实现:每行 1 个版本)
//
// 语义取舍(明确记录,便于日后审计):
// - last_recalled_at / freshness 最多延迟一个窗口(分钟级)。遗忘/衰减判定以「天」为单位,无影响。
// - 进程崩溃会丢最后一个窗口的增量(召回统计,不是记忆数据本身),可接受。
// - recall_count 由 Rust 侧「读库现值 + delta」计算不是读 Go 缓存),顺带修掉旧实现里
// 「本地缓存 +1 后再 +1」导致的计数漂移。
type RecallWriteItem struct {
ID string `json:"id"`
Delta int `json:"delta"`
}
// BatchRecallUpdater 单事务批量更新接口。
// 由 Rust IPC 后端RustLanceDBClient实现其他后端不支持时自动退化为逐条 Update。
type BatchRecallUpdater interface {
UpdateRecallBatch(table string, items []RecallWriteItem, lastRecalledAt string) (int64, error)
}
const (
defaultRecallFlushInterval = 300 * time.Second
defaultRecallFlushMaxItems = 256
)
type pendingRecall struct {
delta int
lastAt time.Time
}
// RecallWriteBuffer 累积召回元数据写,延迟批量落盘。
type RecallWriteBuffer struct {
mu sync.Mutex
pending map[string]*pendingRecall
ldb LanceDB
batch BatchRecallUpdater
interval time.Duration
maxItems int
stopCh chan struct{}
doneCh chan struct{}
stopped bool
startOnce sync.Once
flushCount int64
flushItems int64
flushRows int64
flushErrors int64
fallbackRows int64
}
// RecallWriteBufferInstance 进程级单例(与 SearchCacheInstance / CoOccurTrackerInstance 同模式)
var RecallWriteBufferInstance *RecallWriteBuffer
// NewRecallWriteBuffer 创建缓冲并启动后台 flush 协程interval<=0 表示只按阈值触发。
func NewRecallWriteBuffer(ldb LanceDB, interval time.Duration, maxItems int) *RecallWriteBuffer {
if maxItems <= 0 {
maxItems = defaultRecallFlushMaxItems
}
b := &RecallWriteBuffer{
pending: make(map[string]*pendingRecall),
ldb: ldb,
interval: interval,
maxItems: maxItems,
stopCh: make(chan struct{}),
doneCh: make(chan struct{}),
}
// Rust IPC 后端支持单事务批量更新CASE WHEN 一次提交)
if bu, ok := ldb.(BatchRecallUpdater); ok {
b.batch = bu
}
go b.loop()
return b
}
// InitRecallWriteBuffer 初始化进程级单例server 启动时调用一次)。
// 窗口可用环境变量 RECALL_WRITE_FLUSH_SECONDS 覆盖(运维/验证用)。
func InitRecallWriteBuffer(ldb LanceDB) *RecallWriteBuffer {
if RecallWriteBufferInstance != nil {
return RecallWriteBufferInstance
}
interval := defaultRecallFlushInterval
if v := os.Getenv("RECALL_WRITE_FLUSH_SECONDS"); v != "" {
if n, err := strconv.Atoi(v); err == nil && n >= 0 {
interval = time.Duration(n) * time.Second
}
}
b := NewRecallWriteBuffer(ldb, interval, defaultRecallFlushMaxItems)
RecallWriteBufferInstance = b
log.Printf("[recall-buffer] 已启用累积延迟更新: flush 窗口=%s, 阈值=%d 条, 单事务批量=%v",
interval, b.maxItems, b.batch != nil)
return b
}
// StopRecallWriteBuffer 停止单例并落盘最后一个窗口(进程优雅退出时调用)。
func StopRecallWriteBuffer() {
if b := RecallWriteBufferInstance; b != nil {
b.Stop()
}
}
func (b *RecallWriteBuffer) loop() {
defer close(b.doneCh)
if b.interval <= 0 {
<-b.stopCh
return
}
t := time.NewTicker(b.interval)
defer t.Stop()
for {
select {
case <-b.stopCh:
return
case <-t.C:
b.Flush()
}
}
}
// Record 累积一次召回命中的记忆 ID同窗口内同 ID 合并 delta
// 返回当前待写条数;达到阈值时异步触发一次 flush不阻塞读路径
func (b *RecallWriteBuffer) Record(ids []string) int {
now := time.Now()
b.mu.Lock()
for _, id := range ids {
if id == "" {
continue
}
p, ok := b.pending[id]
if !ok {
p = &pendingRecall{}
b.pending[id] = p
}
p.delta++
p.lastAt = now
}
n := len(b.pending)
b.mu.Unlock()
if n >= b.maxItems {
go b.Flush()
}
return n
}
// Flush 取出当前窗口的全部增量并批量提交(每批最多 1 个 LanceDB 版本)。
func (b *RecallWriteBuffer) Flush() (int, error) {
b.mu.Lock()
if len(b.pending) == 0 {
b.mu.Unlock()
return 0, nil
}
batch := make([]RecallWriteItem, 0, len(b.pending))
var lastAt time.Time
for id, p := range b.pending {
if p.delta <= 0 {
continue
}
batch = append(batch, RecallWriteItem{ID: id, Delta: p.delta})
if p.lastAt.After(lastAt) {
lastAt = p.lastAt
}
}
b.pending = make(map[string]*pendingRecall)
b.mu.Unlock()
if len(batch) == 0 {
return 0, nil
}
sort.Slice(batch, func(i, j int) bool { return batch[i].ID < batch[j].ID })
ts := lastAt.Format(time.RFC3339)
var rows int64
var err 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 {
log.Printf("[recall-buffer] 单事务批量提交失败(%v) → 退化逐条 Update数据不丢版本数不优化", err)
atomic.AddInt64(&b.flushErrors, 1)
}
rows = 0
for _, it := range batch {
if uerr := b.ldb.Update("memories", it.ID, map[string]any{
"recall_count": map[string]string{"$inc": "1"},
"last_recalled_at": ts,
"freshness": "verified",
}); uerr != nil {
log.Printf("[recall-buffer] 逐条 Update 失败 id=%s: %v", it.ID, uerr)
continue
}
rows++
}
atomic.AddInt64(&b.fallbackRows, rows)
}
// 落盘成功后同步进程内缓存(缓存只是热数据,权威值在 LanceDB
if rows > 0 {
_local.mu.Lock()
for _, it := range batch {
if m, ok := _local.memories[it.ID]; ok {
m.RecallCount += it.Delta
if !lastAt.IsZero() {
m.LastRecalledAt = lastAt
}
}
}
_local.mu.Unlock()
}
atomic.AddInt64(&b.flushCount, 1)
atomic.AddInt64(&b.flushItems, int64(len(batch)))
atomic.AddInt64(&b.flushRows, rows)
log.Printf("[recall-buffer] flush: 合并 %d 条记忆 → 落盘 %d 行, 单事务=%v, fallback=%v",
len(batch), rows, !usedFallback, usedFallback)
return int(rows), nil
}
// Stop 停止后台 flush 协程,并把最后一个窗口落盘(进程优雅退出时调用)。
func (b *RecallWriteBuffer) Stop() {
b.mu.Lock()
if b.stopped {
b.mu.Unlock()
return
}
b.stopped = true
b.mu.Unlock()
close(b.stopCh)
select {
case <-b.doneCh:
case <-time.After(10 * time.Second):
log.Printf("[recall-buffer] 停止超时,直接落盘剩余窗口")
}
b.Flush()
}
// Stats 观测用flush 频率 / 合并率 / 版本写入行数)。
func (b *RecallWriteBuffer) Stats() map[string]interface{} {
b.mu.Lock()
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,
}
}

View File

@ -0,0 +1,221 @@
package storage
import (
"fmt"
"sync"
"testing"
"time"
"github.com/xiaoxue/memoryweave/internal/models"
)
// fakeLDB 记录写入调用次数:验证「读路径零同步写 + 窗口内合并 + 单事务批量」。
type fakeLDB struct {
mu sync.Mutex
batchCalls int
batchItems []RecallWriteItem
batchRowTotal int64
updateCalls int
updateIDs []string
forceBatchErr bool
}
func (f *fakeLDB) InsertEpisode(agentID, namespace, content, category string) (string, error) {
return "", nil
}
func (f *fakeLDB) GetTopByQuality(agentID string, limit int) ([]models.MemoryRecord, error) {
return nil, nil
}
func (f *fakeLDB) Stats() (map[string]interface{}, error) { return nil, nil }
func (f *fakeLDB) SoftDelete(id, reason string) error { return nil }
func (f *fakeLDB) GetVersionHistory(id string) ([]map[string]interface{}, error) {
return nil, nil
}
func (f *fakeLDB) GetCandidatesForForgetting() ([]map[string]interface{}, error) {
return nil, nil
}
func (f *fakeLDB) GetSkillCandidates(minRecalls, limit int) ([]models.MemoryRecord, error) {
return nil, nil
}
func (f *fakeLDB) Backup(path string) error { return nil }
func (f *fakeLDB) GetAuditLog(limit int) ([]map[string]interface{}, error) { return nil, nil }
func (f *fakeLDB) IncrementUseful(id string) {}
func (f *fakeLDB) IncrementNotUseful(id string) {}
func (f *fakeLDB) UpdateMemoryContent(id, newContent, source string) error {
return nil
}
func (f *fakeLDB) InsertMemory(m models.MemoryRecord) error { return nil }
func (f *fakeLDB) Search(table string, vector []float32, topK int, namespaceFilter string) ([]models.MemoryRecord, error) {
return nil, nil
}
func (f *fakeLDB) Insert(table string, record any) error { return nil }
func (f *fakeLDB) Update(table, id string, fields map[string]any) error {
f.mu.Lock()
f.updateCalls++
f.updateIDs = append(f.updateIDs, id)
f.mu.Unlock()
return nil
}
// BatchRecallUpdater 实现(单事务批量)
func (f *fakeLDB) UpdateRecallBatch(table string, items []RecallWriteItem, lastRecalledAt string) (int64, error) {
if f.forceBatchErr {
return 0, fmt.Errorf("forced batch failure")
}
f.mu.Lock()
f.batchCalls++
f.batchItems = append(f.batchItems, items...)
f.batchRowTotal += int64(len(items))
f.mu.Unlock()
return int64(len(items)), nil
}
// AC-2同一记忆在一个窗口内多次召回 → 只写 1 次delta 合并),且是单事务批量调用
func TestRecallWriteBufferCoalescesWithinWindow(t *testing.T) {
fake := &fakeLDB{}
b := NewRecallWriteBuffer(fake, 0, 1000) // 只手动 flush由测试控制窗口
defer b.Stop()
// 模拟 5 次 recall每次都命中同 3 条记忆真实场景prefetch 高频重复命中)
for i := 0; i < 5; i++ {
b.Record([]string{"mem_a", "mem_b", "mem_c", ""})
}
// 读路径零同步写
fake.mu.Lock()
if fake.batchCalls != 0 || fake.updateCalls != 0 {
t.Fatalf("读路径不应产生任何写batch=%d update=%d", fake.batchCalls, fake.updateCalls)
}
fake.mu.Unlock()
rows, err := b.Flush()
if err != nil {
t.Fatalf("flush error: %v", err)
}
if rows != 3 {
t.Fatalf("flush rows = %d, want 33 条不同记忆)", rows)
}
fake.mu.Lock()
defer fake.mu.Unlock()
if fake.batchCalls != 1 {
t.Fatalf("batch calls = %d, want 1单事务批量", fake.batchCalls)
}
if fake.updateCalls != 0 {
t.Fatalf("不应走逐条 Update 兜底update calls=%d", fake.updateCalls)
}
if len(fake.batchItems) != 3 {
t.Fatalf("batch items = %d, want 3", len(fake.batchItems))
}
for _, it := range fake.batchItems {
if it.Delta != 5 {
t.Fatalf("id=%s delta = %d, want 5窗口内 5 次命中合并)", it.ID, it.Delta)
}
}
}
// AC-1flush 后缓冲清空 —— 空窗口不产生任何写(版本 0 增长)
func TestRecallWriteBufferEmptyFlushNoWrite(t *testing.T) {
fake := &fakeLDB{}
b := NewRecallWriteBuffer(fake, 0, 1000)
defer b.Stop()
if rows, err := b.Flush(); err != nil || rows != 0 {
t.Fatalf("空窗口 flush = (%d,%v), want (0,nil)", rows, err)
}
fake.mu.Lock()
defer fake.mu.Unlock()
if fake.batchCalls != 0 || fake.updateCalls != 0 {
t.Fatalf("空窗口不应产生写batch=%d update=%d", fake.batchCalls, fake.updateCalls)
}
}
// 阈值触发:待写条数达到 maxItems 时自动 flush不阻塞 Record 调用)
func TestRecallWriteBufferThresholdFlush(t *testing.T) {
fake := &fakeLDB{}
b := NewRecallWriteBuffer(fake, 0, 3)
defer b.Stop()
ids := []string{"m1", "m2", "m3", "m4"}
b.Record(ids)
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
fake.mu.Lock()
done := fake.batchCalls > 0
fake.mu.Unlock()
if done {
break
}
time.Sleep(20 * time.Millisecond)
}
fake.mu.Lock()
defer fake.mu.Unlock()
if fake.batchCalls == 0 {
t.Fatalf("达到阈值(%d)后应自动 flush但 batchCalls=0", b.maxItems)
}
}
// 窗口定时触发interval 到点自动落盘
func TestRecallWriteBufferTickerFlush(t *testing.T) {
fake := &fakeLDB{}
b := NewRecallWriteBuffer(fake, 100*time.Millisecond, 1000)
defer b.Stop()
b.Record([]string{"mem_tick"})
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
b.mu.Lock()
idle := len(b.pending) == 0
b.mu.Unlock()
if idle {
break
}
time.Sleep(20 * time.Millisecond)
}
fake.mu.Lock()
defer fake.mu.Unlock()
if fake.batchCalls != 1 {
t.Fatalf("窗口到点应 flush 1 次,实际 batchCalls=%d", fake.batchCalls)
}
}
// 兜底:批量提交失败时退化为逐条 Update数据不丢
func TestRecallWriteBufferFallbackToPerItem(t *testing.T) {
fake := &fakeLDB{forceBatchErr: true}
b := NewRecallWriteBuffer(fake, 0, 1000)
defer b.Stop()
b.Record([]string{"mem_x", "mem_y"})
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()
if fake.updateCalls != 2 {
t.Fatalf("兜底逐条 Update 调用数 = %d, want 2", fake.updateCalls)
}
st := b.Stats()
if st["flush_errors"].(int64) != 1 {
t.Fatalf("flush_errors = %v, want 1", st["flush_errors"])
}
}
// Stop 时落盘最后一个窗口(优雅退出不丢统计)
func TestRecallWriteBufferStopFlushes(t *testing.T) {
fake := &fakeLDB{}
b := NewRecallWriteBuffer(fake, 0, 1000)
b.Record([]string{"mem_stop"})
b.Stop()
fake.mu.Lock()
defer fake.mu.Unlock()
if fake.batchCalls != 1 {
t.Fatalf("Stop 应落盘最后一个窗口batchCalls=%d", fake.batchCalls)
}
}

View File

@ -485,7 +485,7 @@ func (sc *SQLiteClient) GetCandidatesForForgetting() ([]map[string]interface{},
defer sc.mu.RUnlock()
rows, err := sc.query(
"SELECT id, content, namespace, last_recalled_at, recall_count, tier, importance FROM memories WHERE is_deleted=0 AND tier!='core' ORDER BY last_recalled_at ASC LIMIT 100",
"SELECT id, content, namespace, category, last_recalled_at, recall_count, tier, importance, created_at FROM memories WHERE is_deleted=0 AND tier!='core' ORDER BY last_recalled_at ASC LIMIT 200",
)
if err != nil {
return nil, err
@ -496,10 +496,12 @@ func (sc *SQLiteClient) GetCandidatesForForgetting() ([]map[string]interface{},
"id": r["id"],
"content": r["content"],
"namespace": r["namespace"],
"category": r["category"],
"last_recalled_at": r["last_recalled_at"],
"recall_count": r["recall_count"],
"tier": r["tier"],
"importance": r["importance"],
"created_at": r["created_at"],
})
}
return results, nil

View File

@ -350,7 +350,135 @@ impl LanceDBOps {
Ok(updated)
}
/// 累积召回批量更新2026-09-12 优化B
///
/// 背景recall 读路径旧实现对每条结果调一次 update而 LanceDB 是 MVCC——
/// **每次 update 提交 = 一个版本**,实测 15 版本/分钟 → _versions 膨胀到 17.7G。
///
/// 做法实测约束lance 的 update 只支持基础 SQL 表达式,`CASE WHEN` 会被
/// planner 拒绝 —— 见 2026-09-12 实测 `Expression 'CASE WHEN ...' is not supported SQL in lance`
/// 把整批**按 delta 分组**,每组用一次 `recall_count + delta` 表达式 + `id IN (...)` 谓词提交。
/// 真实召回流里 delta 绝大多数是 1同一窗口内多次命中才 >1所以
/// N 条记忆的批量 ≈ **1 个版本**旧实现N 个版本)。
///
/// items_json: [{"id":"mem_xxx","delta":2}]ts: RFC3339last_recalled_at
pub fn update_recall_batch(&self, items_json: &str, ts: &str) -> Result<u64, Box<dyn std::error::Error>> {
#[derive(serde::Deserialize)]
struct Item {
id: String,
delta: i64,
}
let items: Vec<Item> = serde_json::from_str(items_json)?;
if items.is_empty() {
return Ok(0);
}
let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?;
let tbl = rt().block_on(db.open_table("memories").execute())?;
let esc = |s: &str| s.replace('\'', "''");
let ts_lit = format!("'{}'", esc(ts));
// 按 delta 分组BTreeMap 保证顺序稳定,便于日志/排障)
let mut groups: std::collections::BTreeMap<i64, Vec<String>> = std::collections::BTreeMap::new();
for it in &items {
groups.entry(it.delta).or_default().push(it.id.clone());
}
let mut total: u64 = 0;
for (delta, ids) in &groups {
let predicate = format!(
"id IN ({})",
ids.iter().map(|id| format!("'{}'", esc(id))).collect::<Vec<_>>().join(",")
);
let op = tbl
.update()
.only_if(&predicate)
.column("recall_count", &format!("recall_count + {}", delta))
.column("last_recalled_at", &ts_lit)
.column("freshness", "'verified'");
match rt().block_on(op.execute()) {
Ok(n) => {
eprintln!(
"[lancedb] BATCH UPDATE recall DONE: delta={} ids={} rows={} (单事务)",
delta,
ids.len(),
n
);
total += n;
}
Err(e) => {
// 兜底:谓词/表达式被拒时逐条更新(版本数退化为 N但数据不丢、计数正确
eprintln!(
"[lancedb] BATCH UPDATE(delta={}) failed ({}), fallback per-item ({} ids)",
delta,
e,
ids.len()
);
for id in ids {
let fields = format!(
r#"[{{"column":"recall_count","value":"recall_count + {}"}},{{"column":"last_recalled_at","value":"{}"}},{{"column":"freshness","value":"'verified'"}}]"#,
delta, ts_lit
);
match self.update("memories", id, &fields) {
Ok(n) => total += n,
Err(e2) => eprintln!("[lancedb] fallback update {} failed: {}", id, e2),
}
}
}
}
}
Ok(total)
}
/// 全量扫描 memories 表(用于深整),上限 10000 条
// P2 2026-09-06 安全版遗忘候选全表扫描(替代 febc2c9 风暴版):
// 1. 强制 limit调用方传, main.rs 钳制硬上限 5000
// 2. 不读 vector 列1024 维 × 3000 条 ≈ 12MB+IPC JSON 会卡)
// 3. 真读 last_recalled_at 列scan_all 旧实现 String::new() 恒空 → 遗忘判定失真)
pub fn scan_for_forgetting(&self, limit: usize) -> Result<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?;
let tbl = rt().block_on(db.open_table("memories").execute())?;
let mut results = Box::pin(rt().block_on(
tbl.query()
.only_if("is_deleted = false")
.limit(limit)
.execute(),
)?);
let mut records = Vec::new();
while let Some(Ok(batch)) = rt().block_on(results.next()) {
for i in 0..batch.num_rows() {
records.push(MemoryRecord {
id: col_str(&batch, i, "id"),
agent_id: col_str(&batch, i, "agent_id"),
namespace: col_str(&batch, i, "namespace"),
content: col_str(&batch, i, "content"),
category: col_str(&batch, i, "category"),
vector: Vec::new(), // ⚠️ 不读 vector — IPC 轻量, 遗忘候选不需要向量
tier: col_str(&batch, i, "tier"),
importance: col_f64(&batch, i, "importance"),
quality_score: col_f64(&batch, i, "quality_score"),
recall_count: col_i64(&batch, i, "recall_count"),
useful_count: col_i64(&batch, i, "useful_count"),
not_useful_count: col_i64(&batch, i, "not_useful_count"),
freshness: col_str(&batch, i, "freshness"),
version: col_i64(&batch, i, "version"),
version_history: String::new(),
source: col_str(&batch, i, "source"),
volatile_flag: col_bool(&batch, i, "volatile_flag"),
is_deleted: col_bool(&batch, i, "is_deleted"),
depends_on: String::new(),
derived_from: String::new(),
last_recalled_at: col_str(&batch, i, "last_recalled_at"), // ✅ 真读(勿学 scan_all 的 String::new()
created_at: col_str(&batch, i, "created_at"),
updated_at: col_str(&batch, i, "updated_at"),
});
}
}
eprintln!("[lancedb] scan_for_forgetting → {} records (limit={})", records.len(), limit);
Ok(records)
}
pub fn scan_all(&self) -> Result<Vec<MemoryRecord>, Box<dyn std::error::Error>> {
let db = rt().block_on(lancedb::connect(self.data_dir.to_str().unwrap()).execute())?;
let tbl = rt().block_on(db.open_table("memories").execute())?;
@ -400,11 +528,14 @@ impl LanceDBOps {
let mut m = 0usize; let mut e = 0usize; let mut t = 0usize;
for name in &tables {
if let Ok(tbl) = rt().block_on(db.open_table(name).execute()) {
if let Ok(cnt) = rt().block_on(tbl.count_rows(None)) {
if name == "memories" {
if let Ok(cnt) = rt().block_on(tbl.count_rows(None)) { m = cnt; }
// tombstone_count = memories 中 is_deleted=true 的行(软删审计语义;
// SoftDelete 持久化标记 is_deleted 而非写独立 tombstones 表2026-09-06 修正统计源)
if let Ok(tc) = rt().block_on(tbl.count_rows(Some("is_deleted = true".to_string()))) { t = tc; }
} else if let Ok(cnt) = rt().block_on(tbl.count_rows(None)) {
match name.as_str() {
"memories" => m = cnt,
"episodes" => e = cnt,
"tombstones" => t = cnt,
_ => {}
}
}

View File

@ -300,6 +300,21 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state
}
}
}
"lancedb_update_batch" => {
// 累积召回批量更新2026-09-12 优化B一次 update 提交 = 1 个 LanceDB 版本
let _table = msg["table"].as_str().unwrap_or("memories");
let items = msg["items"].as_str().unwrap_or("[]");
let ts = msg["ts"].as_str().unwrap_or("");
match lancedb.update_recall_batch(items, ts) {
Ok(n) => {
let cnt: usize = serde_json::from_str::<Vec<serde_json::Value>>(items)
.map(|v| v.len())
.unwrap_or(0);
send_ok(&mut stream, &format!(r#"{{"updated":{},"items":{}}}"#, n, cnt));
}
Err(e) => send_error(&mut stream, "lancedb_update_batch", &e.to_string()),
}
}
"lancedb_query" => {
let min_recall = msg["min_recall"].as_i64().unwrap_or(5) as i64;
let limit = msg["limit"].as_u64().unwrap_or(20) as usize;
@ -312,6 +327,18 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state
Err(e) => send_error(&mut stream, "lancedb_query", &e.to_string()),
}
}
"lancedb_scan" => {
// P2 2026-09-06: 遗忘候选安全全表扫描 — 限长(硬上限 5000) + 跳 vector
let mut limit = msg["limit"].as_u64().unwrap_or(2000) as usize;
if limit > 5000 {
limit = 5000;
}
eprintln!("[lancedb] scan_for_forgetting: limit={}", limit);
match lancedb.scan_for_forgetting(limit) {
Ok(r) => send_ok(&mut stream, &serde_json::to_string(&r).unwrap_or_default()),
Err(e) => send_error(&mut stream, "lancedb_scan", &e.to_string()),
}
}
_ => {
let req: ConsolidateRequest = match serde_json::from_value(msg) {
Ok(r) => r, Err(e) => {