package memory import ( "context" "database/sql" "encoding/json" "errors" "fmt" "strings" "time" ) // 本文件是记忆整理的存储层:分区扫描、候选落库与查询、运行留痕。它复用 // memory 包既有的 *sql.DB 与格式化辅助(formatTime/parseTime/scanItem), // 不新建存储机制。 // // 第一版唯一的写入就是候选表与运行表:**没有任何一条 memory_items 会被改写 // 或删除**(design/23 §3 非目标 + §7「apply 才写记忆」)。 // tidyCandidatePending 是候选的初始状态(design/23 §10)。 const tidyCandidatePending = "pending" // tidyPartition 是整理的隔离单元:design/23 §6 的分区键 // (persona_id, scope, project_id)。分区之间**永不比较**。 type tidyPartition struct { personaID string scope Scope projectID string } // label 渲染分区标签,供日志与截断登记使用(不含任何记忆正文)。 func (p tidyPartition) label() string { return fmt.Sprintf("%s/%s/%s", p.personaID, p.scope, p.projectID) } // tidyStatusActive 是「在册」状态。整理只扫描在册条目:已合并/已归档的行 // 不再参与比对(design §10 status 列)。 const tidyStatusActive = "active" // listTidyPartitionItemsSQL 按分区读取**在册**(status='active')条目, // 最新在前;rowid 打破 created_at 并列,使同一刻写入的条目顺序稳定 // (与 ListByPersona 同一口径)。分区过滤写在 SQL 里,而不是读全表再在内存 // 里筛 —— design/23 §6 要求在分区切分层就隔断。 // // 🔴 **刻意不过时效闸**(design/03 §11.2「主会话已裁定:保持不过时效闸」, // 同一取舍的先例见 design/23 §10.1 ①):整理比对的是**正文**,与「这条对外 // 可不可见」无关;已过期的临时记忆随后由后台清扫物理删除(design/23 §19), // 绝不会被整理「复活」。可见性判据只有一处 —— 四条列表读路径共用 timegate.go // 的 timeGateClause。**不要再给这条 SQL 加时效闸。** 本文件另两条分区口径 // 查询(listTidyPartitionsSQL / countTidyPartitionItemsSQL)是同一裁定的实例。 const listTidyPartitionItemsSQL = `SELECT id, persona_id, role_id, project_id, scope, kind, content, tags, confidence, source_conversation, created_at, updated_at, decay_weight, retention_tier, sensitive FROM memory_items WHERE persona_id = ? AND scope = ? AND project_id = ? AND status = ? ORDER BY created_at DESC, rowid DESC LIMIT ?` const listTidyPartitionsSQL = `SELECT DISTINCT persona_id, scope, project_id FROM memory_items WHERE status = ? ORDER BY persona_id, scope, project_id` const countTidyPartitionItemsSQL = `SELECT COUNT(*) FROM memory_items WHERE persona_id = ? AND scope = ? AND project_id = ? AND status = ?` const insertTidyCandidateSQL = `INSERT OR IGNORE INTO memory_tidy_candidates (id, persona_id, scope, project_id, kind, action, source_ids, target_id, proposed_content, merged_content, reason, evidence, rule_version, score_min, score_max, status, idempotency_key, run_id, created_at, decided_at, decided_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '', '')` const insertTidyRunSQL = `INSERT INTO memory_tidy_runs (id, started_at, finished_at, scanned, candidates, applied, skipped, errors, vetoed, veto_reasons, rule_version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` // listTidyCandidatesByRunSQL 按 **run_id + status** 取候选(最新在前)。run_id // 传空串表示「不按轮次筛」—— 那正是「全部待定应用」这条粒度(切片 TIDY-1 // 三种粒度里最粗的一档)。 const listTidyCandidatesByRunSQL = `SELECT id, persona_id, scope, project_id, kind, action, source_ids, target_id, proposed_content, merged_content, reason, evidence, rule_version, score_min, score_max, status, idempotency_key, run_id, created_at FROM memory_tidy_candidates WHERE status = ? AND (? = '' OR run_id = ?) ORDER BY created_at DESC, rowid DESC LIMIT ?` const countTidyCandidatesByRunSQL = `SELECT COUNT(*) FROM memory_tidy_candidates WHERE status = ? AND (? = '' OR run_id = ?)` const selectTidyCandidateSQL = `SELECT id, persona_id, scope, project_id, kind, action, source_ids, target_id, proposed_content, merged_content, reason, evidence, rule_version, score_min, score_max, status, idempotency_key, run_id, created_at FROM memory_tidy_candidates WHERE id = ?` const selectTidyRunSQL = `SELECT id, started_at, finished_at, scanned, candidates, applied, skipped, errors, vetoed, veto_reasons, rule_version FROM memory_tidy_runs WHERE id = ?` // selectLastTidyRunSQL 取最近一轮整理的开始时刻(切片 TIDY-1 的调度判据): // 「上次 run 的时刻」是**持久事实**,不是进程内记忆 —— 后端两天重启十几次 // 正是旧实现(只看内存 lastRun)永远走不完 24h 周期的根因。 const selectLastTidyRunSQL = `SELECT started_at FROM memory_tidy_runs ORDER BY started_at DESC, rowid DESC LIMIT 1` // incrementTidyRunAppliedSQL 在批量应用之后把该轮的 applied 计数累加上去 // (design/23 §10 的 memory_tidy_runs.applied)。运行时那一轮写的是 0(当轮 // 绝不自动应用,§7),这里补的是「事后来应用了几条」。 const incrementTidyRunAppliedSQL = `UPDATE memory_tidy_runs SET applied = applied + ? WHERE id = ?` const listTidyRunsSQL = `SELECT id, started_at, finished_at, scanned, candidates, applied, skipped, errors, vetoed, veto_reasons, rule_version FROM memory_tidy_runs ORDER BY started_at DESC, rowid DESC LIMIT ?` // listTidyPartitions 返回所有含在册条目的分区。DISTINCT 让每个分区的 // 扫描上限(design §9 每分区 5000 条)各自独立。 func (s *Store) listTidyPartitions(ctx context.Context) ([]tidyPartition, error) { rows, err := s.db.QueryContext(ctx, listTidyPartitionsSQL, string(tidyStatusActive)) if err != nil { return nil, fmt.Errorf("memory: list tidy partitions: %w", err) } defer func() { _ = rows.Close() }() parts := make([]tidyPartition, 0) for rows.Next() { var p tidyPartition var scope string if err := rows.Scan(&p.personaID, &scope, &p.projectID); err != nil { return nil, fmt.Errorf("memory: scan tidy partition: %w", err) } p.scope = Scope(scope) parts = append(parts, p) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("memory: iterate tidy partitions: %w", err) } return parts, nil } // listTidyPartitionItems 读取一个分区内的在册条目(最新在前,最多 limit 条)。 func (s *Store) listTidyPartitionItems(ctx context.Context, p tidyPartition, limit int) ([]MemoryItem, error) { rows, err := s.db.QueryContext(ctx, listTidyPartitionItemsSQL, p.personaID, string(p.scope), p.projectID, string(tidyStatusActive), limit) if err != nil { return nil, fmt.Errorf("memory: list tidy partition items: %w", err) } return scanItems(rows, "tidy partition item") } // countTidyPartitionItems 数一个分区里在册条目的**真总数**。只在扫描上限命中时 // 调用一次:日志里的「总数」必须是真数,而不是被 LIMIT 截出来的 len(items) // (那是**下限**,把「至少 5000」写成「正好 5000」就是一句谎)。 func (s *Store) countTidyPartitionItems(ctx context.Context, p tidyPartition) (int, error) { var total int if err := s.db.QueryRowContext(ctx, countTidyPartitionItemsSQL, p.personaID, string(p.scope), p.projectID, string(tidyStatusActive)).Scan(&total); err != nil { return 0, fmt.Errorf("memory: count tidy partition items: %w", err) } return total, nil } // insertTidyCandidate 落库一条候选,返回它是否是**新建**的。幂等键 // (design/23 §9)唯一:同一批数据重复运行命中已存在行时返回 (false, nil), // 绝不重复产候选。 func (s *Store) insertTidyCandidate(ctx context.Context, c TidyCandidate) (bool, error) { c.IdempotencyKey = tidyIdempotencyKey(c) sourceJSON, err := json.Marshal(c.SourceIDs) if err != nil { return false, fmt.Errorf("memory: encode tidy source ids: %w", err) } evidenceJSON, err := json.Marshal(c.Evidence) if err != nil { return false, fmt.Errorf("memory: encode tidy evidence: %w", err) } if c.CreatedAt.IsZero() { c.CreatedAt = time.Now().UTC() } res, err := s.db.ExecContext(ctx, insertTidyCandidateSQL, c.ID, c.PersonaID, string(c.Scope), c.ProjectID, string(c.Kind), string(c.Action), string(sourceJSON), c.TargetID, c.ProposedContent, c.MergedContent, c.Reason, string(evidenceJSON), c.RuleVersion, c.ScoreMin, c.ScoreMax, c.Status, c.IdempotencyKey, c.RunID, formatTime(c.CreatedAt)) if err != nil { return false, fmt.Errorf("memory: insert tidy candidate: %w", err) } n, err := res.RowsAffected() if err != nil { return false, fmt.Errorf("memory: tidy candidate rows affected: %w", err) } return n > 0, nil } // persistTidyRun 写一条运行留痕(design/23 §10 memory_tidy_runs)。 func (s *Store) persistTidyRun(ctx context.Context, run TidyRun) error { if run.ID == "" { run.ID = mustNewTidyID() } reasons, err := json.Marshal(run.VetoReasons) if err != nil { return fmt.Errorf("memory: encode tidy veto reasons: %w", err) } if _, err := s.db.ExecContext(ctx, insertTidyRunSQL, run.ID, formatTime(run.StartedAt), formatTime(run.FinishedAt), run.Scanned, run.Candidates, run.Applied, run.Skipped, run.Errors, run.Vetoed, string(reasons), run.RuleVersion); err != nil { return fmt.Errorf("memory: insert tidy run: %w", err) } return nil } // ListTidyCandidates 返回指定状态的整理候选(最新在前,最多 limit 条, // 非正 limit 取 defaultTidyCandidateLimit)。它是读侧入口:候选正文含隐私, // 只应由 owner-only 的读侧使用(design/23 §11)。 func (s *Store) ListTidyCandidates(ctx context.Context, status string, limit int) ([]TidyCandidate, error) { return s.listTidyCandidates(ctx, status, "", limit) } // ListTidyCandidatesByRun 返回某一轮整理产出的候选(status 默认 pending)。 // runID 为空串表示**不按轮次筛** —— 即「全部待定应用」那一档(切片 TIDY-1)。 func (s *Store) ListTidyCandidatesByRun(ctx context.Context, runID, status string, limit int) ([]TidyCandidate, error) { return s.listTidyCandidates(ctx, status, runID, limit) } // listTidyCandidates 是上面两个读侧入口的共用实现:runID 为空串时走「不筛轮次」 // 的 SQL 分支(`? = ” OR run_id = ?`),因此只有一个查询形状要维护。 func (s *Store) listTidyCandidates(ctx context.Context, status, runID string, limit int) ([]TidyCandidate, error) { if strings.TrimSpace(status) == "" { status = tidyCandidatePending } if limit <= 0 { limit = defaultTidyCandidateLimit } rows, err := s.db.QueryContext(ctx, listTidyCandidatesByRunSQL, status, runID, runID, limit) if err != nil { return nil, fmt.Errorf("memory: list tidy candidates: %w", err) } defer func() { _ = rows.Close() }() out := make([]TidyCandidate, 0) for rows.Next() { c, err := scanTidyCandidate(rows) if err != nil { return nil, fmt.Errorf("memory: scan tidy candidate: %w", err) } out = append(out, c) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("memory: iterate tidy candidates: %w", err) } return out, nil } // GetTidyCandidate 按 id 读一条候选(含已决定的行)。不存在时返回 ErrNotFound。 func (s *Store) GetTidyCandidate(ctx context.Context, id string) (TidyCandidate, error) { if strings.TrimSpace(id) == "" { return TidyCandidate{}, ErrNotFound } row := s.db.QueryRowContext(ctx, selectTidyCandidateSQL, id) c, err := scanTidyCandidate(row) if errors.Is(err, sql.ErrNoRows) { return TidyCandidate{}, ErrNotFound } if err != nil { return TidyCandidate{}, fmt.Errorf("memory: get tidy candidate: %w", err) } return c, nil } // countTidyCandidates 数某状态下(可选按轮次)的候选总数。批量应用用它并把 // 「真总数 vs 本轮应用上限」如实回报给调用方(超限绝不静默)。 func (s *Store) countTidyCandidates(ctx context.Context, status, runID string) (int, error) { if strings.TrimSpace(status) == "" { status = tidyCandidatePending } var total int if err := s.db.QueryRowContext(ctx, countTidyCandidatesByRunSQL, status, runID, runID). Scan(&total); err != nil { return 0, fmt.Errorf("memory: count tidy candidates: %w", err) } return total, nil } // GetTidyRun 按 id 读一条运行留痕。不存在时返回 ErrTidyRunRecordNotFound —— // 面板拿一个拼错的 runId 来批量应用时,必须是「查不到这一轮」而不是「这一轮 // 没有待定候选」。 func (s *Store) GetTidyRun(ctx context.Context, id string) (TidyRun, error) { if strings.TrimSpace(id) == "" { return TidyRun{}, ErrTidyRunRecordNotFound } row := s.db.QueryRowContext(ctx, selectTidyRunSQL, id) r, err := scanTidyRun(row) if errors.Is(err, sql.ErrNoRows) { return TidyRun{}, ErrTidyRunRecordNotFound } if err != nil { return TidyRun{}, fmt.Errorf("memory: get tidy run: %w", err) } return r, nil } // LastTidyRunStartedAt 返回最近一轮整理的开始时刻;一次都没跑过时返回零值 // 时间与 nil(**不是**错误:那是「还没有历史」这一正常状态,调度据此判断 // 是否需要补跑)。 func (s *Store) LastTidyRunStartedAt(ctx context.Context) (time.Time, error) { var startedAt string err := s.db.QueryRowContext(ctx, selectLastTidyRunSQL).Scan(&startedAt) if errors.Is(err, sql.ErrNoRows) { return time.Time{}, nil } if err != nil { return time.Time{}, fmt.Errorf("memory: read last tidy run: %w", err) } t, err := parseTime(startedAt) if err != nil { return time.Time{}, err } return t, nil } // ListTidyRuns 返回整理运行留痕(最新在前,最多 limit 条)。非正 limit 取 // 一个保守的默认条数。 func (s *Store) ListTidyRuns(ctx context.Context, limit int) ([]TidyRun, error) { if limit <= 0 { limit = defaultTidyCandidateLimit } rows, err := s.db.QueryContext(ctx, listTidyRunsSQL, limit) if err != nil { return nil, fmt.Errorf("memory: list tidy runs: %w", err) } defer func() { _ = rows.Close() }() out := make([]TidyRun, 0) for rows.Next() { r, err := scanTidyRun(rows) if err != nil { return nil, fmt.Errorf("memory: scan tidy run: %w", err) } out = append(out, r) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("memory: iterate tidy runs: %w", err) } return out, nil } // scanTidyRun 读取一行运行留痕:veto_reasons 以 JSON 存储,时间戳为定宽 UTC 文本。 func scanTidyRun(sc rowScanner) (TidyRun, error) { var ( r TidyRun startedAt string finishedAt string reasonsJSON string ) if err := sc.Scan(&r.ID, &startedAt, &finishedAt, &r.Scanned, &r.Candidates, &r.Applied, &r.Skipped, &r.Errors, &r.Vetoed, &reasonsJSON, &r.RuleVersion); err != nil { return TidyRun{}, err } var err error if r.StartedAt, err = parseTime(startedAt); err != nil { return TidyRun{}, err } if r.FinishedAt, err = parseTime(finishedAt); err != nil { return TidyRun{}, err } if err := json.Unmarshal([]byte(reasonsJSON), &r.VetoReasons); err != nil { return TidyRun{}, fmt.Errorf("memory: decode tidy veto reasons: %w", err) } return r, nil } // scanTidyCandidate 读取一行候选:source_ids 与 evidence 以 JSON 存储。 func scanTidyCandidate(sc rowScanner) (TidyCandidate, error) { var ( c TidyCandidate scope, kind string action string sourceJSON string evidenceRaw string createdAt string ) if err := sc.Scan(&c.ID, &c.PersonaID, &scope, &c.ProjectID, &kind, &action, &sourceJSON, &c.TargetID, &c.ProposedContent, &c.MergedContent, &c.Reason, &evidenceRaw, &c.RuleVersion, &c.ScoreMin, &c.ScoreMax, &c.Status, &c.IdempotencyKey, &c.RunID, &createdAt); err != nil { return TidyCandidate{}, err } c.Scope = Scope(scope) c.Kind = Kind(kind) c.Action = TidyAction(action) if err := json.Unmarshal([]byte(sourceJSON), &c.SourceIDs); err != nil { return TidyCandidate{}, fmt.Errorf("memory: decode tidy source ids: %w", err) } if c.SourceIDs == nil { c.SourceIDs = []string{} } if err := json.Unmarshal([]byte(evidenceRaw), &c.Evidence); err != nil { return TidyCandidate{}, fmt.Errorf("memory: decode tidy evidence: %w", err) } t, err := parseTime(createdAt) if err != nil { return TidyCandidate{}, err } c.CreatedAt = t return c, nil }