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 完成

go/internal/api/routes/consolidation_pipe.go

@@ -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

go/internal/api/server.go

@@ -1245,9 +1245,23 @@ 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++
 							// 解析 last_recalled_at
 							lastAccessed := time.Now().Add(-30 * 24 * time.Hour) // 默认30天前
 							if t, ok := m["last_recalled_at"].(string); ok && t != "" {
@@ -1277,9 +1291,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 +1441,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) {

go/internal/governance/graph_sqlite.go

@@ -941,21 +941,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
 	}

go/internal/storage/embedder.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"os"
 	"sync"
+	"time"
 )
 
 // Embedder bge-m3 编码客户端。优先使用本地 vLLM（端口 8000），fallback 到模力方舟 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,
 	}
 }

go/internal/storage/lancedb_ipc.go

@@ -497,7 +497,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 +509,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 {

go/internal/storage/sqlite.go

@@ -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