Cyrene_v2/internal/runtime/memory/tidy.go

package memory

import (
	"context"
	"errors"
	"fmt"
	"sort"
	"strings"
)

// 本文件是记忆整理（design/23-记忆整理与去重）的整理器与编排：逐分区扫描 →
// 候选召回 → 包含判定 → 规则否决 → 候选落库 + 运行留痕。
//
// 三条不可走样的口径（design/23 §0/§5.1）：
//  1. **相似度只用来召回候选对**，永不直接当合并判据；
//  2. **合并的唯一条件是「事实集合的包含关系」**（A ⊆ B 才并入 B）；
//     相似但并列的事实**什么都不做**（判 keep）—— link 只有提示、没有执行语义，
//     因此 2026-10-07 起不再产出它（详见 tidy_rules.go 的 containNone 分支）；
//  3. **规则否决一票否决**（跨作用域/跨角色/跨项目/群聊来源/时效污染/
//     数值不一致/否定冲突/低置信），命中即踢出候选。
//
// 写入面是候选表 + 运行表（§7 的人工在环），外加 2026-10-07 起的**轮末自动落地**：
// 规则层已判为高置信的那一小撮（字面重复 / 事实等价 ∧ ≥0.95 / 一次性事件且不像
// 习惯）由 tidy_auto.go 直接落地，其余仍等人点。**任何路径都不改正文、不删行**。

// TidyAction 是整理建议的动作类别（design/23 §10 的 action 枚举）。
type TidyAction string

const (
	// TidyMerge 表示 A 的事实集合是 B 的子集，建议把 A 并入 B（保留 B）。
	// **高置信的一部分会由轮末自动档直接落地**（字面重复 / 事实等价 ∧ ≥0.95，
	// 见 tidy_auto.go）；其余仍等人点（design/23 §7 的人工在环）。
	TidyMerge TidyAction = "merge"
	// TidyLink 是 design/23 §1.3 给「两条相似但**互不包含**」这一判定结论起的动作名
	// （[1]/[3]「红牛 vs 冰美式」），语义上代表「只建链接/分组，绝不合并」。
	//
	// 🔴 2026-10-07 起**规则层不再产出它**：design/23 §14 把「link 关系如何存储与
	// 呈现」列为待细化 ⇒ 它没有落点、没有数据模型，产出来只会得到一条永远应用不了
	// 的待办（活库曾积压 501 条）。常量保留是**为了读历史行**：库里既有的 link 候选
	// 仍要能在面板展示、能被判定为「不可应用」（tidy_apply.go 的 applyable），
	// 也能被按动作批量拒绝（RejectTidyCandidatesByAction）。它不参与任何新判定。
	TidyLink TidyAction = "link"
	// TidyArchive 表示含绝对日期/时刻的一次性事件，建议归档而非合并
	// （design/23 §5.1 T0-5）。
	TidyArchive TidyAction = "archive"
	// TidyReclassify 表示 kind 误分类（design/23 §5.1 T0-6）：只改分类列，**不碰正文**。
	// 2026-10-07「全自动挡吧」起，理由为 kind_mismatch 且形状完整（含可回滚的
	// 改前 kind）的这类候选由轮末自动档直接落地（判据见 tidy_auto.go 的
	// tidyAutoReclassifyKind）；不满足判据的仍交人工。
	TidyReclassify TidyAction = "reclassify"
	// TidyKeep 是「拿不准 = 什么都不做」（design/23 §5.2-10）。它是合法动作，
	// 但第一版不为它落库（只进否决计数与日志），以免候选队列被噪声填满。
	TidyKeep TidyAction = "keep"
)

// TidyVetoReason 是规则否决的原因标签：进日志、进运行表的原因分布
// （design/23 §9「每轮记否决原因分布」），也是测试断言的抓手。
type TidyVetoReason string

