Cyrene_v2/internal/runtime/memory/tidy_dedupe_apply_test.go

package memory

import (
	"context"
	"fmt"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
)

// 本文件是 `dedupe_group` 的**应用 / 回滚**验收用例集（软合并、整簇原子、
// 按行计的分批上限、临时层数据安全）。分簇与候选构造的用例在
// tidy_dedupe_test.go —— 分开只为守 AGENTS §10 的文件行数上限（≤500 行）。

// TestTidyApplyDedupeGroupIsSoftAndRevertible 钉住应用/回滚：整簇软合并（行不删、
// 正文不改、merge_count 累加），revert 把**每一行**都复原。
func TestTidyApplyDedupeGroupIsSoftAndRevertible(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	tags := []string{"身份", "名字"}
	items := []MemoryItem{
		seedDedupeItem(t, store, dedupeTestItem("助理的名字是 Cyrene。", tags, base)),
		seedDedupeItem(t, store, dedupeTestItem("助手名为 Cyrene。", tags, base.Add(time.Minute))),
		seedDedupeItem(t, store, dedupeTestItem("助手/人格/角色名为 Cyrene。", tags, base.Add(2*time.Minute))),
	}
	cands, _ := dedupeCandsOf(t, store, items)
	if len(cands) != 1 {
		t.Fatalf("簇数 = %d, want 1", len(cands))
	}
	cand := cands[0]
	if created, err := store.insertTidyCandidate(ctx, cand); err != nil || !created {
		t.Fatalf("insert candidate: created=%v err=%v", created, err)
	}

	res, err := store.ApplyTidyCandidate(ctx, cand.ID, TidyApplyOptions{Actor: "owner:test"})
	if err != nil {
		t.Fatalf("ApplyTidyCandidate: %v", err)
	}
	if res.Status != TidyApplyApplied || len(res.AffectedIDs) != 2 {
		t.Fatalf("应用结果 = %+v, want applied 且影响 2 行", res)
	}
	for _, id := range cand.SourceIDs {
		row := loadTidyItemRow(t, store, id)
		if row.status != ingestStatusMerged || row.mergedInto != cand.TargetID {
			t.Fatalf("被吸收行 %s = %+v, want merged → %s", id, row, cand.TargetID)
		}
	}
	survivor := loadTidyItemRow(t, store, cand.TargetID)
	if survivor.status != tidyStatusActive {
		t.Fatalf("存活行状态 = %q, want active", survivor.status)
	}
	if got := countMemoryItems(t, store); got != 3 {
		t.Fatalf("应用后行数 = %d, want 3（软合并绝不删行）", got)
	}
	if got := loadTidyItemRow(t, store, cand.TargetID).mergeCount; got != 3 {
		t.Fatalf("存活行 merge_count = %d, want 3（1 + 吸收 2）", got)
	}
	if got := loadTidyItemRow(t, store, cand.TargetID).content; got != items[2].Content {
		t.Fatalf("存活行正文被改了：%q", got)
	}

	reverted, err := store.RevertTidyCandidate(ctx, cand.ID, TidyApplyOptions{Actor: "owner:test"})
	if err != nil {
		t.Fatalf("RevertTidyCandidate: %v", err)
	}
	if reverted.Status != TidyApplyApplied || len(reverted.AffectedIDs) != 2 {
		t.Fatalf("回滚结果 = %+v, want applied 且复原 2 行", reverted)
	}
	for _, id := range cand.SourceIDs {
		row := loadTidyItemRow(t, store, id)
		if row.status != tidyStatusActive || row.mergedInto != "" {
			t.Fatalf("回滚后 %s = %+v, want active 且 merged_into 清空", id, row)
		}
	}
	if got := loadTidyItemRow(t, store, cand.TargetID).mergeCount; got != 1 {
		t.Fatalf("回滚后 merge_count = %d, want 1", got)
	}
	if got, _ := store.GetTidyCandidate(ctx, cand.ID); got.Status != tidyCandidatePending {
		t.Fatalf("回滚后候选状态 = %q, want pending（可重新决定）", got.Status)
	}
}

