Cyrene_v2/internal/runtime/memory/tidy_dedupe.go

package memory

import (
	"context"
	"sort"
	"strings"
	"time"
)

// 本文件是 `dedupe_group` 候选动作（2026-10-24，M34/M27）的**簇形成**与候选构造。
//
// 动因是 M27 只读体检的定量结论（`DevDocs/audit/记忆库价值体检-2026-10-24.md`）：
// active 3657 条里可复用 ≈32%，而「近似冗余」≈50%；但整理层 10-14 起**一条候选都不产**
// —— 因为它的合并判据是**包含关系**（`judgeContainment`），而问题的形态是
// 「同一件事换了个说法」：`助手名为 Cyrene。` 与 `助手/人格名为 Cyrene。` 互不包含。
//
// 🔴 **为什么不放大 `containNone`**（体检报告 §9-1 的两条路里选了后一条）：
// `containNone` 是一条**护栏**（防「冰美式 vs 红牛」这类同主题不同事实被合并），
// 放宽它等于把护栏拆掉。因此这里新增一个**更窄**的动作：它不改变任何一对的
// 包含判定，只在「簇」这个粒度上认两种**机械可判**的同一性：
//
//  1. **同 scope + 同 kind + 完全相同 tag 集**（tag 集非空，且每个 tag ≥2 字、
//     至少 2 个 tag —— 体检报告 §7 第 2 批的建议：排除 `('作息','工作')` 这类
//     过于通用的 tag 集，宁可少减也不误杀）；
//  2. **前 12 个字相同**（`dedupeGroupPrefixRunes`，与体检报告 R2 同一口径），
//     并附加**相似度下限** `dedupeGroupPrefixSimilarity` 与**同 kind** 两条收窄：
//     前者挡住「同一开头、后半句讲了别的实体」的对（`主人喜欢用「喵」…` vs
//     `主人喜欢用「汪」…` 这类同构不同事实），后者避免「合并顺手改了分类」——
//     kind 的改动仍只走 reclassify 那条路（design/23 §5.1 T0-6）。
//
// 分区（persona_id + scope + project_id）是**硬门**：候选只在一个分区内形成，
// 跨 scope 的同事实（体检报告 §7 第 3 批，83 簇/421 行）**明确不做**（会动 persona
// 写权语义，塔已拍不做）。
//
// 每簇的**安全门**与既有合并路径**逐条同源**（不新造一套判据）：
//   - 群聊/陌生人来源 ⇒ 排除（design/23 §5.2-9、§6，纵向防御同 tidy_rules）；
//   - 含绝对日期/时刻 ⇒ 排除（那些走 archive，design/23 §5.1 T0-5）；
//   - 含有时限状态词（假期/期间/本周…）⇒ 排除（§5.2-1）；
//   - 置信 < tidyLowConfidenceThreshold ⇒ 排除（§5.2-6）；
//   - 数值 token 与存活行不一致 ⇒ 该成员不入簇（§5.2-2）；
//   - 否定标记与存活行不一致 ⇒ 该成员不入簇（§5.2-3「X vs 非X 是覆盖不是重复」）。
//
// 保留规则（体检报告 §7 第 2 批的同一条）：**内容最长（超集） > merge_count 大 >
// created_at 早**，仍并列时按 id 字典序（确定性）。「最长」优先是因为簇里的冗余
// 大多是「把同一件事越写越全」，最长那条通常信息量最大；merge_count 大说明它已经
// 吸收过别的表述；两者都并列时留最早的那条（它是 first_seen，时间跨度落在它身上）。