const (
	// VetoNone 表示未命中任何否决规则。
	VetoNone TidyVetoReason = ""
	// VetoScope 是跨作用域硬门（design/23 §5.2-7）：user/persona/project 之间永不合。
	VetoScope TidyVetoReason = "scope"
	// VetoPersona 是跨角色硬门（design/23 §5.2-8）。
	VetoPersona TidyVetoReason = "persona"
	// VetoProject 是跨项目硬门（design/23 §5.2-8）。
	VetoProject TidyVetoReason = "project"
	// VetoSource 是群聊/陌生人来源（design/23 §5.2-9、§6）：只读不写。
	VetoSource TidyVetoReason = "source"
	// VetoTemporal 是时效污染（design/23 §5.2-1）：一次性事件/有期限状态
	// 不等于长期事实（[8]「9月8日晚」≠ 熬夜习惯）。
	VetoTemporal TidyVetoReason = "temporal"
	// VetoNumeric 是数值/配置不一致（design/23 §5.2-2）：数值不同即不同事实。
	VetoNumeric TidyVetoReason = "numeric"
	// VetoConflict 是否定/冲突（design/23 §5.2-3）：X vs 非X 是覆盖不是重复。
	VetoConflict TidyVetoReason = "conflict"
	// VetoLowConfidence 是低置信否决门（design/23 §5.2-6）。
	VetoLowConfidence TidyVetoReason = "low_confidence"
)

// ErrNilStore 在整理器没有收到记忆存储时返回。
var ErrNilStore = errors.New("memory: tidy needs a store")

// ErrTidyEmbedCount 在 embedder 返回的向量条数与请求条数不一致时返回：
// 缺失向量会让相似度与条目错位，宁可整体回退到词项余弦。
var ErrTidyEmbedCount = errors.New("memory/tidy: embedder returned a mismatched vector count")

// tidyCandidateSink 是**本轮的候选出口**：一个分区比完就把候选交给它。
// 生产轮是落库（persistCandidates），试算轮只收集（tidy_dryrun.go）——
// 规则引擎本身与出口无关，这正是「禁止写第二套实现」的落点。
type tidyCandidateSink func(ctx context.Context, cands []TidyCandidate, stats *TidyStats)

// tidyRecall 是**本轮的召回策略**：给定已截断的参与比对条目，产出候选对。
// 它把「相似度来源不可用时怎么办」的差异收在一处：生产轮降级到零依赖词项
// 余弦，试算轮如实上报错误。budget 是本轮的时间预算闸（见 tidy_budget.go），
// 由调用方一路传下来，使两条策略共享**同一个**预算（而不是各自从 ctx 再派生）。
type tidyRecall func(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error)

// tidyRoundPolicy 是本轮整理的出口策略：候选去哪 + 召回来源不可用时怎么办。
// 「生产轮」与「试算轮」的**全部**差异都在这两个字段上，规则引擎其余部分
// 逐字共用。
type tidyRoundPolicy struct {
	sink   tidyCandidateSink
	recall tidyRecall
}

// RunOnce 跑一轮整理，并写一条运行留痕。它返回本轮的观测计数。
//
// 失败隔离（design/23 §9）：单个分区失败不影响其它分区，失败计数进运行表与
// 日志；只有「所有分区都失败」或运行留痕本身写不进去时才回错误。
func (t *Tidier) RunOnce(ctx context.Context) (TidyStats, error) {
	started := t.now().UTC()
	// 运行 id 与轮次开始**同时**定下来：候选要按它归属（切片 TIDY-1 的按轮
	// 批量应用），运行留痕也要用它，因此不能等到扫完再生成。
	runID := mustNewTidyID()
	stats, partitions, err := t.beginRound(ctx)
	if err != nil {
		return stats, err
	}
	stats.RunID = runID
	// ledger 是**本轮实际发生**的召回来源记账口（不是配置里那一种）：生产轮可以
	// 多分区混合（有的走语义、有的降级），因此汇总行只能由实测记录推出。
	ledger := &recallLedger{}

	failures := t.scanPartitions(ctx, partitions, &stats, tidyRoundPolicy{
		sink: func(ctx context.Context, cands []TidyCandidate, stats *TidyStats) {
			t.persistCandidates(ctx, runID, cands, stats)
		},
		recall: func(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget) ([]tidyPair, error) {
			return t.fallbackRecall(ctx, p, items, stats, budget, ledger)
		},
	})

	// 轮末自动落地（2026-10-07 主人口径）：规则层判为高置信的待定候选直接落地，
	// 不再等人点。它排在留痕之前，因此 memory_tidy_runs.applied 这一轮就是真数
	// （而不是先写 0、事后靠按 run 批量应用补记）。
	auto := t.autoApply(ctx, &stats)

	run := TidyRun{
		ID:          runID,
		StartedAt:   started,
		FinishedAt:  t.now().UTC(),
		Scanned:     stats.Scanned,
		Candidates:  stats.Candidates,
		Applied:     auto.Applied,
		Skipped:     stats.Skipped + stats.Duplicates,
		Errors:      stats.Errors,
		Vetoed:      stats.Vetoed,
		VetoReasons: stats.VetoReasons,
		RuleVersion: TidyRuleVersion,
	}
	stats.DurationMS = run.FinishedAt.Sub(run.StartedAt).Milliseconds()
	if err := t.deps.Store.persistTidyRun(ctx, run); err != nil {
		failures = append(failures, err)
	}

	t.logRun(stats, run, ledger)
	t.auditAutoApply(auto)

	switch {
	case len(failures) == 0:
		return stats, nil
	case len(failures) == len(partitions) && len(partitions) > 0:
		return stats, fmt.Errorf("memory/tidy: every partition failed: %w", errors.Join(failures...))
	default:
		return stats, fmt.Errorf("memory/tidy: run finished with errors: %w", errors.Join(failures...))
	}
}