// TestTidyApplyDedupeGroupRollsBackWholeClusterOnStaleMember 钉住原子的**安全方向**：
// 簇里有一行已不在册 ⇒ 整簇不吞（一行都不改），候选如实 skip。
func TestTidyApplyDedupeGroupRollsBackWholeClusterOnStaleMember(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	tags := []string{"身份", "名字"}
	items := []MemoryItem{
		seedDedupeItem(t, store, dedupeTestItem("助理的名字是 Cyrene。", tags, base)),
		seedDedupeItem(t, store, dedupeTestItem("助手名为 Cyrene。", tags, base.Add(time.Minute))),
		seedDedupeItem(t, store, dedupeTestItem("助手/人格/角色名为 Cyrene。", tags, base.Add(2*time.Minute))),
	}
	cands, _ := dedupeCandsOf(t, store, items)
	cand := cands[0]
	if created, err := store.insertTidyCandidate(ctx, cand); err != nil || !created {
		t.Fatalf("insert candidate: created=%v err=%v", created, err)
	}
	// 让簇里一个成员先被别的动作合并掉（模拟过时候选）。
	if _, err := store.db.ExecContext(ctx,
		`UPDATE memory_items SET status = ? WHERE id = ?`,
		ingestStatusMerged, cand.SourceIDs[0]); err != nil {
		t.Fatalf("stale member update: %v", err)
	}
	res, err := store.ApplyTidyCandidate(ctx, cand.ID, TidyApplyOptions{Actor: "owner:test"})
	if err != nil {
		t.Fatalf("ApplyTidyCandidate: %v", err)
	}
	if res.Status != TidyApplySkipped || res.Reason != tidySkipSourceNotActive {
		t.Fatalf("结果 = %+v, want skipped/%s（整簇原子，不许吞一半）", res, tidySkipSourceNotActive)
	}
	if got := loadTidyItemRow(t, store, cand.TargetID).mergeCount; got != 1 {
		t.Fatalf("存活行 merge_count = %d, want 1（整簇回滚，一行都没改）", got)
	}
	if got := loadTidyItemRow(t, store, cand.SourceIDs[1]).status; got != tidyStatusActive {
		t.Fatalf("第二个被吸收行状态 = %q, want active（半吞禁止）", got)
	}
}

// TestTidyApplyDedupeGroupHonoursRowBudget 钉住「批量按行计、超限如实 truncated」：
// 每簇 2 个被吸收行（行预算按被吸收行计），预算 200 ⇒ 最多 100 簇（= 200 行），
// 第 101 簇起留给下一次点击，回执必须如实说 truncated。
func TestTidyApplyDedupeGroupHonoursRowBudget(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	total := maxTidyApplyRowsPerBatch/2 + 5     // 簇数多到必然撞上 200 行预算
	wantApplied := maxTidyApplyRowsPerBatch / 2 // 每簇 2 个被吸收行

	for i := 0; i < total; i++ {
		tags := []string{"身份", "代号"}
		content := fmt.Sprintf("小茂密的第 %d 号说法：蝴蝶结是小铃铛", i)
		items := []MemoryItem{
			seedDedupeItem(t, store, dedupeTestItem(content+"甲", tags, base)),
			seedDedupeItem(t, store, dedupeTestItem(content+"乙", tags, base.Add(time.Minute))),
			seedDedupeItem(t, store, dedupeTestItem(content+"丙", tags, base.Add(2*time.Minute))),
		}
		cands, _ := dedupeCandsOf(t, store, items)
		if len(cands) != 1 {
			t.Fatalf("第 %d 组簇数 = %d, want 1", i, len(cands))
		}
		if created, err := store.insertTidyCandidate(ctx, cands[0]); err != nil || !created {
			t.Fatalf("insert candidate %d: created=%v err=%v", i, created, err)
		}
	}
	batch, err := store.ApplyTidyCandidates(ctx, "", TidyApplyOptions{Actor: "owner:test", Limit: 500})
	if err != nil {
		t.Fatalf("ApplyTidyCandidates: %v", err)
	}
	if !batch.Truncated {
		t.Fatalf("truncated = false, want true（行预算 200 被撞上时必须如实说）")
	}
	if batch.Applied != wantApplied {
		t.Fatalf("应用簇数 = %d, want %d（200 行 ÷ 3 行/簇）", batch.Applied, wantApplied)
	}
	if batch.RowsAffected != wantApplied*2 {
		t.Fatalf("改动行数 = %d, want %d", batch.RowsAffected, wantApplied*2)
	}
	if batch.Deduped != wantApplied {
		t.Fatalf("deduped 计数 = %d, want %d", batch.Deduped, wantApplied)
	}
}