const (
	// TidyDedupeGroup 是「机械簇去重」的动作名（design/23 §10 的 action 枚举新增值）。
	//
	// 它与 `merge` 的差别只在**判据**（簇的同一性 vs 事实集合的包含），落库形态完全
	// 相同：一簇一条候选，`source_ids` = 被吸收的行，`target_id` = 存活行，应用是
	// **软合并**（被吸收行 → status='merged' + merged_into），因此**可逐条 revert**。
	//
	// 🔴 **不在自动档里**：一次点击会改多行，体检报告 §7.2 要求在环核对（先做 100 簇、
	// 人读 20 簇再放开），因此它**只走人工/批量点**，不进 tidy_auto 的白名单。
	TidyDedupeGroup TidyAction = "dedupe_group"

	// tidyReasonDedupeGroup 是 dedupe_group 候选的理由标签（落 reason 列）。
	tidyReasonDedupeGroup = "dedupe_group"

	// dedupeGroupPrefixRunes 是「前 N 字相同」的 N（体检报告 R2 的实测口径）。
	dedupeGroupPrefixRunes = 12

	// dedupeGroupMinTagRunes / dedupeGroupMinTagCount 是 tag 集簇的门槛：
	// 每个 tag 至少 2 个字、至少 2 个 tag。单字 tag（「喵」「住」）与单个 tag
	// 太宽，簇会退化成「同主题就合并」——那正是体检报告 §7 第 2 批警告的形态。
	dedupeGroupMinTagRunes = 2
	dedupeGroupMinTagCount = 2

	// dedupeGroupPrefixSimilarity 是前缀簇成员的**附加**相似度下限（词项余弦，
	// 与召回的 lexicalSimilarity 同一个函数、同一个量纲）。
	//
	// 这条收窄是**必须的**：语料实测 871 对「前 12 字相同」的候选对里，相似度分布从
	// 0.11 到 0.99 —— 最低那批是「同开头、不同事实」（`助手（persona）的说话风格…`
	// vs `助手（persona）的象征花为蓝玫瑰…`，0.111），合并它们是**丢事实**。
	// 按阈值看语料（同一批 871 对）：
	//
	//   ≥0.60 → 40%   ≥0.70 → 14%   ≥0.75 → 6%   ≥0.80 → 2%   ≥0.85 → 1%
	//
	// 0.70 取在这里：≥0.70 的 120 对里抽 25 对人读**全是同一句话的改写**，而 0.11~0.35
	// 段全是不同事实 ⇒ 0.70 是「既能捞回改写、又不碰同开头不同事实」的那一档。
	// 它比 tidy 的召回门（0.80）低是刻意的：召回门只是「值不值得看一眼」，而这里是
	// 「要不要自动判同一簇」，两者风险不同 —— 但**两者都不是判定依据**（簇的同一性
	// 由「前 12 字」给出），这条下限只是把明显不同的对排除掉。
	dedupeGroupPrefixSimilarity = 0.70

	// dedupeGroupTagSetSimilarity 是 **tag 集簇**成员的相似度下限（r1 复审 P1 要求
	// 「tag_set 簇也要过相似度下限，不能只有 prefix12 走这道门」）。
	//
	// ⚠️ 它取 **0.10**，比前缀簇那档（0.70）低一个量级，而且**刻意如此** ——
	// 语料实测（active 3667、已过取值门/数值门/否定门的 503 对 tag 集成员对）：
	//
	//	floor 0.10 → 保留 98%   floor 0.20 → 86%   floor 0.30 → 63%
	//	floor 0.40 → 40%        floor 0.50 → 20%
	//
	// 而**挡住的那 60% 全是同一事实的改写**，不是「不同事实」——例如
	// `用户可能需要提醒吃早饭，容易空着肚子出门。`（0.047）对
	// `早餐习惯：吃半碗炒面，需搭配牛奶或鸡蛋…`，或 `叶酱有工作，白天要上班…`（0.109）
	// 对 `主人是上班族，工作日上班、单位有食堂…`。原因是语料里同一事实的多种写法
	// 共享的是**语义**而不是二元组，词项余弦天然偏低。
	//
	// ⇒ 这道门的定位被数据改了：它只挡「正文几乎没有任何共同词项」的退化对
	// （同 tag 却各说各话），**不是**同一性判据。同一性由三件套决定：数值门、
	// 否定门、（P1 新增的）**取值门**。抬高这道门只会删掉合法去重、并不会多挡住
	// 危险对 —— 需要更准的判据时该走 T1 模型层（不在本切片）。
	dedupeGroupTagSetSimilarity = 0.10
)

// dedupe_group 的证据字段（进 TidyEvidence，供面板 diff 预览与运维判读）。
const (
	// tidyDedupeRuleTagSet / tidyDedupeRulePrefix 是簇的判据名（可审计）。
	tidyDedupeRuleTagSet = "tag_set"
	tidyDedupeRulePrefix = "prefix12"

	// 保留规则名（evidence.kept_reason）：存活行为什么是它。
	tidyDedupeKeepLongest    = "longest_content"
	tidyDedupeKeepMergeCount = "merge_count"
	tidyDedupeKeepOldest     = "oldest_created_at"
	tidyDedupeKeepID         = "id_order"
)

// tidyDedupeKey 是簇键：判据 + 键值。两者一起进 evidence，便于复核「为什么它们同簇」。
type tidyDedupeKey struct {
	rule string
	key  string
}

