Cyrene_v2/internal/runtime/memory/tidy_dedupe_apply.go

package memory

import (
	"context"
	"database/sql"
	"errors"
	"time"
)

// 本文件是 `dedupe_group`（2026-10-24，M34）的**应用与回滚**执行面。
//
// 它与 merge 共用同一套语义（design/23 §8 的软合并）：被吸收行
// `status='merged' + merged_into=<存活 id>`，存活行 `merged_from` 追加、`merge_count`
// 累加；**绝不改正文、绝不删行**。因此回滚直接复用 `revertTidyMerge`（它本来就按
// `SourceIDs` 逐条复原，不是为「一对一」写的）。
//
// 与单条 merge 的两点差别、以及它们的理由：
//
//  1. **一条候选吸收多行**（一簇 N 条）。整簇在**一个事务**里落，任何一条不在册
//     ⇒ 整簇回滚（`errTidyItemNotActive`），不会出现「吞了一半」的中间态 ——
//     半吞的簇最坏：面板显示 applied，但被吞行里有一半还在 active 集合里，
//     回滚也无从判断该复原哪几条。
//  2. **按行计的分批上限**（`maxTidyApplyRowsPerBatch`）。一条候选动 N 行，若只按
//     「候选条数」限批，一次批量应用可能在一条命令里改上千行记忆 —— 这正是体检
//     报告 §7.2 要求分批的理由（「第 2 批必须分批 + 每批留 run_id」）。行预算在
//     `ApplyTidyCandidates` 里统一执行，超限如实报 `truncated`。
//
// 🔴 **不进自动档**：`tidyAutoApplyEligible` 的 switch 只认 merge/archive/reclassify，
// 默认分支返回 false，因此 dedupe_group 永远只走人工/批量点（测试钉着这一条）。

// maxTidyApplyRowsPerBatch 是一次批量应用**最多改动的记忆行数**（dedupe_group 按
// 被吸收行计）。200 的来源是体检报告 §7 第 2 批的建议值：一簇平均 3~4 行，
// 200 行 ≈ 50 簇，一屏可核对；它比候选条数上限（100）**更严**的场合是「簇很大」，
// 这正是需要护栏的那种数据。
const maxTidyApplyRowsPerBatch = 200

// tidyApplyRowCost 返回一条候选**会改动的记忆行数**，供批量应用的按行预算使用。
// 只有 dedupe_group 是多行动作；其余三种动作恒为 1（merge/archive/reclassify 各碰一行）。
func tidyApplyRowCost(c TidyCandidate) int {
	if c.Action == TidyDedupeGroup && len(c.SourceIDs) > 0 {
		return len(c.SourceIDs)
	}
	return 1
}

// applyTidyDedupeGroup 把一簇被吸收行软合并进存活行。
//
// 前置校验与 merge 逐条同源：分区硬门、只读来源硬门、目标行必须存在且与候选同分区。
// 任何一条不通过都如实 skip（不假装做过），且**一行都不写**。
func (s *Store) applyTidyDedupeGroup(ctx context.Context, cand TidyCandidate, opts TidyApplyOptions, now time.Time) (TidyApplyResult, error) {
	base := TidyApplyResult{
		CandidateID: cand.ID, Action: cand.Action, TargetID: cand.TargetID, AffectedIDs: []string{},
	}
	if len(cand.SourceIDs) == 0 || cand.TargetID == "" {
		return skipTidyApply(base, tidySkipInvalidCandidate), nil
	}
	survivor, err := s.Get(ctx, cand.TargetID)
	if errors.Is(err, ErrNotFound) {
		return skipTidyApply(base, tidySkipSourceMissing), nil
	}
	if err != nil {
		return base, err
	}
	if skip := tidyPartitionMismatch(cand, survivor); skip != "" {
		return skipTidyApply(base, skip), nil
	}
	if nonOwnerSource(survivor.SourceConversation) {
		// 只读来源（群聊/陌生人）绝不改写（design/23 §5.2-9/§6）：纵深防御，
		// 规则层本就把这类条目排除在分簇之外。
		return skipTidyApply(base, tidySkipSourceRestricted), nil
	}
	absorbed := make([]MemoryItem, 0, len(cand.SourceIDs))
	for _, id := range cand.SourceIDs {
		if id == cand.TargetID {
			return skipTidyApply(base, tidySkipInvalidCandidate), nil
		}
		it, gerr := s.Get(ctx, id)
		if errors.Is(gerr, ErrNotFound) {
			return skipTidyApply(base, tidySkipSourceMissing), nil
		}
		if gerr != nil {
			return base, gerr
		}
		if skip := tidyPartitionMismatch(cand, it); skip != "" {
			return skipTidyApply(base, skip), nil
		}
		if nonOwnerSource(it.SourceConversation) {
			return skipTidyApply(base, tidySkipSourceRestricted), nil
		}
		absorbed = append(absorbed, it)
	}
	err = s.tidyApplyTx(ctx, cand, opts, now, func(ctx context.Context, tx *sql.Tx) error {
		for _, it := range absorbed {
			if err := markTidyAbsorbed(ctx, tx, it.ID, survivor.ID, now); err != nil {
				return err
			}
			// 逐条 bump：bumpSurvivor 在事务内**重读**存活行的 merge_count，
			// 因此连续调用会把 N 条都记进 merged_from、把计数加 N 次。
			if err := bumpSurvivor(ctx, tx, survivor, it, now); err != nil {
				return err
			}
		}
		return nil
	})
	affected := make([]string, 0, len(absorbed))
	for _, it := range absorbed {
		affected = append(affected, it.ID)
	}
	return finishTidyApply(base, affected, err, tidySkipSourceNotActive)
}