// TestTidyDryRunReportsDedupeGroupsWithoutWriting 钉住「**先试算、不写库**」这条要求：
// 试算轮必须给出 dedupe_group 的**簇数与可减少行数**（面板据此决定要不要落库），
// 并且三张表逐行不变（零副作用）—— 试算与真跑共用同一条规则路径，因此这份报告
// 就是「真跑会产出的东西」。
func TestTidyDryRunReportsDedupeGroupsWithoutWriting(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	// 三条**互不包含**、但 tag 集完全相同的说法（否则它们会先被包含判定吃掉，
	// 那时 dedupe 会被「已覆盖」抑制掉 —— 那是对的行为，但测不到试算口径）。
	tags := []string{"宠物", "小茂密"}
	for i, content := range []string{
		"小茂密是白毛的小猫，喜欢撒娇",
		"白毛小茂密爱撒娇，是只小猫",
		"爱撒娇的小猫名叫小茂密，通体白毛",
	} {
		seedDedupeItem(t, store, dedupeTestItem(content, tags, base.Add(time.Duration(i)*time.Minute)))
	}
	before := snapshotTidyTables(t, store)
	emb := stubEmbedder{vecs: map[string]embedding.Vector{}}

	dr, err := newTestTidier(t, store, tidyTestCfg(0.80), emb).DryRun(ctx)
	if err != nil {
		t.Fatalf("DryRun: %v", err)
	}
	if dr.Stats.DedupeGroups != 1 || dr.Stats.DedupeRows != 2 {
		t.Fatalf("试算口径 = groups:%d rows:%d, want 1 / 2（三行一簇，可减两行）",
			dr.Stats.DedupeGroups, dr.Stats.DedupeRows)
	}
	dedupeCands := candidatesByAction(dr.Candidates, TidyDedupeGroup)
	if len(dedupeCands) != 1 {
		t.Fatalf("试算候选里的 dedupe_group = %d 条, want 1", len(dedupeCands))
	}
	assertRowsUnchanged(t, "试算后三张表", before.items, snapshotTidyTables(t, store).items)
	after := snapshotTidyTables(t, store)
	assertRowsUnchanged(t, "试算后候选表", before.candidates, after.candidates)
	assertRowsUnchanged(t, "试算后运行表", before.runs, after.runs)
}

// TestDedupeGroupNeverTouchesTemporaryRows 钉住临时层的**数据安全**不变式：
// `expires_at` 非空的记忆（临时层）绝不进簇 —— 无论当存活行还是被吸收行。
//
// 为什么是红线：临时行 7 天后由后台清扫**物理删除**（design/23 §19）。若它当存活行，
// 被吸收的长期事实会永远停在 `status='merged'` 且 `merged_into` 指向一个已不存在的
// id（信息不可达）；若它当被吸收行，一次合并就把一条正在生效的临时记忆藏了起来。
func TestDedupeGroupNeverTouchesTemporaryRows(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)
	tags := []string{"身份", "名字"}
	durable := seedDedupeItem(t, store, dedupeTestItem("助理的名字是 Cyrene。", tags, base))
	temporary := seedDedupeItem(t, store, dedupeTestItem("助手名为 Cyrene。", tags, base.Add(time.Minute)))
	if _, err := store.db.ExecContext(ctx,
		`UPDATE memory_items SET expires_at = ? WHERE id = ?`,
		expiryText(base.Add(7*24*time.Hour)), temporary.ID); err != nil {
		t.Fatalf("mark temporary: %v", err)
	}
	cands, stats := dedupeCandsOf(t, store, []MemoryItem{durable, temporary})
	if len(cands) != 0 || stats.Groups != 0 {
		t.Fatalf("簇数 = %d, want 0（临时层行不许进簇：存活行会让被吸收行指向一个将被删除的 id）",
			len(cands))
	}

	// 三条时：两条长期 + 一条临时 ⇒ 正是两条长期的那一簇，临时行不在候选里。
	durable2 := seedDedupeItem(t, store, dedupeTestItem("助理/助手名为 Cyrene。", tags, base.Add(2*time.Minute)))
	cands, _ = dedupeCandsOf(t, store, []MemoryItem{durable, temporary, durable2})
	if len(cands) != 1 {
		t.Fatalf("簇数 = %d, want 1", len(cands))
	}
	for _, id := range cands[0].SourceIDs {
		if id == temporary.ID {
			t.Fatalf("临时行出现在被吸收行里：%v", cands[0].SourceIDs)
		}
	}
	if cands[0].TargetID == temporary.ID {
		t.Fatalf("临时行当上了存活行 —— 它 7 天后会被物理删除")
	}
}

