AskaEth
6a3f3a329a
M27 体检（DevDocs/audit/记忆库价值体检-2026-10-24.md）两条根治项：
1) 新增 `dedupe_group` 候选动作（不放大 containNone）：
- 簇只认「同 scope+kind+完全相同 tag 集」或「同 scope+kind+前 12 字相同」
（后者附加词项余弦 ≥0.70 与同 kind 收窄）；
- 每簇保留超集：内容最长 > merge_count 大 > created_at 早 > id；
- 安全门与既有合并路径同源：受限来源/绝对时刻/时效状态/低置信一律不进簇，
数值与否定与存活行不一致的成员被逐条剔出；临时层行（expires_at 非空）
两个方向都不进簇（存活行被清扫会给被吸收行留悬空 merged_into）；
- 已被 archive/merge 候选覆盖的条目不重复出第二条待办（抑制）；
- 落库形态与 merge 同构：source_ids 多条 + target_id，应用为整簇原子的软合并，
回滚复用 revertTidyMerge（逐条复原）；
- 批量应用按**行**计上限 200 行/批（maxTidyApplyRowsPerBatch），超限如实 truncated；
- 不进自动档（人工/批量点）；试算（DryRun）零写库给出簇数与可减少行数。
2) 留存闸改组合口径（治根）+ N6/N7：
- 护栏②从「含习惯/约定词就一票放行」改为「锚点词 **且** 含具体约束
（数字/禁忌/频次规律）」；无约束的锚点交给更窄的规则判；
- N1/N3/N4/N5 这几条词项规则改为「无锚点时才开火」（`主人喜欢说晚安`
这类偏好不再因寒暄词被判临时）；
- 新增 N6（指示性时点，句首前 12 字内）与 N7（进行中状态），安全阀：习惯词、
硬性长期事实词、数值/约束、寒暄/情绪词一律按回去；
- 命中即第五态 defer，`expires_at = 判定时刻 + tempTtlSec`（默认 7 天，
已落在 $DATA_DIR/config/memory-tidy.json，零新 .env 键）；**只开新增侧**，
存量一条不补写（用例钉住）。
契约（加性，不改既有字段语义）：tidy dry-run/run 回执加 dedupeGroups /
dedupeRowsReducible / dedupeTruncated；批量应用回执加 deduped / rowsAffected。
实测（生产库只读 mode=ro，active 3667）：
- 闸：改前 2 条(0.05%) → 改后 89 条(2.43%)：n6 58 / n7 29 / n1 2；
89 条里 30 条含锚点词，逐条人读均为「约定明天中午见」「约定下周留个午后」
这类到期即失效的安排（唯一可议：1 条把「中午互动的日常」一起带进临时层）。
- dedupe_group：281 簇 / 可减 594 行（16.2% of active）；tag_set 265 / prefix12 16；
簇大小 2:170 3:48 4:27 5:10 6:10 7:7 8:2 9:2 10:1。
验收：go build ./... + go vet ./... + go test ./... 全绿。
297 lines
12 KiB
Go
297 lines
12 KiB
Go
package memory
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// 本文件是记忆整理**应用路径**（切片 TIDY-1）的存储层：候选认领、软合并 /
|
||
// 软删 / 改 kind 的事务写入，以及回滚。它复用 memory 包既有的 *sql.DB、
|
||
// 分区口径（tidyPartition / tidyStatusActive）与软合并且已存在的辅助
|
||
// （preinsert_store.go 的 bumpSurvivor / dropSurvivorLink / revertMergedSQL /
|
||
// selectMergedIntoSQL —— 那是 design/23 §8 的**同一套**软合并列语义，不是第二套）。
|
||
//
|
||
// 与 D3（入库前审查）那套的关系：语义同源（status/merged_into/merged_from/
|
||
// merge_count），但**闸口不同** —— D3 的顺序队列保证一条记忆只审一次，因此它的
|
||
// markAbsorbed 不带 `AND status='active'`；整理的候选是**事后**执行的，可能
|
||
// 过时（源行已被别的动作改过），所以这里每一条 UPDATE 都带状态条件，并把它
|
||
// 当作幂等/过时的判据（RowsAffected == 0）。
|
||
|
||
const (
|
||
// tidyCandidateApplied 是候选「已应用」状态（design/23 §10 的 status 闭集）。
|
||
tidyCandidateApplied = "applied"
|
||
|
||
// tidyStatusArchived 是软删后的行状态（design/23 §10：active/merged/archived）。
|
||
// 归档行不再参与检索与整理，但**行与原文都还在**，因此可 revert（§8）。
|
||
tidyStatusArchived = "archived"
|
||
)
|
||
|
||
// claimTidyCandidateSQL 是**执行闸**：只有把候选从 pending 原子地推到
|
||
// applied 的那一次调用才继续往下改记忆。WHERE 里的状态条件就是幂等性的根，
|
||
// 因此它绝不能省（少了它，重复应用会再写一次 memory_items）。
|
||
const claimTidyCandidateSQL = `UPDATE memory_tidy_candidates
|
||
SET status = ?, decided_at = ?, decided_by = ? WHERE id = ? AND status = ?`
|
||
|
||
// markTidyAbsorbedSQL 标记被吸收行：status='merged' + merged_into=存活 id。
|
||
// `AND status='active'` 是第二道闸：源行已被别的动作改过时 RowsAffected==0，
|
||
// 调用方据此回滚整个事务（不产生第二次合并计数）。
|
||
const markTidyAbsorbedSQL = `UPDATE memory_items
|
||
SET status = ?, merged_into = ?, updated_at = ? WHERE id = ? AND status = ?`
|
||
|
||
// setTidyItemStatusSQL 是软删 / 复原共用的一条：只改 status 与 updated_at，
|
||
// 带 from 状态条件。
|
||
const setTidyItemStatusSQL = `UPDATE memory_items
|
||
SET status = ?, updated_at = ? WHERE id = ? AND status = ?`
|
||
|
||
// reclassifyTidyItemSQL 只改 kind 一列（design/23 §5.1 T0-6），带状态条件。
|
||
const reclassifyTidyItemSQL = `UPDATE memory_items
|
||
SET kind = ?, updated_at = ? WHERE id = ? AND status = ?`
|
||
|
||
// undoTidyReclassifySQL 撤回一次 kind 修正：只在当前 kind 仍是当时写的建议值、
|
||
// 且行仍在册时生效（否则说明有人后来手动改过，不该被回滚覆盖）。
|
||
const undoTidyReclassifySQL = `UPDATE memory_items
|
||
SET kind = ?, updated_at = ? WHERE id = ? AND status = ? AND kind = ?`
|
||
|
||
// tidyApplyTx 在**一个事务**里跑一次应用：先认领候选（条件更新 = 执行闸），
|
||
// 再执行 mutate，最后提交。任何一步失败都回滚 —— 因此「候选被标 applied 但
|
||
// 记忆没改」这种半成品状态不可能落库。
|
||
func (s *Store) tidyApplyTx(ctx context.Context, cand TidyCandidate, opts TidyApplyOptions,
|
||
now time.Time, mutate func(context.Context, *sql.Tx) error) error {
|
||
tx, err := s.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: apply candidate begin: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
claimed, err := claimTidyCandidate(ctx, tx, cand.ID, tidyCandidatePending, tidyCandidateApplied, opts.Actor, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !claimed {
|
||
return errTidyCandidateDecided
|
||
}
|
||
if err := mutate(ctx, tx); err != nil {
|
||
return err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return fmt.Errorf("memory/tidy: apply candidate commit: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// tidyRevertTx 与 tidyApplyTx 同构，只是认领的方向相反（applied → pending）。
|
||
func (s *Store) tidyRevertTx(ctx context.Context, cand TidyCandidate, opts TidyApplyOptions,
|
||
mutate func(context.Context, *sql.Tx) error) error {
|
||
tx, err := s.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: revert candidate begin: %w", err)
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
|
||
claimed, err := claimTidyCandidate(ctx, tx, cand.ID, tidyCandidateApplied, tidyCandidatePending, opts.Actor, time.Now().UTC())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !claimed {
|
||
return errTidyCandidateDecided
|
||
}
|
||
if err := mutate(ctx, tx); err != nil {
|
||
return err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return fmt.Errorf("memory/tidy: revert candidate commit: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// claimTidyCandidate 原子地把候选从**当前状态 from** 推到 to。返回是否真的改了行：
|
||
// false 表示它已经被决定过（幂等命中，不是错误）。
|
||
//
|
||
// from 参数是这道闸的关键：应用要求 from='pending'，回滚要求 from='applied'。
|
||
// 两者共用一个语句，但**不能共用一个硬编码的 from** —— 那会让回滚永远匹配不到行。
|
||
func claimTidyCandidate(ctx context.Context, tx *sql.Tx, id, from, to, actor string, now time.Time) (bool, error) {
|
||
res, err := tx.ExecContext(ctx, claimTidyCandidateSQL, to, formatTime(now), actor, id, from)
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: claim candidate %s: %w", id, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: claim candidate %s rows: %w", id, err)
|
||
}
|
||
return n > 0, nil
|
||
}
|
||
|
||
// markTidyAbsorbed 标记被吸收行并刷新存活行的合并元信息。源行不在册时返回
|
||
// errTidyItemNotActive（事务因此回滚）。
|
||
func markTidyAbsorbed(ctx context.Context, tx *sql.Tx, absorbedID, survivorID string, now time.Time) error {
|
||
res, err := tx.ExecContext(ctx, markTidyAbsorbedSQL,
|
||
string(ingestStatusMerged), survivorID, formatTime(now), absorbedID, tidyStatusActive)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: mark absorbed item: %w", err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: mark absorbed item rows: %w", err)
|
||
}
|
||
if n == 0 {
|
||
return errTidyItemNotActive
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// setTidyItemStatus 在 from → to 之间改一行的 status。返回是否真的改了行。
|
||
func setTidyItemStatus(ctx context.Context, tx *sql.Tx, id, from, to string, now time.Time) (bool, error) {
|
||
res, err := tx.ExecContext(ctx, setTidyItemStatusSQL, to, formatTime(now), id, from)
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: set item status %s: %w", id, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: set item status rows: %w", err)
|
||
}
|
||
return n > 0, nil
|
||
}
|
||
|
||
// archiveTidyItem 把一条在册记忆软删（active → archived）。不在册时返回
|
||
// errTidyItemNotActive。它**只改 status/updated_at**，绝不碰 content。
|
||
func archiveTidyItem(ctx context.Context, tx *sql.Tx, id string, now time.Time) error {
|
||
changed, err := setTidyItemStatus(ctx, tx, id, tidyStatusActive, tidyStatusArchived, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !changed {
|
||
return errTidyItemNotActive
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// reclassifyTidyItem 把一条在册记忆的 kind 改为 kind。不在册时返回
|
||
// errTidyItemNotActive。
|
||
func reclassifyTidyItem(ctx context.Context, tx *sql.Tx, id string, kind Kind, now time.Time) error {
|
||
res, err := tx.ExecContext(ctx, reclassifyTidyItemSQL, string(kind), formatTime(now), id, tidyStatusActive)
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: reclassify item %s: %w", id, err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return fmt.Errorf("memory/tidy: reclassify item rows: %w", err)
|
||
}
|
||
if n == 0 {
|
||
return errTidyItemNotActive
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// addTidyRunApplied 把该轮留痕的 applied 计数累加（design/23 §10）。
|
||
func (s *Store) addTidyRunApplied(ctx context.Context, runID string, applied int) error {
|
||
if _, err := s.db.ExecContext(ctx, incrementTidyRunAppliedSQL, applied, runID); err != nil {
|
||
return fmt.Errorf("memory/tidy: bump run applied %s: %w", runID, err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// revertTidyAction 在事务里撤销一条已应用候选对记忆的改动，返回真正被复原的
|
||
// item id（一条都没复原时返回空切片 —— 那说明用户已经用 /memories/{id}/revert
|
||
// 手工撤过了，此时只把候选退回 pending，不假装改过东西）。
|
||
func revertTidyAction(ctx context.Context, tx *sql.Tx, cand TidyCandidate, now time.Time) ([]string, error) {
|
||
switch cand.Action {
|
||
case TidyMerge, TidyDedupeGroup:
|
||
// dedupe_group 与 merge 是**同一套软合并**（被吸收行 → merged + merged_into），
|
||
// 因此回滚逐字复用 revertTidyMerge：它本来就按 SourceIDs 逐条复原。
|
||
return revertTidyMerge(ctx, tx, cand, now)
|
||
case TidyArchive:
|
||
return revertTidyArchive(ctx, tx, cand, now)
|
||
case TidyReclassify:
|
||
return revertTidyReclassify(ctx, tx, cand, now)
|
||
default:
|
||
return []string{}, nil
|
||
}
|
||
}
|
||
|
||
// revertTidyMerge 复原一次软合并：逐条把被吸收行从 merged 复原为 active，并从
|
||
// 存活行的 merged_from 里摘掉它（dropSurvivorLink）。只复原**这条候选做的那次**
|
||
// 合并（merged_into 必须等于候选的 target）。
|
||
func revertTidyMerge(ctx context.Context, tx *sql.Tx, cand TidyCandidate, now time.Time) ([]string, error) {
|
||
out := make([]string, 0, len(cand.SourceIDs))
|
||
for _, id := range cand.SourceIDs {
|
||
ok, err := revertTidyMergeInTx(ctx, tx, id, cand.TargetID, now)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if ok {
|
||
out = append(out, id)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// revertTidyMergeInTx 复原单条被吸收行。它只认 `status='merged'` 且
|
||
// `merged_into=<该候选的存活 id>` 的行；targetID 为空时接受任意存活 id
|
||
// （容错：老候选可能没落 target）。
|
||
func revertTidyMergeInTx(ctx context.Context, tx *sql.Tx, absorbedID, targetID string, now time.Time) (bool, error) {
|
||
var survivorID string
|
||
err := tx.QueryRowContext(ctx, selectMergedIntoSQL, absorbedID, string(ingestStatusMerged)).Scan(&survivorID)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return false, nil
|
||
}
|
||
if err != nil {
|
||
return false, fmt.Errorf("memory/tidy: read merged item %s: %w", absorbedID, err)
|
||
}
|
||
if targetID != "" && survivorID != targetID {
|
||
// 它现在挂在**别的**存活行上（后来又被合并过一次）：这条候选的回滚
|
||
// 不该把它从新关系里拽出来。
|
||
return false, nil
|
||
}
|
||
if _, err := tx.ExecContext(ctx, revertMergedSQL,
|
||
string(tidyStatusActive), formatTime(now), absorbedID, string(ingestStatusMerged)); err != nil {
|
||
return false, fmt.Errorf("memory/tidy: restore merged item %s: %w", absorbedID, err)
|
||
}
|
||
if survivorID != "" {
|
||
if err := dropSurvivorLink(ctx, tx, survivorID, absorbedID, now); err != nil {
|
||
return false, err
|
||
}
|
||
}
|
||
return true, nil
|
||
}
|
||
|
||
// revertTidyArchive 复原被软删（archived）的行：archived → active。
|
||
func revertTidyArchive(ctx context.Context, tx *sql.Tx, cand TidyCandidate, now time.Time) ([]string, error) {
|
||
out := make([]string, 0, len(cand.SourceIDs))
|
||
for _, id := range cand.SourceIDs {
|
||
changed, err := setTidyItemStatus(ctx, tx, id, tidyStatusArchived, tidyStatusActive, now)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if changed {
|
||
out = append(out, id)
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// revertTidyReclassify 撤回一次 kind 修正：只在当前 kind 仍等于当时写的建议值、
|
||
// 且行仍在册时生效（有人后来手动改过就保持不动）。旧 kind 来自候选证据 KindA。
|
||
func revertTidyReclassify(ctx context.Context, tx *sql.Tx, cand TidyCandidate, now time.Time) ([]string, error) {
|
||
kind := Kind(cand.Evidence.ProposedKind)
|
||
previous := Kind(cand.Evidence.KindA)
|
||
if !previous.valid() || !kind.valid() || len(cand.SourceIDs) != 1 {
|
||
return []string{}, nil
|
||
}
|
||
res, err := tx.ExecContext(ctx, undoTidyReclassifySQL,
|
||
string(previous), formatTime(now), cand.SourceIDs[0], tidyStatusActive, string(kind))
|
||
if err != nil {
|
||
return nil, fmt.Errorf("memory/tidy: undo reclassify %s: %w", cand.SourceIDs[0], err)
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("memory/tidy: undo reclassify rows: %w", err)
|
||
}
|
||
if n == 0 {
|
||
return []string{}, nil
|
||
}
|
||
return []string{cand.SourceIDs[0]}, nil
|
||
}
|