// tidyDedupeCluster 是一个待落候选的簇：存活行 + 被吸收行。
type tidyDedupeCluster struct {
	Rule     string
	Key      string
	Survivor MemoryItem
	Absorbed []MemoryItem
	// KeptReason 是存活行胜出的那条规则（evidence.kept_reason 的来源）。
	KeptReason string
}

// tidyDedupeStats 是分区内 dedupe 分簇的观测口径（进 TidyStats）。
type tidyDedupeStats struct {
	// Groups 是**真正产出候选**的簇数。
	Groups int
	// Rows 是这些簇**可减少的 active 行数**（被吸收行合计）——「预计减少行数」。
	Rows int
}

// dedupeInPartition 是 runPartition 调用点：读 merge_count、把**已被既有候选覆盖**的
// 条目屏蔽掉，再分簇产候选。
//
// 屏蔽（covered）是「不产生第二条待办」的落点：一对已经拿到 archive/merge 候选的
// 条目不再进簇。它是纯粹的减法（覆盖集合由前一步的**产物**给出），因此不是第二套判据。
func (t *Tidier) dedupeInPartition(ctx context.Context, p tidyPartition, items []MemoryItem,
	existing []TidyCandidate) ([]TidyCandidate, tidyDedupeStats) {
	covered := make(map[string]struct{}, len(existing)*2)
	for _, c := range existing {
		for _, id := range c.SourceIDs {
			covered[id] = struct{}{}
		}
		if c.TargetID != "" {
			covered[c.TargetID] = struct{}{}
		}
	}
	// 元信息（merge_count + `expires_at` 是否非空）**不是可降级的旁支**：它承载的是
	// 「临时层行绝不进簇」这条数据安全红线（临时行 7 天后会被后台物理删除，
	// design/23 §19；让长期事实并进一个将被删除的存活行 = 信息不可达）。
	//
	// 因此查询失败时**这个分区这一轮不分簇**（fail-safe 的方向是「少做」而不是
	// 「猜」）：成本只是下一轮再来，而猜错的代价是不可逆的。
	// 这是 r1 复审 P2-1 的落点 —— 旧实现降级成 nil map 后 `temporary` 恒为 false，
	// 恰好把那道红线绕开了。
	meta, err := t.deps.Store.dedupeItemMeta(ctx, p)
	if err != nil {
		t.log.Warn("memory tidy dedupe skipped: item meta unavailable",
			"partition", p.label(), "error", err)
		return nil, tidyDedupeStats{}
	}
	return t.dedupeCandidates(items, meta, covered, t.now().UTC())
}

// dedupeCandidates 在**一个分区内**把 items 分成机械簇并产候选。它是纯函数式的
// （只读入参、无 IO、无模型），因此可完全单测，也在试算轮与真跑轮**共用**
// （两条路径都从 runPartition 进来，见 tidy.go）。
func (t *Tidier) dedupeCandidates(items []MemoryItem, meta map[string]tidyDedupeMeta,
	covered map[string]struct{}, now time.Time) ([]TidyCandidate, tidyDedupeStats) {
	clusters := dedupeClusters(items, meta, covered)
	if len(clusters) == 0 {
		return nil, tidyDedupeStats{}
	}
	out := make([]TidyCandidate, 0, len(clusters))
	stats := tidyDedupeStats{}
	for _, cl := range clusters {
		out = append(out, dedupeGroupCandidate(cl, now))
		stats.Groups++
		stats.Rows += len(cl.Absorbed)
	}
	return out, stats
}

// dedupeClusters 把条目分组、按确定性顺序认领成员、并为每簇选出存活行。
//
// 「每一条只进一个簇」是靠**全局认领集合**保证的（大簇优先、键字典序次之）：
// 一条记忆同时出现在两个簇里会让两次应用互相打脸（第二次的 `AND status='active'`
// 条件更新如实跳过，但面板会看到一条永远 skip 的候选）。
func dedupeClusters(items []MemoryItem, meta map[string]tidyDedupeMeta, covered map[string]struct{}) []tidyDedupeCluster {
	groups := dedupeGroupItems(items, meta, covered)
	ordered := dedupeOrderedGroups(groups)
	return claimDedupeClusters(items, meta, ordered)
}

