fix(forget-loop): auto_forget 闭环修复 — GetCandidatesForForgetting 改用全表扫描 + decay 触发条件对齐 + tombstone 持久化

根因（记忆只增不减）： 1. RustLanceDBClient.GetCandidatesForForgetting 只遍历进程内 _local.memories 缓存，该缓存仅被 commit/recall/search 触碰；zhiyid 启动（LastFiredAt 零值）即触发一次 decay，此时缓存近乎为空 → 0 候选 → Forgetter/ShouldForget 从不执行，auto_forget 0 条。 2. server.go decay 把空 last_recalled_at 默认当 30 天前，低于 ShouldForget ~53 天阈值 → 从未被召回的旧记忆永不遗忘；且 recall_count 仅断言 .(int)，JSON 解码 float64 会静默归 0。 3. SoftDelete 只标记内存缓存不落盘（重启即复活），tombstone 从未持久化。修复： - Rust sidecar: main.rs 暴露 lancedb_scan（scan_all 全量 is_deleted=false）；lancedb_ops 3 处读表补 last_recalled_at（此前 search/query/scan_all 返回空串）。 - Go GetCandidatesForForgetting 主路径走 lancedb_scan（sidecar 旧版自动退回缓存），按最久未访问升序取前 200（镜像 SQLite ORDER BY last_recalled_at ASC LIMIT 100，避免 merge/conflict 消费方 O(n²) 风暴）；字段强转 Go 类型（recall_count=int）。 - decay 触发条件：last_recalled_at 空 → 回退 created_at 作为空闲锚；两者皆空（未知年龄）默认保留不猜删；新增 [decay] scanned/forgotten 日志便于观测。 - admin /forget 同锚对齐（原 parseTimeStr("")=零值时间 → ShouldForget 恒真，全表从未召回记忆会被误删的隐患）。 - SoftDelete 持久化 is_deleted=true 到 LanceDB（Rust update 实测把 'true' 写为 bool）。部署验证（2026-09-05 23:08）： - 新 sidecar + 新 zhiyid 重启后 [decay] scanned=200 forgotten=1（构造 2026-06-01 旧记忆被 auto_forget 删除）；重启后 tombstone 不复活。 - 真实语料最老 2026-08-02（34 天）< 53 天阈值 → forgotten=0，无误删。 - 附带修复 merge 去重：同源重复蒸馏事实 merged_into 持久化（此前同样看不见全表）。 - 备份 /home/muc/bin/{zhiyid-new,zhiyi-consolidate}.bak-20260905 可回滚。
2026-09-05 23:11:06 +08:00
parent 506f76bd00
commit febc2c9139
5 changed files with 193 additions and 48 deletions
@@ -68,6 +68,15 @@ func (aa *AdminAPI) Forget(w http.ResponseWriter, r *http.Request) {
 	forgotten := 0
 	for _, mem := range results {
 		lastAccess := parseTimeStr(mem["last_recalled_at"])
+		if lastAccess.IsZero() {
+			// 触发条件对齐（2026-09-05）：空 last_recalled_at 回退 created_at
+			lastAccess = parseTimeStr(mem["created_at"])
+		}
+		if lastAccess.IsZero() {
+			// 未知年龄（两锚皆空）：不猜删，保留（旧逻辑 parseTimeStr("") = 零值
+			// 时间 → ShouldForget 恒真 → 全表从未召回记忆会被误删）
+			continue
+		}
 		recallCnt := intVal(mem["recall_count"])
 		tier := strVal(mem["tier"])
 		content := strVal(mem["content"])
@@ -1230,42 +1230,57 @@ func NewServer() http.Handler {
 				case "prune":
 					graphStore.Prune(0.15)
 				case "decay":
-					// 扫描所有记忆，按衰减率计算 freshness
+					// 扫描遗忘候选，按 ShouldForget 判定软删除（auto_forget）
 					memories, e := ldb.GetCandidatesForForgetting()
 					if e == nil {
+						scanned, forgotten := 0, 0
 						for _, m := range memories {
-							// 解析 last_recalled_at
-							lastAccessed := time.Now().Add(-30 * 24 * time.Hour) // 默认30天前
-							if t, ok := m["last_recalled_at"].(string); ok && t != "" {
-								if parsed, err := time.Parse(time.RFC3339, t); err == nil {
-									lastAccessed = parsed
-								}
+							scanned++
+							// 触发条件修复（2026-09-05）：last_recalled_at 为空时回退
+							// created_at 作为空闲锚（从未被召回 = 自创建起即空闲）；两者皆空
+							// （未知年龄）默认保留，不猜删。旧逻辑空值默认"30天前"，低于
+							// ShouldForget 的 ~53 天阈值 → 永不遗忘。
+							lastAccessed := time.Now()
+							t, _ := m["last_recalled_at"].(string)
+							if t == "" {
+								t, _ = m["created_at"].(string)
+							}
+							if parsed, err := time.Parse(time.RFC3339, t); err == nil {
+								lastAccessed = parsed
 							}
 							recallCount := 0
-							if rc, ok := m["recall_count"].(int); ok {
+							switch rc := m["recall_count"].(type) {
+							case int:
 								recallCount = rc
+							case float64:
+								recallCount = int(rc)
+							case int64:
+								recallCount = int(rc)
 							}
 							tier := "normal"
-													if ts, ok := m["tier"].(string); ok {
-														tier = ts
-													}
-													// E4.3: 获取实体图谱度（namespace 作为实体）
-													entity := ""
-													if ns, ok := m["namespace"].(string); ok {
-														entity = ns
-													}
-													degree := 0
-													if entity != "" {
-														degree = graphStore.GetEntityDegree(entity)
-													}
-													if forgetter.ShouldForget(lastAccessed, recallCount, tier, degree) {
-							if id, ok := m["id"].(string); ok {
-								forgetter.ScanAndForget(id)
-								// 实际执行软删除（持久化到 LanceDB）
-								_ = ldb.SoftDelete(id, "auto_forget")
+							if ts, ok := m["tier"].(string); ok {
+								tier = ts
+							}
+							// E4.3: 获取实体图谱度（namespace 作为实体）
+							entity := ""
+							if ns, ok := m["namespace"].(string); ok {
+								entity = ns
+							}
+							degree := 0
+							if entity != "" {
+								degree = graphStore.GetEntityDegree(entity)
+							}
+							if forgetter.ShouldForget(lastAccessed, recallCount, tier, degree) {
+								if id, ok := m["id"].(string); ok {
+									forgetter.ScanAndForget(id)
+									// 实际执行软删除（持久化到 LanceDB，重启不复活）
+									if err := ldb.SoftDelete(id, "auto_forget"); err == nil {
+										forgotten++
+									}
+								}
 							}
 						}
-						}
+						log.Printf("[decay] scanned=%d forgotten=%d", scanned, forgotten)
 					}
 				case "gap_scan":
 					// 检查已有缺口：过期 7 天的自动关闭
@@ -9,6 +9,7 @@ import (
 	"log"
 	"math"
 	"net"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -54,6 +55,13 @@ type ipcResp struct {
 	ErrorDetail string `json:"error_detail"`
 }

+// updateField lancedb_update 的单列更新项：Value 为 SQL 表达式
+// （字符串需用单引号包裹并转义，布尔用 true/false 字面量，数值裸写）
+type updateField struct {
+	Column string `json:"column"`
+	Value  string `json:"value"`
+}
+
 func (rc *RustLanceDBClient) rpc(req ipcReq) (*ipcResp, error) {
 	conn, err := net.DialTimeout("unix", rc.sockPath, 5*time.Second)
 	if err != nil {
@@ -398,10 +406,6 @@ func (rc *RustLanceDBClient) Update(table, id string, fields map[string]any) err
 	}

 	// 2. 通过 IPC 持久化到 Rust LanceDB（仍在锁内，防止并发写同一行）
-	type updateField struct {
-		Column string `json:"column"`
-		Value  string `json:"value"`
-	}
 	var fieldList []updateField
 	for k, v := range fields {
 		switch val := v.(type) {
@@ -459,37 +463,140 @@ func (rc *RustLanceDBClient) GetTopByQuality(agentID string, limit int) ([]model
 	return out, nil
 }

+// SoftDelete 软删除（tombstone 语义）：
+//  1. 标记内存缓存 _local.memories（保持原有行为，搜索立即跳过）
+//  2. 通过 IPC lancedb_update 持久化 is_deleted=true 到 Rust LanceDB memories 表。
+//     Rust 侧 search/query/scan_all 均带 is_deleted = false 过滤，重启后删除不失效。
+//
+// 缓存标记与 IPC 之间不加锁（无 read-modify-write，不存在 stale-read 竞态），
+// 避免 sidecar 失联时 5s dial 超时阻塞整个 _local.mu（recall/搜索热路径）。
 func (rc *RustLanceDBClient) SoftDelete(id, reason string) error {
 	_local.mu.Lock()
-	defer _local.mu.Unlock()
-	if m, ok := _local.memories[id]; ok { m.IsDeleted = true }
+	if m, ok := _local.memories[id]; ok {
+		m.IsDeleted = true
+	}
+	_local.mu.Unlock()
+
+	// 持久化 tombstone 到 Rust LanceDB（值按 SQL 表达式传递：true=布尔字面量，'...'=字符串字面量）
+	fj, _ := json.Marshal([]updateField{
+		{Column: "is_deleted", Value: "true"},
+		{Column: "updated_at", Value: "'" + time.Now().Format(time.RFC3339) + "'"},
+	})
+	if _, err := rc.rpc(ipcReq{Type: "lancedb_update", Table: "memories", ID: id, Fields: string(fj)}); err != nil {
+		log.Printf("[ipc] SoftDelete persist FAILED: id=%s reason=%s error=%v", id, reason, err)
+		return fmt.Errorf("soft delete persist (id=%s): %w", id, err)
+	}
+	log.Printf("[ipc] SoftDelete persisted: id=%s reason=%s", id, reason)
 	return nil
 }

-// TODO: 需要 Rust 侧支持 lancedb_update 才能持久化 SoftDelete
-// 当前仅标记内存，重启后失效。测试数据已直接通过 Python LanceDB API 清理。
-
 func (rc *RustLanceDBClient) GetVersionHistory(id string) ([]map[string]interface{}, error) { return nil, nil }
+
+// GetCandidatesForForgetting 返回遗忘候选（auto_forget / admin forget / merge 共用）。
+//
+// 数据源修复（2026-09-05，记忆只增不减根因）：主路径改为 Rust 全表扫描
+// lancedb_scan（is_deleted=false）。此前实现只遍历进程内 _local.memories 缓存，
+// 而该缓存仅被 commit/recall/search 触碰——zhiyid 刚启动（LastFiredAt 零值）时
+// decay 立即触发一次，此时缓存近乎为空 → 0 候选 → Forgetter/ShouldForget 从未
+// 真正执行，auto_forget 长期 0 条。镜像 SQLiteClient 参考实现（ORDER BY
+// last_recalled_at ASC LIMIT 100）：按"最久未访问"升序取前 200 条，既保证每次
+// decay 扫到最该遗忘的一批，也避免 merge/conflict 消费方拿到全表后 O(n²)。
 func (rc *RustLanceDBClient) GetCandidatesForForgetting() ([]map[string]interface{}, error) {
+	if resp, err := rc.rpc(ipcReq{Type: "lancedb_scan"}); err == nil && resp.Status == "ok" && resp.ReportJSON != "" {
+		var raw []map[string]interface{}
+		if err := json.Unmarshal([]byte(resp.ReportJSON), &raw); err == nil {
+			if cands := forgettingCandidatesFromRaw(raw); len(cands) > 0 {
+				return cands, nil
+			}
+		}
+	}
+
+	// 兜底：sidecar 不支持 lancedb_scan（旧版）时退回进程内缓存
 	_local.mu.RLock()
 	defer _local.mu.RUnlock()
 	var out []map[string]interface{}
 	for id, m := range _local.memories {
-		if !m.IsDeleted && m.Tier != "core" {
-			out = append(out, map[string]interface{}{
-				"id":               id,
-				"content":          m.Content,
-				"namespace":        m.Namespace,
-				"last_recalled_at": m.LastRecalledAt,
-				"recall_count":     m.RecallCount,
-				"tier":             m.Tier,
-				"importance":       m.Importance,
-			})
+		if m.IsDeleted || m.Tier == "core" {
+			continue
 		}
+		cand := map[string]interface{}{
+			"id":               id,
+			"content":          m.Content,
+			"namespace":        m.Namespace,
+			"category":         m.Category,
+			"recall_count":     m.RecallCount,
+			"tier":             m.Tier,
+			"importance":       m.Importance,
+			"last_recalled_at": "",
+			"created_at":       "",
+		}
+		if !m.LastRecalledAt.IsZero() {
+			cand["last_recalled_at"] = m.LastRecalledAt.Format(time.RFC3339)
+		}
+		if !m.CreatedAt.IsZero() {
+			cand["created_at"] = m.CreatedAt.Format(time.RFC3339)
+		}
+		out = append(out, cand)
 	}
 	return out, nil
 }

+// forgettingCandidatesFromRaw 把 lancedb_scan 全表记录转成遗忘候选。
+// - 过滤 is_deleted / tier=core；
+// - 字段类型转成调用方期望的 Go 类型：recall_count 必须是 int（JSON 解码是
+//   float64，server.go decay 的 .(int) 断言会静默失败 → 所有记忆都被当成
+//   0 次召回）；
+// - 排序锚 = last_recalled_at，为空则用 created_at；两者皆空（未知年龄）排
+//   最后——decay/admin 对未知年龄默认保留，不猜删。
+func forgettingCandidatesFromRaw(raw []map[string]interface{}) []map[string]interface{} {
+	const limit = 200
+	type cand struct {
+		m      map[string]interface{}
+		anchor string
+	}
+	list := make([]cand, 0, len(raw))
+	for _, r := range raw {
+		if mapBool(r, "is_deleted") || mapStr(r, "tier") == "core" {
+			continue
+		}
+		lr := mapStr(r, "last_recalled_at")
+		ca := mapStr(r, "created_at")
+		anchor := lr
+		if anchor == "" {
+			anchor = ca
+		}
+		list = append(list, cand{m: map[string]interface{}{
+			"id":               mapStr(r, "id"),
+			"content":          mapStr(r, "content"),
+			"namespace":        mapStr(r, "namespace"),
+			"category":         mapStr(r, "category"),
+			"last_recalled_at": lr,
+			"created_at":       ca,
+			"recall_count":     mapInt(r, "recall_count"),
+			"tier":             mapStr(r, "tier"),
+			"importance":       mapFloat(r, "importance"),
+		}, anchor: anchor})
+	}
+	sort.SliceStable(list, func(i, j int) bool {
+		ai, aj := list[i].anchor, list[j].anchor
+		if ai == "" {
+			return false
+		}
+		if aj == "" {
+			return true
+		}
+		return ai < aj
+	})
+	if len(list) > limit {
+		list = list[:limit]
+	}
+	out := make([]map[string]interface{}, 0, len(list))
+	for _, c := range list {
+		out = append(out, c.m)
+	}
+	return out
+}
+
 func mapStr(m map[string]interface{}, key string) string {
 	if v, ok := m[key]; ok {
 		if s, ok := v.(string); ok {
@@ -121,7 +121,7 @@ impl LanceDBOps {
                    is_deleted: col_bool(&batch, i, "is_deleted"),
                    depends_on: String::new(),
                    derived_from: String::new(),
-                    last_recalled_at: String::new(),
+                    last_recalled_at: col_str(&batch, i, "last_recalled_at"),
                    created_at: col_str(&batch, i, "created_at"),
                    updated_at: col_str(&batch, i, "updated_at"),
                });
@@ -168,7 +168,7 @@ impl LanceDBOps {
                    is_deleted: col_bool(&batch, i, "is_deleted"),
                    depends_on: String::new(),
                    derived_from: String::new(),
-                    last_recalled_at: String::new(),
+                    last_recalled_at: col_str(&batch, i, "last_recalled_at"),
                    created_at: col_str(&batch, i, "created_at"),
                    updated_at: col_str(&batch, i, "updated_at"),
                });
@@ -384,7 +384,7 @@ impl LanceDBOps {
                    is_deleted: col_bool(&batch, i, "is_deleted"),
                    depends_on: String::new(),
                    derived_from: String::new(),
-                    last_recalled_at: String::new(),
+                    last_recalled_at: col_str(&batch, i, "last_recalled_at"),
                    created_at: col_str(&batch, i, "created_at"),
                    updated_at: col_str(&batch, i, "updated_at"),
                });
@@ -312,6 +312,20 @@ fn handle_client(mut stream: UnixStream, args: &Args, lancedb: LanceDBOps, state
                Err(e) => send_error(&mut stream, "lancedb_query", &e.to_string()),
            }
        }
+        "lancedb_scan" => {
+            // 全量扫描 memories（is_deleted=false），供 decay 遗忘候选 / consolidation 全量步骤使用
+            // 向量列对 Go 侧无用且 2934×1024 会让 IPC JSON 膨胀到 ~30MB，序列化前清空
+            match lancedb.scan_all() {
+                Ok(mut r) => {
+                    for rec in r.iter_mut() {
+                        rec.vector.clear();
+                    }
+                    eprintln!("[lancedb] scan_all returned {} records", r.len());
+                    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) => {