// logRun 输出本轮的可观测口径：扫描 N / 候选 M / 建议 K / 否决原因分布。
// 它带 source（memory/tidy），因此这条汇总永远可回溯到整理子系统。
//
// ledger 给出的是**本轮实际发生**的召回来源（不是配置态）。这一条是 M6 2026-10-13
// 的 medium finding 的落点：旧实现按「有没有配 embedder」写 recall_mode，全量降级
// 在汇总行里被写成 embedding。现在：
//   - recall_mode = ""：没有任何分区真正算过相似度（没有候选可比）；
//   - recall_mode = lexical ∧ recall_degraded_partitions > 0：语义不可用、降级了；
//   - recall_mode = mixed：多分区里一部分走了语义、一部分降级 —— 不再被抹平。
func (t *Tidier) logRun(stats TidyStats, run TidyRun, ledger *recallLedger) {
	fields := []any{
		"run_id", run.ID,
		"rule_version", run.RuleVersion,
		"partitions", stats.Partitions,
		"scanned", run.Scanned,
		"compared", stats.Compared,
		"pairs", stats.Pairs,
		"candidates", run.Candidates,
		"duplicates", stats.Duplicates,
		"skipped", stats.Skipped,
		"vetoed", run.Vetoed,
		"veto_reasons", tidyReasonsLabel(run.VetoReasons),
		"kept", stats.Kept,
		"errors", run.Errors,
		"duration_ms", run.FinishedAt.Sub(run.StartedAt).Milliseconds(),
		"recall_mode", ledger.label(),
		"recall_degraded_partitions", ledger.degradedPartitions(),
		"auto_apply", t.cfg.autoApplyEnabled(),
		"auto_applied", stats.AutoApplied,
		"auto_merge", stats.AutoMerged,
		"auto_archive", stats.AutoArchived,
		"auto_reclassify", stats.AutoReclassified,
		"auto_eligible", stats.AutoEligible,
		"auto_truncated", stats.AutoTruncated,
		"dedupe_groups", stats.DedupeGroups,
		"dedupe_rows", stats.DedupeRows,
		"dedupe_truncated", stats.DedupeTruncated,
	}
	if len(stats.Truncated) > 0 {
		fields = append(fields, "truncated", strings.Join(stats.Truncated, ","))
	}
	t.log.Info("memory tidy run finished", fields...)
}

// beginRound 开一轮：准备口径容器并列出待整理分区。分区列表拿不到时整轮失败，
// 且**不写运行留痕** —— 那一轮根本没开始（口径与 design §9 一致）。
func (t *Tidier) beginRound(ctx context.Context) (TidyStats, []tidyPartition, error) {
	stats := TidyStats{
		VetoReasons: make(map[string]int),
		ByAction:    make(map[string]int),
	}
	partitions, err := t.deps.Store.listTidyPartitions(ctx)
	if err != nil {
		return stats, nil, fmt.Errorf("memory/tidy: list partitions: %w", err)
	}
	stats.Partitions = len(partitions)
	return stats, partitions, nil
}