// dedupeGroupItems 把可进簇的条目按簇键归组（键 → 成员下标）。
func dedupeGroupItems(items []MemoryItem, meta map[string]tidyDedupeMeta, covered map[string]struct{}) map[string]*tidyDedupeGroup {
	groups := make(map[string]*tidyDedupeGroup)
	for idx := range items {
		if _, taken := covered[items[idx].ID]; taken {
			continue
		}
		if meta[items[idx].ID].temporary {
			// 临时层行不许进簇（见 tidyDedupeMeta.temporary 的说明）。
			continue
		}
		if !dedupeGroupEligible(items[idx]) {
			continue
		}
		for _, key := range dedupeClusterKeys(items[idx]) {
			g := groups[key.key]
			if g == nil {
				g = &tidyDedupeGroup{rule: key.rule, key: key.key}
				groups[key.key] = g
			}
			g.members = append(g.members, idx)
		}
	}
	return groups
}

// dedupeOrderedGroups 只留下真正的簇（≥2 条）并按**确定性顺序**排好：大簇优先
// （可减行数最多的先落），同大小按 (key, rule) 字典序 —— 与 map 的随机遍历顺序无关，
// 因此同一批数据反复运行产出同一批候选（幂等键随之稳定）。
func dedupeOrderedGroups(groups map[string]*tidyDedupeGroup) []*tidyDedupeGroup {
	ordered := make([]*tidyDedupeGroup, 0, len(groups))
	for _, g := range groups {
		if len(g.members) < 2 {
			continue
		}
		ordered = append(ordered, g)
	}
	sort.Slice(ordered, func(i, j int) bool {
		if len(ordered[i].members) != len(ordered[j].members) {
			return len(ordered[i].members) > len(ordered[j].members)
		}
		if ordered[i].key != ordered[j].key {
			return ordered[i].key < ordered[j].key
		}
		return ordered[i].rule < ordered[j].rule
	})
	return ordered
}

// claimDedupeClusters 逐簇认领成员并产簇：**每一条只进一个簇**（靠全局认领集合），
// 存活行与全部被吸收行一起认领 —— 存活行绝不能再当别簇的被吸收行。
func claimDedupeClusters(items []MemoryItem, meta map[string]tidyDedupeMeta,
	ordered []*tidyDedupeGroup) []tidyDedupeCluster {
	claimed := make(map[int]struct{}, len(items))
	out := make([]tidyDedupeCluster, 0, len(ordered))
	for _, g := range ordered {
		free := make([]tidyDedupeMember, 0, len(g.members))
		for _, idx := range g.members {
			if _, taken := claimed[idx]; taken {
				continue
			}
			free = append(free, tidyDedupeMember{item: items[idx], mergeCount: meta[items[idx].ID].mergeCount})
		}
		if len(free) < 2 {
			continue
		}
		survivor, keptReason := pickDedupeSurvivor(free)
		absorbed := dedupeAbsorbable(g.rule, survivor, free)
		if len(absorbed) == 0 {
			continue
		}
		keep := make(map[string]struct{}, len(absorbed)+1)
		keep[survivor.item.ID] = struct{}{}
		for _, it := range absorbed {
			keep[it.item.ID] = struct{}{}
		}
		for _, idx := range g.members {
			if _, ok := keep[items[idx].ID]; ok {
				claimed[idx] = struct{}{}
			}
		}
		out = append(out, tidyDedupeCluster{
			Rule: g.rule, Key: g.key, Survivor: survivor.item, Absorbed: dedupeItems(absorbed),
			KeptReason: keptReason,
		})
	}
	return out
}

// tidyDedupeGroup 是内部用的分组累加器（键 → 成员下标）。
type tidyDedupeGroup struct {
	rule    string
	key     string
	members []int
}

// dedupeGroupEligible 报告一条记忆能不能进机械簇。它是既有硬门/否决规则的**并集**，
// 因此 dedupe_group 的判定面严格窄于 `judgePair`（后者还要求包含关系）。
func dedupeGroupEligible(it MemoryItem) bool {
	if strings.TrimSpace(it.ID) == "" {
		// 没有 id 的行不可能被候选引用（候选靠 source_ids / target_id 指向真实行），
		// 把它放进簇只会产出一条永远 skip 的候选。纵深防御：这条在落库数据上不会出现。
		return false
	}
	if nonOwnerSource(it.SourceConversation) {
		// 群聊/陌生人来源只读不写（design/23 §5.2-9、§6）：连簇都不进。
		return false
	}
	if hasAbsoluteTimestamp(it.Content) {
		// 一次性事件走 archive（§5.1 T0-5），不进合并。
		return false
	}
	if hasTemporalState(it.Content) {
		// 有期限状态（假期/期间/本周…）不是长期事实（§5.2-1）。
		return false
	}
	if it.Confidence < tidyLowConfidenceThreshold {
		// 低置信是**否决门**（§5.2-6）：低置信条目的孪生被它挡住了 2479 对，
		// 那正是体检报告 §6.2 的最大否决项。dedupe_group 不越过这道门。
		return false
	}
	if len([]rune(it.Content)) < minTidyContentRunes {
		return false
	}
	return true
}

