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 }