// scanPartitions 逐个分区跑规则并把候选交给 policy.sink。它返回失败列表
// （单个分区失败不阻塞其它分区，design §9）；成功时返回 nil。
//
// 时间预算（tidy_budget.go）在**轮**级派生一次：ctx 有墙钟上限（同步/异步试算）
// 时，预算用尽后的分区不再开跑，而是如实登记为 (budget) 截断。
func (t *Tidier) scanPartitions(ctx context.Context, partitions []tidyPartition, stats *TidyStats, policy tidyRoundPolicy) []error {
	budget := t.newTidyBudget(ctx)
	var failures []error
	for _, p := range partitions {
		if err := t.runPartition(ctx, p, stats, policy, budget); err != nil {
			stats.Errors++
			failures = append(failures, err)
			t.log.Warn("memory tidy partition failed", "partition", p.label(), "error", err)
		}
	}
	return failures
}

// fallbackRecall 是**生产轮**的召回：相似度来源不可用时降级到零依赖的词项
// 余弦（离线/无 embedding 配置时可跑，宁可比语义路径更保守），并记一条 WARN。
// 两个来源**不可混用阈值**：阈值只有一个（design §13 开放点 2 的 0.80）。
//
// 降级这一笔**必须记进 ledger**：汇总行/面板上的召回来源只能由实测记录推出，
// 否则「全量降级」会被报成「向量正常」（M6 2026-10-13 的 medium finding）。
func (t *Tidier) fallbackRecall(ctx context.Context, p tidyPartition, items []MemoryItem, stats *TidyStats, budget tidyBudget, ledger *recallLedger) ([]tidyPair, error) {
	pairs, err := t.recallPairs(ctx, p, items, stats, budget, ledger)
	if err == nil {
		return pairs, nil
	}
	ledger.noteLexical(true)
	t.log.Warn("memory tidy embedding recall unavailable, falling back to lexical",
		"partition", p.label(), "error", err)
	return t.collectPairs(p, items, t.cfg.RecallThreshold,
		lexicalSimFunc(items), tidyRecallLexical, stats, budget), nil
}

