Cyrene_v2/internal/runtime/memory/tidy_dedupe_apply.go

package memory

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"
	"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 逐条同源（分区硬门、只读来源硬门、行必须存在、**临时层不复核**），
// 全部在 loadDedupeGroupRows 里；任何一条不通过都如实 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, absorbed, skip, err := s.loadDedupeGroupRows(ctx, cand)
	if err != nil {
		return base, err
	}
	if skip != "" {
		return skipTidyApply(base, skip), nil
	}
	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)
}

// loadDedupeGroupRows 读出存活行与被吸收行并做**全部前置校验**；skip 非空表示应如实
// 跳过（原因由它给出），此时两行都不可用。
//
// 四类校验，缺一不可：
//  1. **分区硬门**（§5.2-7/8）：候选与行必须同 persona/scope/project；
//  2. **只读来源硬门**（§5.2-9、§6）：群聊/陌生人来源绝不改写；
//  3. **临时层复核**（r1 复审 P2-1）：`expires_at` 非空的行既不许当存活行（会被
//     物理删除，design/23 §19 ⇒ 被吸收行悬空）也不许当被吸收行（本来就要消失）——
//     候选从生成到被人点是过时候选，必须在落库前按主键再判一次；
//  4. 形状：source 不得等于 target。
func (s *Store) loadDedupeGroupRows(ctx context.Context, cand TidyCandidate) (MemoryItem, []MemoryItem, string, error) {
	empty := MemoryItem{}
	survivor, err := s.Get(ctx, cand.TargetID)
	if errors.Is(err, ErrNotFound) {
		return empty, nil, tidySkipSourceMissing, nil
	}
	if err != nil {
		return empty, nil, "", err
	}
	if skip := dedupeRowGuard(cand, survivor); skip != "" {
		return empty, nil, skip, nil
	}
	if temporary, terr := s.tidyItemTemporary(ctx, survivor.ID); terr != nil {
		return empty, nil, "", terr
	} else if temporary {
		return empty, nil, tidySkipTemporaryRow, nil
	}
	absorbed := make([]MemoryItem, 0, len(cand.SourceIDs))
	for _, id := range cand.SourceIDs {
		if id == cand.TargetID {
			return empty, nil, tidySkipInvalidCandidate, nil
		}
		it, gerr := s.Get(ctx, id)
		if errors.Is(gerr, ErrNotFound) {
			return empty, nil, tidySkipSourceMissing, nil
		}
		if gerr != nil {
			return empty, nil, "", gerr
		}
		if skip := dedupeRowGuard(cand, it); skip != "" {
			return empty, nil, skip, nil
		}
		if temporary, terr := s.tidyItemTemporary(ctx, it.ID); terr != nil {
			return empty, nil, "", terr
		} else if temporary {
			return empty, nil, tidySkipTemporaryRow, nil
		}
		absorbed = append(absorbed, it)
	}
	return survivor, absorbed, "", nil
}

// dedupeRowGuard 是「这一行能不能被这条候选改动」的静态判据（分区 + 只读来源）。
func dedupeRowGuard(cand TidyCandidate, it MemoryItem) string {
	if skip := tidyPartitionMismatch(cand, it); skip != "" {
		return skip
	}
	if nonOwnerSource(it.SourceConversation) {
		// 只读来源（群聊/陌生人）绝不改写（design/23 §5.2-9/§6）：纵深防御，
		// 规则层本就把这类条目排除在分簇之外。
		return tidySkipSourceRestricted
	}
	return ""
}

// tidyItemTemporarySQL 只读一行的 `expires_at`（临时层判据，design/03 §9.6）。
const tidyItemTemporarySQL = `SELECT expires_at FROM memory_items WHERE id = ?`

// tidyItemTemporary 报告一条记忆是否在**临时层**（`expires_at` 非空）。
//
// 它服务应用侧的**过时复核**：`MemoryItem` 的读侧 SELECT 不含这一列（该列只写不读），
// 因此候选生成时用分区级元信息判一次、真正落库前再按主键判一次 —— 两次都要，
// 因为「候选生成」与「人点 apply」之间隔着任意长的墙钟（r1 复审 P2-1）。
func (s *Store) tidyItemTemporary(ctx context.Context, id string) (bool, error) {
	var expiresAt string
	err := s.db.QueryRowContext(ctx, tidyItemTemporarySQL, id).Scan(&expiresAt)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("memory/tidy: read item expiry %s: %w", id, err)
	}
	return strings.TrimSpace(expiresAt) != "", nil
}