// TestTidyRunReportsDedupeTruncation 钉住「truncated 如实」：每分区候选上限切掉了
// 多少 dedupe 簇必须**单独**报出来。变异（不给 DedupeTruncated 计数、或按错误的
// 段计算）会让这条用例红 —— 被截断的簇在面板上会彻底消失，那正是「静默少做」。
func TestTidyRunReportsDedupeTruncation(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	store := newTestStore(t)
	base := tidyTestTime(t)

	// 两个互不相干的簇（tag 集不同 ⇒ 两个簇），一共 2 条 dedupe 候选。
	seed := []struct {
		content string
		tags    []string
	}{
		{"小茂密是白毛的小猫，喜欢撒娇", []string{"宠物", "小茂密"}},
		{"白毛小茂密爱撒娇，是只小猫", []string{"宠物", "小茂密"}},
		{"主人对花溪的住处赞不绝口", []string{"地点", "住处"}},
		{"花溪的住处让主人很满意，安静", []string{"地点", "住处"}},
	}
	for i, s := range seed {
		seedDedupeItem(t, store, dedupeTestItem(s.content, s.tags, base.Add(time.Duration(i)*time.Minute)))
	}
	cfg := tidyTestCfg(0.80)
	cfg.CandidateLimit = 1
	emb := stubEmbedder{vecs: map[string]embedding.Vector{}}
	stats, err := newTestTidier(t, store, cfg, emb).RunOnce(ctx)
	if err != nil {
		t.Fatalf("RunOnce: %v", err)
	}
	if stats.DedupeGroups != 2 {
		t.Fatalf("分簇数 = %d, want 2（分簇口径不受候选上限影响）", stats.DedupeGroups)
	}
	if stats.DedupeTruncated != 1 {
		t.Fatalf("DedupeTruncated = %d, want 1（上限 1 只容得下一簇）", stats.DedupeTruncated)
	}
	if len(stats.Truncated) == 0 {
		t.Fatalf("Truncated 为空 —— 被上限切掉的部分必须如实登记分区标签")
	}
}

// TestTidyApplyDedupeGroupSkipsTemporaryRows 是 r1 复审 **P2-1 后半**的回归：
// 候选生成之后某一行才变成临时层（`expires_at` 非空）时，apply **必须整簇 skip**。
//
// 为什么必须有这道复核：候选从生成到「人点 apply」之间隔着任意长的墙钟。若存活行
// 在那之后进了临时层，7 天后它被物理删除（design/23 §19），被吸收行就会永远停在
// `status='merged'` 且 `merged_into` 指向一个不存在的 id。
func TestTidyApplyDedupeGroupSkipsTemporaryRows(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	base := tidyTestTime(t)

	cases := []struct {
		name       string
		markTarget bool
	}{
		{"存活行变成临时层", true},
		{"被吸收行变成临时层", false},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			store := newTestStore(t)
			tags := []string{"身份", "名字"}
			items := []MemoryItem{
				seedDedupeItem(t, store, dedupeTestItem("助理的名字是 Cyrene。", tags, base)),
				seedDedupeItem(t, store, dedupeTestItem("助手名为 Cyrene。", tags, base.Add(time.Minute))),
			}
			cands, _ := dedupeCandsOf(t, store, items)
			if len(cands) != 1 {
				t.Fatalf("簇数 = %d, want 1", len(cands))
			}
			cand := cands[0]
			if created, err := store.insertTidyCandidate(ctx, cand); err != nil || !created {
				t.Fatalf("insert candidate: created=%v err=%v", created, err)
			}
			// 候选落库之后、apply 之前：把目标/被吸收行变成临时层。
			victim := cand.SourceIDs[0]
			if tc.markTarget {
				victim = cand.TargetID
			}
			if _, err := store.db.ExecContext(ctx,
				`UPDATE memory_items SET expires_at = ? WHERE id = ?`,
				expiryText(base.Add(7*24*time.Hour)), victim); err != nil {
				t.Fatalf("mark temporary: %v", err)
			}
			res, err := store.ApplyTidyCandidate(ctx, cand.ID, TidyApplyOptions{Actor: "owner:test"})
			if err != nil {
				t.Fatalf("ApplyTidyCandidate: %v", err)
			}
			if res.Status != TidyApplySkipped || res.Reason != tidySkipTemporaryRow {
				t.Fatalf("结果 = %+v, want skipped/%s", res, tidySkipTemporaryRow)
			}
			for _, id := range cand.SourceIDs {
				if got := loadTidyItemRow(t, store, id).status; got != tidyStatusActive {
					t.Fatalf("被吸收行 %s 状态 = %q, want active（整簇 skip，一行都不写）", id, got)
				}
			}
			if got, _ := store.GetTidyCandidate(ctx, cand.ID); got.Status != tidyCandidatePending {
				t.Fatalf("候选状态 = %q, want pending（skip 不该消费掉候选）", got.Status)
			}
		})
	}
}