// runPartition 整理一个分区（persona_id + scope + project_id）。
// 分区之间**永不比较**（design/23 §6 硬约束）。候选去处与召回降级策略都由
// policy 决定，因此生产轮与试算轮走的是**同一条**扫描→召回→判定→构造候选
// 的路径。
func (t *Tidier) runPartition(ctx context.Context, p tidyPartition, stats *TidyStats, policy tidyRoundPolicy, budget tidyBudget) error {
	// 预算用尽 ⇒ 这个分区本轮根本不看。**如实登记**（响应 truncated + WARN）：
	// 「没扫到」与「扫了没有候选」必须能区分，否则面板会把预算截断读成
	// 「这个分区没有建议」。
	if budget.exhausted() {
		stats.Truncated = append(stats.Truncated, budgetTruncationLabel(p))
		t.log.Warn("memory tidy partition skipped by time budget", "partition", p.label())
		return nil
	}

	items, err := t.deps.Store.listTidyPartitionItems(ctx, p, t.cfg.ScanLimit)
	if err != nil {
		return fmt.Errorf("memory/tidy: list partition %s: %w", p.label(), err)
	}
	stats.Scanned += len(items)
	if len(items) == t.cfg.ScanLimit {
		// 扫描上限命中：本分区更大的部分这一轮不看（design §9「超限分片，下一轮
		// 继续」）。本实现只做「最新优先」的截断、不推进分片游标，因此这里必须
		// 留下痕迹，否则「配置了 5000 却永远只看最新 5000」会无人知晓。
		// 日志带**真总数**（LIMIT 之后的 len(items) 只是下限）：另查一次 COUNT，
		// 否则 WARN 会把「至少 5000」说成「正好 5000」。
		total, totalErr := t.deps.Store.countTidyPartitionItems(ctx, p)
		stats.Truncated = append(stats.Truncated, p.label()+"(scan)")
		if totalErr != nil {
			t.log.Warn("memory tidy partition hit the scan limit (total unknown)",
				"partition", p.label(), "limit", t.cfg.ScanLimit, "error", totalErr)
		} else {
			t.log.Warn("memory tidy partition hit the scan limit",
				"partition", p.label(), "limit", t.cfg.ScanLimit, "total", total)
		}
	}

	comparable, archives := t.classifyItems(items, stats)

	// 截断与计数只做一次：召回策略拿到的是**已截断**的条目，回退重算不重复计数。
	compared := t.limitCompareItems(p, comparable, stats)
	pairs, err := policy.recall(ctx, p, compared, stats, budget)
	if err != nil {
		return fmt.Errorf("memory/tidy: recall partition %s: %w", p.label(), err)
	}

	candidates := t.mergeCandidates(archives, pairs, stats)

	// dedupe_group（2026-10-24）：机械簇（同 tag 集 / 前 12 字相同）**不依赖相似度
	// 召回**，因此它跑在可比条目全集上（`comparable`）、**不受**比对截断影响 ——
	// 截断是给 O(n²) 的两两余弦用的护栏，不该让机械分簇跟着少看一半条目。
	//
	// 它排在包含判定**之后**是为了**抑制重复待办**：一对已经拿到 archive/merge
	// 候选的条目不再进簇 —— 否则同一对会同时出现「并入超集」与「并入簇存活行」
	// 两条候选，第二条应用时必然 skip（source_not_active），面板上看像 bug。
	// 覆盖面由前一步的**产物**决定（不是又一套判据），因此抑制是纯粹的减法。
	dedupe, dedupeStats := t.dedupeInPartition(ctx, p, comparable, candidates)
	stats.DedupeGroups += dedupeStats.Groups
	stats.DedupeRows += dedupeStats.Rows

	// dedupe 候选追加在既有两类之后：**不改变**归档/合并在候选上限里的优先序
	// （那两类是既有的、被验证过的口径；新动作不许把它们挤出上限）。
	candidates = append(candidates, dedupe...)
	candidates = t.trimPartitionCandidates(p, candidates, len(dedupe), stats)
	policy.sink(ctx, candidates, stats)
	return nil
}

// mergeCandidates 把包含判定的产物收敛成候选：归档建议在前，merge 在后；
// 被否决的对进否决分布，互不包含的对进 Kept（**不产候选**，2026-10-07 口径）。
func (t *Tidier) mergeCandidates(archives []TidyCandidate, pairs []tidyPair, stats *TidyStats) []TidyCandidate {
	candidates := archives
	for _, pr := range pairs {
		verdict := judgePair(pr.a, pr.b, pr.score)
		if verdict.Veto != VetoNone {
			stats.Vetoed++
			stats.VetoReasons[string(verdict.Veto)]++
			continue
		}
		if verdict.Action != TidyMerge {
			// 相似但互不包含 ⇒ keep：**不产候选**（2026-10-07，见 tidy_rules.go 的
			// containNone 分支）。它必须单独计数，否则「抛掉多少对」会消失。
			stats.Kept++
			continue
		}
		candidates = append(candidates, buildCandidates(pr, verdict, t.now().UTC())...)
	}
	return candidates
}

// trimPartitionCandidates 施加每分区候选上限（design §9），并**如实登记**被丢掉的
// 部分：总量进 stats.Skipped 与 Truncated 标签，其中属于 dedupe 段的那部分单独进
// stats.DedupeTruncated（新动作被截断时面板必须看得到，不许「静默少做」）。
//
// dedupeCount 是本批候选**尾部**属于 dedupe 的条数（调用方把 dedupe 追加在最后）。
func (t *Tidier) trimPartitionCandidates(p tidyPartition, candidates []TidyCandidate,
	dedupeCount int, stats *TidyStats) []TidyCandidate {
	limit := t.cfg.CandidateLimit
	if len(candidates) <= limit {
		return candidates
	}
	dedupeStart := len(candidates) - dedupeCount
	// 上限切在 dedupe 段之前 ⇒ 丢掉的**全部**是 dedupe 候选；切在段内 ⇒ 丢掉尾部。
	keptDedupe := 0
	if limit > dedupeStart {
		keptDedupe = limit - dedupeStart
	}
	if keptDedupe > dedupeCount {
		keptDedupe = dedupeCount
	}
	droppedDedupe := dedupeCount - keptDedupe
	stats.DedupeTruncated += droppedDedupe
	// WARN 带总数/上限/分区名：候选被上限截断意味着面板看到的**少于**规则判出的，
	// 必须能从这里读出被丢掉了多少。
	t.log.Warn("memory tidy candidates truncated by per-partition limit",
		"partition", p.label(), "candidates", len(candidates), "limit", limit,
		"dedupe_truncated", droppedDedupe)
	stats.Skipped += len(candidates) - limit
	stats.Truncated = append(stats.Truncated, p.label()+"(candidates)")
	return candidates[:limit]
}