// tidyDedupeMember 是分簇用的条目视图：记忆行 + merge_count。
//
// 为什么要单独带 merge_count：`MemoryItem` 的读侧 SELECT（store.go 的 13 列）**不含**
// `merge_count`（它只写不读），而「存活行选谁」的次要判据正是它（合并次数多的那条
// 已经吸收过别的表述，留它信息量最大）。为一个次要判据去改 13 列读面会牵动
// Get/List*/preinsert/tidy 四处 SQL，风险远大于收益；改为按分区一次轻量查询把
// id → merge_count 取出来（见 Store.dedupeMergeCounts）。
type tidyDedupeMember struct {
	item       MemoryItem
	mergeCount int
}

// dedupeItems 把成员视图降回 MemoryItem（簇只对外暴露记忆行）。
func dedupeItems(members []tidyDedupeMember) []MemoryItem {
	out := make([]MemoryItem, 0, len(members))
	for _, m := range members {
		out = append(out, m.item)
	}
	return out
}

// pickDedupeSurvivor 按「内容最长 > merge_count 大 > created_at 早 > id 字典序」选存活行，
// 并返回命中的那条规则名（进 evidence.kept_reason，回答「为什么留下的是它」）。
func pickDedupeSurvivor(members []tidyDedupeMember) (tidyDedupeMember, string) {
	best := members[0]
	for _, it := range members[1:] {
		switch compareDedupePriority(it, best) {
		case -1:
			best = it
		case 0:
			if it.item.ID < best.item.ID {
				best = it
			}
		}
	}
	return best, dedupeKeepReason(members, best)
}

// compareDedupePriority 比较两条记忆谁更该存活：-1 表示 a 优先，1 表示 b 优先，
// 0 表示按 id 决胜（调用方处理）。
func compareDedupePriority(a, b tidyDedupeMember) int {
	if la, lb := len([]rune(a.item.Content)), len([]rune(b.item.Content)); la != lb {
		if la > lb {
			return -1
		}
		return 1
	}
	if a.mergeCount != b.mergeCount {
		if a.mergeCount > b.mergeCount {
			return -1
		}
		return 1
	}
	if !a.item.CreatedAt.Equal(b.item.CreatedAt) {
		if a.item.CreatedAt.Before(b.item.CreatedAt) {
			return -1
		}
		return 1
	}
	return 0
}

// dedupeKeepReason 报告存活行是靠哪一条胜出的（证据面）。三条判据依次判「是否唯一
// 胜出」，都并列时落到最早创建（created_at）那条 —— 与 compareDedupePriority 同序。
func dedupeKeepReason(members []tidyDedupeMember, survivor tidyDedupeMember) string {
	longest := len([]rune(survivor.item.Content))
	uniqueLongest := true
	uniqueMergeCount := true
	for _, it := range members {
		if it.item.ID == survivor.item.ID {
			continue
		}
		if len([]rune(it.item.Content)) == longest {
			uniqueLongest = false
		}
		if it.mergeCount == survivor.mergeCount {
			uniqueMergeCount = false
		}
	}
	switch {
	case uniqueLongest:
		return tidyDedupeKeepLongest
	case uniqueMergeCount:
		return tidyDedupeKeepMergeCount
	case survivor.item.CreatedAt.Before(earliestOtherCreatedAt(members, survivor)):
		return tidyDedupeKeepOldest
	default:
		return tidyDedupeKeepID
	}
}

// earliestOtherCreatedAt 返回除存活行以外最早创建的时刻（用于区分「最早胜出」与
// 「id 决胜」这两种证据）。空簇成员只在单条簇里出现，那时调用方不会问证据。
func earliestOtherCreatedAt(members []tidyDedupeMember, survivor tidyDedupeMember) time.Time {
	var earliest time.Time
	for _, it := range members {
		if it.item.ID == survivor.item.ID {
			continue
		}
		if earliest.IsZero() || it.item.CreatedAt.Before(earliest) {
			earliest = it.item.CreatedAt
		}
	}
	return earliest
}