// classifyItems 把分区内条目分流成**可比对**与**归档候选**（跳过的不产出）：
//   - 群聊/陌生人来源 → 排除在扫描范围外（design/23 §6，只读不写）；
//   - 含绝对日期/时刻 → 一次性事件，走归档而非合并（[8]「9月8日晚22:38」），
//     且**不参与任何 merge**（design §5.1-5）；
//   - 过短内容 → 没有可判定的事实要素，跳过（不产噪声候选）。
func (t *Tidier) classifyItems(items []MemoryItem, stats *TidyStats) (comparable []MemoryItem, archives []TidyCandidate) {
	comparable = make([]MemoryItem, 0, len(items))
	for _, it := range items {
		switch t.classifyForTidy(it, stats) {
		case tidySkip:
			continue
		case tidyArchive:
			archives = append(archives, archiveCandidate(it, t.now().UTC()))
			continue
		default:
			comparable = append(comparable, it)
		}
	}
	return comparable, archives
}

// tidyDisposition 是扫描时一条记忆的处置分流。
type tidyDisposition int

const (
	tidyCompare tidyDisposition = iota
	tidyArchive
	tidySkip
)

// classifyForTidy 决定一条记忆进入比对、进归档候选，还是被排除：
//   - 群聊/陌生人来源 → 排除在扫描范围外（design/23 §6，只读不写）；
//   - 含绝对日期/时刻 → 一次性事件，走归档而非合并（[8]「9月8日晚22:38」）；
//   - 过短内容 → 没有可判定的事实要素，跳过（不产噪声候选）。
func (t *Tidier) classifyForTidy(it MemoryItem, stats *TidyStats) tidyDisposition {
	if nonOwnerSource(it.SourceConversation) {
		stats.Vetoed++
		stats.VetoReasons[string(VetoSource)]++
		return tidySkip
	}
	if hasAbsoluteTimestamp(it.Content) {
		return tidyArchive
	}
	if len([]rune(it.Content)) < minTidyContentRunes {
		return tidySkip
	}
	return tidyCompare
}

// persistCandidates 落库一批候选：幂等键已存在时只计入 duplicates
// （design/23 §9「重复跑不重复产候选」），新建的计入 Candidates 并按动作进
// ByAction（回执的领域来源）。runID 是本轮的留痕 id：「按 run 批量应用」据此
// 找到这一轮产出的候选；幂等跳过的行保留它**首次**被产出的轮次，不被改写。
func (t *Tidier) persistCandidates(ctx context.Context, runID string, cands []TidyCandidate, stats *TidyStats) {
	for _, c := range cands {
		c.RunID = runID
		created, err := t.deps.Store.insertTidyCandidate(ctx, c)
		if err != nil {
			stats.Errors++
			t.log.Warn("memory tidy candidate insert failed", "action", string(c.Action), "error", err)
			continue
		}
		if !created {
			stats.Duplicates++
			continue
		}
		stats.Candidates++
		if stats.ByAction != nil {
			stats.ByAction[string(c.Action)]++
		}
	}
}

// tidyReasonsLabel 把否决原因分布渲染成稳定的日志字符串（按原因排序）。
func tidyReasonsLabel(reasons map[string]int) string {
	if len(reasons) == 0 {
		return "none"
	}
	keys := make([]string, 0, len(reasons))
	for k := range reasons {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	parts := make([]string, 0, len(keys))
	for _, k := range keys {
		parts = append(parts, fmt.Sprintf("%s=%d", k, reasons[k]))
	}
	return strings.Join(parts, ",")
}