feat(memory): 临时记忆到期物理删除(有界清扫 + 同步清向量 + 删前审计)

design/23 §19 的第 ② / ③ 步:读侧时效闸(M1)只负责「看不见」,本切片负责
「真删掉」。三条护栏照 §19.3:删除是本包内唯一 DELETE 出口、有界(500 行/批 ×
20 批)、删前留审计。

- internal/runtime/memory/expire.go:扫描集合写死为「expires_at ≠ '' ∧
  expires_at ≤ now ∧ status='active' ∧ 来源非受限」。空 expires_at(长期/日常
  记忆,含 M2 登记的「INSERT 与补写之间中断」窗口)结构性进不了集合;受限来源
  复用 nonOwnerSource(与整理/审查同一处判据)。键集游标 (expires_at, rowid)
  保证被跳过的行只读一次、扫描位置一定前移。审计串 = id + persona_id + scope +
  expires_at + 按 rune 截断(48)的摘要 —— 本仓唯一一处允许携带内容片段的审计。
- internal/runtime/memory/expire_loop.go:清扫循环挂在 TidyService 的生命周期上
  (AttachExpirySweep 由组合根调用),随心跳 ctx 跑、随 Stop 停;一轮只写一行
  汇总日志,逐条的那一行进审计。
- internal/runtime/memory/vector/delete.go:新增可选删除能力 vector.Deleter +
  两个后端实现,删行时同步清向量(M1 reviewer 的 finding:过期向量占 top-K 预算)。
- cmd/cyrene/wire_memory_tidy.go:挂上清扫;顺带把已过时的「这十键」改成「这十一
  键」(M2 加了第十一键 tempTtlSec)。
This commit is contained in:
AskaEth committed 2026-10-13 21:43:27 +08:00
1 parent 80096c8bd1
commit a017ab6cf8
7 files changed
+1262 -6

No files matched your search

+18 -6
View File
@@ -10,7 +10,7 @@ import (
// 本文件把记忆整理(DevDocs/design/23-记忆整理与去重 §4)接进组合根。
//
// 四件事只在这里定:
// 六件事只在这里定:
// - **形态**:定期后台任务(安静时段降频)的心跳(memory.TidyService),不是
// 会话内派 subagent。它在 main.go 与 backthink 并列启动、在 close 里先于关库停止;
// - **依赖面**:只给它记忆存储与 embedder —— 服务侧拿不到任何会话/历史读数,
@@ -26,7 +26,12 @@ import (
// 变成「无限制」,也不能让整理频率变成忙轮询);
// - **启动播种**(2026-10-07 追加,主人「后端**自动创建**独立的面板配置文件**要做**」):
// 那个文件不存在时由后端自己写一份「当前生效值」进去(seedMemoryTidyPanel),
// 使「面板优先」从第一次启动起就成立,而不是靠 `.env` 一直兜着。
// 使「面板优先」从第一次启动起就成立,而不是靠 `.env` 一直兜着;
// - **到期清扫的挂点**(2026-10-13 追加,切片 M3,design/23 §19):临时记忆到期
// 后要**真删掉**(读侧时效闸只负责「看不见」)。清扫是 memory 包内的后台件
// (expire.go / expire_loop.go),这里只把它挂到整理服务的生命周期上
// (AttachExpirySweep)—— 它随心跳 ctx 跑、随 Stop 停,因此不新开关闭链、
// 不新开端口、不加 env 键。
//
// 刻意**不做**的一件事(design §4.1 F):
// - 不把 sess-memorytidy 加进 flowPub 的图源白名单(默认关:整理是系统任务,
@@ -49,6 +54,11 @@ func (a *app) wireMemoryTidy() error {
if err != nil {
return fmt.Errorf("memory tidy: %w", err)
}
// 到期清扫(design/23 §19 的第 ② / ③ 步)挂到整理服务的生命周期上:它随心跳
// 的 ctx 跑、随 a.memoryTidy.Stop 停(`design/23 §19.3-3` 的「可停」),因此
// 这里**不新起一条关闭链**、不新开监听端口、不加任何 env 键。清扫自身的能力面
// 只在 memory 包内(expire.go / expire_loop.go),组合根这一句只负责挂上去。
svc.AttachExpirySweep()
a.memoryTidy = svc
return nil
}
@@ -58,11 +68,13 @@ func (a *app) wireMemoryTidy() error {
// 而不是每次都靠 `.env` 兜。失败只记日志、**绝不阻断启动**:运行期仍按 `.env` +
// 代码默认跑,行为与播种前逐字相同(与 materializeModels 同一口径)。
//
// 闸与 T1 的四键(ingestGate / t1Judge / t1PerWindow / t1WindowSec)在 `.env` 里
// **没有出处**(`.env` 已冻结,它们只由这个面板文件与代码默认决定,见
// wireMemoryReview 的三个 helper),而播种只发生在**面板文件不存在**时 —— 那正是
// 闸与 T1 的四键(ingestGate / t1Judge / t1PerWindow / t1WindowSec)**外加**临时
// 记忆 TTL 一键(tempTtlSec,切片 M2 加进来的第五个 review 键)在 `.env` 里
// **都没有出处**(`.env` 已冻结,它们只由这个面板文件与代码默认决定,见
// wireMemoryReview 的四个 helper:memoryIngestGateOff / memoryT1JudgeOff /
// memoryT1RateLimit / memoryTempTTL),而播种只发生在**面板文件不存在**时 —— 那正是
// 「零值 + normalizeReviewConfig」这一条路径,因此这里给零值 ReviewConfig 就是
// **当时**的生效值,不是猜的。审查开关本身(MEMORY_REVIEW_ENABLED)不在这十键里,
// **当时**的生效值,不是猜的。审查开关本身(MEMORY_REVIEW_ENABLED)不在这十一键里,
// 因此不必为它去读一次 env。
func (a *app) seedMemoryTidyPanel(tidy memory.TidyConfig) {
path := runtimeconfig.DefaultMemoryTidyPath(resolveDataDir(a.cfg))
+319
View File
@@ -0,0 +1,319 @@
package memory
import (
"context"
"fmt"
"strings"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector"
)
// 本文件是**临时期限到期处置的第 ② 步与第 ③ 步**:后台到期清扫(物理 DELETE)
// 与「删前留审计」(design/23 §19,主人 Q1 裁定「临时记忆到期删除」)。
//
// 三步是刻意分开的(§19.1):
//
// ① 隐藏 —— 读侧时效闸(timegate.go):每次读比一次列值,**零写入**、即时;
// ② 删除 —— 本文件:显式、有界、有审计、可停的后台路径;
// ③ 审计 —— 本文件:**每行删除前**先写一条 `memory.tidy.expire`。
//
// 把删除混进读路径是这条链上最容易犯的错:读路径会因此变成「有副作用且不可逆」,
// 一次带 bug 的查询就能删记忆。因此删除只有这里一个出口,只由后台件调用。
//
// 🔴 **与「写路径唯一」红线怎么共存**:`design/23 §15.1 D3-1` 那条约束的是
// **INSERT(入库)**,`Store.Add` 是唯一 INSERT 路径。到期删除是一条 **DELETE
// (清除)**,不是第二条入库路径;既有先例是 `Store.Delete` / `DeleteScoped` 与
// `preinsert_gc.go` 的硬删留痕,两者都只删行、不造行。本清扫与之同类:**自带审计**,
// 不碰 `Store.Add`,也不复用任何「想写记忆」的调用点。
//
// 🔴 **删除范围写死**(`design/23 §19.2`,逐条可测):
//
// expires_at <> '' ∧ expires_at <= now ∧ status = 'active' ∧ 来源非受限
//
// - **空 `expires_at` = 永不过期,绝不删**。长期 / 日常记忆的 `expires_at` 是
// 空串(`store_schema.go` 的 DDL 默认值),因此它们**结构性**进不了扫描集合 ——
// 这不是靠注释约定,而是靠 SQL 里那条 `expires_at <> ''` 谓词。
// - **受限来源(群聊 / 陌生人)不参与**:判据复用整理/审查同一处实现
// `nonOwnerSource`(`tidy_rules.go`),因此三处口径不可能各说各话。
// - **已软删 / 已归档的行不参与**(`status='active'` 谓词):它们不是「到期的
// 临时记忆」,是整理软动作留下的行;该不该清是另一件事。
const (
// DefaultExpireSweepInterval 是到期清扫的默认周期:每小时一次
// (`design/23 §19.1` 表格「到期后 ≤ 一个清扫周期(默认每小时扫一次)」)。
// 与 `DefaultIngestReviewGCInterval` 同值同由:清扫只是删几行,频率再高无收益。
DefaultExpireSweepInterval = time.Hour
// expireSweepBatchSize 是单批取行 / 删除的行数上限,maxExpireSweepBatches 是
// 一轮清扫的批数上限。两者一起把「一轮清扫」的时长、单次内存占用与**连接池
// 占用**(本项目的 SQLite 池是单连接)都变得有界 —— 照 `preinsert_gc.go:65-69`
// 的批删范式(`design/23 §19.3-3`「一批删除有条数 / 批数上限」)。
// 上限之积 = 一轮最多删 1 万行;积压时剩下的留给下一轮,不一次 DELETE 全表。
expireSweepBatchSize = 500
maxExpireSweepBatches = 20
// expireAuditSummaryRunes 是审计摘要的**按 rune 固定上限**(`design/23 §19.4`)。
// 48 个 rune 足够一眼看出「删的是哪条」,同时把审计表增长压住 —— 到期审计是
// **逐条高基数**行(一行一个 id),不截断等于把正文搬进审计表。取 48 而不是
// 更宽的值:摘要只需可辨认,不需要可复原(正文早已不可逆地删掉了)。
expireAuditSummaryRunes = 48
// expireAuditVerb 是到期删除的审计动作名,与 `memory.tidy.run/.apply/.reject`
// 同族(`design/23 §19.4`)。**只此一处**允许审计携带内容片段,且
// 只在删除这一动作上(本包其余审计一律只带计数)。
expireAuditVerb = "memory.tidy.expire"
)
// ExpirySweepPolicy 是一轮清扫的口径。零值即默认(由 normalized 填批次上限),
// 因此调用方可以只写 `ExpirySweepPolicy{}`。
//
// 它**没有「关掉清扫」的开关** —— 这是刻意的:临时记忆的到期删除是「声明了生命
// 周期就兑现定义」,留一个能关掉它的开关等于让库里的临时层无界留存(读侧闸只会
// 让它们「看不见」,不会让它们消失)。真要不产生临时记忆,该关的是第五态
// `defer` 的判据,而不是清扫。
type ExpirySweepPolicy struct {
// BatchSize 是单批上限;<=0 取 expireSweepBatchSize。
BatchSize int
// MaxBatches 是一轮的批数上限;<=0 取 maxExpireSweepBatches。
MaxBatches int
}
// normalized 填默认值。负值与非正值一律取默认(配置异常不该让清扫停摆)。
func (p ExpirySweepPolicy) normalized() ExpirySweepPolicy {
if p.BatchSize <= 0 {
p.BatchSize = expireSweepBatchSize
}
if p.MaxBatches <= 0 {
p.MaxBatches = maxExpireSweepBatches
}
return p
}
// ExpirySweepStats 是一轮清扫的汇总(**只记计数,绝不记任何行内容**)。
// 它进调用方的一条汇总日志,不进审计 —— 审计是逐行的那一条(ExpireFunc)。
type ExpirySweepStats struct {
// Deleted 是本轮物理删掉的行数。
Deleted int64
// Restricted 是本轮**跳过**的受限来源行数(群聊 / 陌生人)。它非零只说明库里
// 存在这类历史遗留行,是运维信号,不是错误。
Restricted int
// Batches 是本轮实际执行的取行批数。
Batches int
}
// ExpiredMemory 是一条**即将被物理删除**的到期临时记忆,只带审计需要的字段。
// 它刻意不带 tags / confidence 之类的列:审计口径就是 `design/23 §19.4` 那一行
// `id + persona_id + scope + expires_at + 截断摘要`,多带一列就多一分泄漏面。
type ExpiredMemory struct {
ID string
PersonaID string
Scope string
ExpiresAt string
Content string
}
// ExpireFunc 在**每行被删除之前**调用一次,是本清扫的审计出口
// (`design/23 §19.3-3`:每行删除前先写审计)。它没有返回值:审计出口失败与否
// 都不该改变「这一行删不删」—— 审计是留痕,不是闸。为 nil 时不写审计
// (只有不关心留痕的单元测试会这么用;生产的后台件恒传一个)。
type ExpireFunc func(row ExpiredMemory)
// expireCursor 是清扫的**键集游标**(`(expires_at, rowid)`),让被跳过的受限行
// 只被读到一次:每批处理完就前移到该批最后一行,下一批从它之后取。
//
// 为什么不用 `OFFSET`:受限行会被跳过但不删,用 OFFSET 分页时它们会**反复占住
// 每一批的头部**,极端情况下(头部 500 行都是受限行)真该删的行永远轮不到。
// 键集游标让进度与「删了多少」解耦 —— 只要取到了行,扫描位置就一定前移。
type expireCursor struct {
expiresAt string
rowid int64
}
// expiredMemoryRow 是扫描回来的一行:审计字段 + 受限判定要用的溯源 + 游标。
type expiredMemoryRow struct {
ExpiredMemory
sourceConversation string
cursor expireCursor
}
// selectExpiredMemoriesSQL 按 `design/23 §19.2` 的扫描集合取一批到期行。
//
// 三条谓词各自对应一条边界:`expires_at <> ”` 排除「永不过期」的行(空串是
// 长期 / 日常记忆的默认值,见 store_schema.go 的 DDL),`expires_at <= ?` 取到期
// 的行,`status = ?` 只取在册行。时间戳用定宽 UTC 文本比较:字典序等价于时间序,
// 与写入侧 `formatTime`(timefmt.go)同源 —— 换成 SQLite 的 datetime('now') 会
// 静默判错(timegate.go 有这条反证的实证)。
//
// 游标谓词用 `(expires_at, rowid)` **元组比较**:expires_at 是文本列,同一批
// 落库的行可能同值,只比 expires_at 会让扫描卡在原地。
const selectExpiredMemoriesSQL = `SELECT id, persona_id, scope, expires_at, content,
source_conversation, rowid
FROM memory_items
WHERE status = ? AND expires_at <> '' AND expires_at <= ?
AND (expires_at > ? OR (expires_at = ? AND rowid > ?))
ORDER BY expires_at, rowid
LIMIT ?`
// deleteExpiredMemorySQL 按主键删一行。它**只**由上一条语句选出来的 id 驱动,
// 因此不存在「按用户输入删行」的面。
const deleteExpiredMemorySQL = `DELETE FROM memory_items WHERE id = ?`
// SweepExpiredMemories 跑一轮到期清扫,返回汇总。now 由调用方给出(便于测试注入
// 固定时钟;本包不在 Store 里藏墙钟,与 SweepIngestReviews 同口径)。
//
// 边界(`design/23 §19.3-3`):
//
// - **有界**:一轮最多 maxExpireSweepBatches × expireSweepBatchSize 行,
// 每批一次取行 + 逐行删除,连接池占用与时长都有上界;
// - **有审计**:每行删除前调 onExpire(§19.4 的 `memory.tidy.expire`);
// - **可停**:ctx 取消即刻返回;循环体在每批开跑前复核 ctx.Err()。
//
// 失败语义:取行 / 删行失败时**返回错误并带上已完成的统计**(调用方记 WARN 后等
// 下一轮)。已删的行保持删除、未删的行绝不误删 —— 一轮失败只是「这一轮没扫干净」。
func (s *Store) SweepExpiredMemories(
ctx context.Context, now time.Time, policy ExpirySweepPolicy, onExpire ExpireFunc,
) (ExpirySweepStats, error) {
var stats ExpirySweepStats
if s == nil {
return stats, ErrNilStore
}
p := policy.normalized()
// 与写入侧同源:列里存的是 formatTime 产出的定宽 UTC 文本,右边也必须由
// 同一个函数产出(见 timegate.go 的「同源」口径)。
cutoff := formatTime(now.UTC())
cursor := expireCursor{rowid: -1}
for batch := 0; batch < p.MaxBatches; batch++ {
if err := ctx.Err(); err != nil {
return stats, fmt.Errorf("memory: sweep expired memories: %w", err)
}
rows, err := s.listExpiredMemories(ctx, cutoff, cursor, p.BatchSize)
if err != nil {
return stats, err
}
if len(rows) == 0 {
break
}
stats.Batches++
// 游标前移与「删没删」无关:受限行被跳过时也必须往前走,否则下一批会
// 取回同一批行(见 expireCursor 的注释)。
cursor = rows[len(rows)-1].cursor
for _, row := range rows {
if nonOwnerSource(row.sourceConversation) {
stats.Restricted++
continue
}
if onExpire != nil {
onExpire(row.ExpiredMemory)
}
if err := s.purgeExpiredMemory(ctx, row.ID); err != nil {
return stats, err
}
stats.Deleted++
}
if len(rows) < p.BatchSize {
break
}
}
return stats, nil
}
// listExpiredMemories 取一批到期行(含游标与溯源列)。limit 非正时返回空切片,
// 因此调用方不必自己防零。
//
// 它把整批**读进内存后才返回**(游标在返回前关闭),所以调用方的删除不会与一个
// 还开着的 SELECT 游标打架 —— 本项目的 SQLite 池是单连接,同一连接上「边读边删」
// 是自寻死路(写要等读,读永远读不完)。
func (s *Store) listExpiredMemories(
ctx context.Context, cutoff string, cursor expireCursor, limit int,
) ([]expiredMemoryRow, error) {
if limit <= 0 {
return nil, nil
}
rows, err := s.db.QueryContext(ctx, selectExpiredMemoriesSQL,
string(tidyStatusActive), cutoff, cursor.expiresAt, cursor.expiresAt, cursor.rowid, limit)
if err != nil {
return nil, fmt.Errorf("memory: list expired items: %w", err)
}
defer func() { _ = rows.Close() }()
out := make([]expiredMemoryRow, 0, limit)
for rows.Next() {
var r expiredMemoryRow
if err := rows.Scan(&r.ID, &r.PersonaID, &r.Scope, &r.ExpiresAt, &r.Content,
&r.sourceConversation, &r.cursor.rowid); err != nil {
return nil, fmt.Errorf("memory: scan expired item: %w", err)
}
r.cursor.expiresAt = r.ExpiresAt
out = append(out, r)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("memory: iterate expired items: %w", err)
}
return out, nil
}
// purgeExpiredMemory 删掉一行到期临时记忆,并把它的向量从持久索引里同步清掉。
//
// 两步的顺序不是随手写的:**先删行、再清向量**。反过来(先清向量)的话,删行
// 失败会留下一行「有行、无向量」的记忆 —— 那一行不会被返回(读侧闸挡着),
// 但检索的语义路径会有一个查不到向量的 id;而先删行时最坏情况只是留一个孤儿
// 向量,那是本 finding 描述的既有状态(可下一轮再清),不会更坏。
func (s *Store) purgeExpiredMemory(ctx context.Context, id string) error {
if _, err := s.db.ExecContext(ctx, deleteExpiredMemorySQL, id); err != nil {
return fmt.Errorf("memory: delete expired item %s: %w", id, err)
}
s.purgeExpiredVector(ctx, id)
return nil
}
// purgeExpiredVector 把刚被删除的行从持久语义索引里同步清掉(切片 M1 的 reviewer
// 交上来的 finding):过期行的向量若留在索引里,会占掉 `idx.Search` 的 top-K
// 预算,只降召回质量、不泄漏(该行已被读侧时效闸挡住,不可能被返回)。
//
// 未接索引、或后端不支持删除(只读的桩)时是空操作;失败只记 WARN ——
// **向量清不掉不该让「行已删」这件事失败**,也不该让下一行不删。
func (s *Store) purgeExpiredVector(ctx context.Context, id string) {
idx := s.vectorIndex()
if idx == nil {
return
}
deleter, ok := idx.(vector.Deleter)
if !ok {
s.log.Debug("memory: vector index does not support delete, expired vector left behind", "id", id)
return
}
if err := deleter.Delete(ctx, id); err != nil {
s.log.Warn("memory: expired vector purge failed", "id", id, "error", err)
}
}
// expireAuditResult 渲染 `design/23 §19.4` 要求的定长结构化串:
//
// id=<id> persona_id=<..> scope=<..> expires_at=<..> summary=<截断摘要>
//
// 摘要是**本仓唯一**允许进入审计的内容片段,边界限这一个动作 + 固定截断
// (主会话已裁定允许)。它先做空白归一(正文里的换行 / 制表符会把审计行拆成
// 多行,那既破坏「一行一条」的可解析性,也把事情变成「把正文刷进日志」),
// 再按 rune 截断。
func expireAuditResult(row ExpiredMemory) string {
return fmt.Sprintf("id=%s persona_id=%s scope=%s expires_at=%s summary=%s",
row.ID, row.PersonaID, row.Scope, row.ExpiresAt, clipExpireSummary(row.Content))
}
// clipExpireSummary 把正文压成一行至多 expireAuditSummaryRunes 个 rune 的摘要。
// 空白(含换行)一律归一成单个空格;截断时追加省略号,让「这条被截过」在审计
// 里可见 —— 省略号本身也占一个 rune,总长仍恒为 ≤ 上限(上限-1 + 省略号)。
//
// 空正文(理论上不该存在,正文是 NOT NULL 且入库时校验非空)返回空串,
// 让审计行仍可解析。
func clipExpireSummary(content string) string {
flat := strings.Join(strings.Fields(content), " ")
runes := []rune(flat)
if len(runes) <= expireAuditSummaryRunes {
return flat
}
return string(runes[:expireAuditSummaryRunes-1]) + "…"
}
+124
View File
@@ -0,0 +1,124 @@
package memory
import (
"context"
"sync"
"time"
)
// 本文件是到期清扫的**后台件形态**:一个独立循环(与 `preinsert_gc.go` 的
// `sweepLoop` 同形),启动期先扫一次、此后每 DefaultExpireSweepInterval 扫一次。
//
// 设计口径(`design/23 §19.3-2`):删除只由内存模块的后台件串行触发,与整理 /
// 审查同一单写者纪律,**不新开并发写口**、不新开监听端口、不加任何 `.env` 键
// (`.env` 已冻结,`design/13 §2.1`),也不碰会话 / 历史读数。
//
// 🔴 **挂在哪个 ctx 上**:清扫与整理**共用同一条生命周期** —— 它的 ctx 就是整理
// 心跳的 ctx,它的 goroutine 登记进 TidyService 的 `wg`。因此
// `cmd/cyrene/wire_services.go` 关闭链里的 `a.memoryTidy.Stop(ctx)` 会一并取消它
// 并等它退出(§19.3-3 的「可停」),不必再为清扫加一条关闭路径、也不必改
// `tidy_service.go`(那个文件归别的切片)。
//
// 挂点用的是 TidyService 已有的注入缝 `wait`(心跳的等待原语,见
// tidy_service.go:64-66):它由心跳 goroutine 以**心跳的 cctx** 调用,因此「第一次
// wait 被调用」既是「Start 已经跑起来、ctx 已经建好」的确定信号,又恰好把清扫要的
// 那个可取消 ctx 递到手上。清扫随后跑在**自己的** goroutine 与**自己的**小时级
// ticker 上,不借用整理的调度节奏(整理的周期是 24h 档,不能拿来当清扫周期)。
// AttachExpirySweep 把到期清扫挂到整理服务的生命周期上。它由组合根在装配期调用
// (`cmd/cyrene/wire_memory_tidy.go`),每个服务实例调用一次。
//
// 它在**第一次心跳等待**时启动清扫循环:此时 `Start` 已经建好可取消的 ctx,
// 而那个 ctx 就是心跳 goroutine 手里这一个。ctx 每次变化(Start → Stop → Start)
// 都会重新挂一个循环,因此「重启一次就永久失去清扫」不会发生。
//
// 它不启动任何 goroutine 直到那一次等待发生:整理被配置关掉(`Start` 直接返回)
// 时,清扫也不会跑 —— 这与 `preinsert_gc.go` 的清扫随审查服务启停是同一条口径
// (后台件随它的宿主启停,而不是凭空常驻)。
func (s *TidyService) AttachExpirySweep() {
if s == nil {
return
}
inner := s.wait
if inner == nil {
inner = waitTidyTimer
}
var (
mu sync.Mutex
attached context.Context
)
s.wait = func(ctx context.Context, d time.Duration) bool {
mu.Lock()
if attached != ctx {
attached = ctx
s.wg.Add(1)
go s.expirySweepLoop(ctx)
}
mu.Unlock()
return inner(ctx, d)
}
}
// expirySweepLoop 是清扫的常驻循环:启动期先扫一次(清掉上次退出后到期的行),
// 此后按 DefaultExpireSweepInterval 扫一次。它随 ctx 取消退出(Stop 会 join 它)。
//
// 它**不**读任何记忆正文之外的输入:清扫只按 expires_at / status / 溯源列删行。
func (s *TidyService) expirySweepLoop(ctx context.Context) {
defer s.wg.Done()
s.expirySweepOnce(ctx)
ticker := time.NewTicker(DefaultExpireSweepInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
s.expirySweepOnce(ctx)
}
}
}
// expirySweepOnce 跑一轮到期清扫。失败只记 WARN:清扫是旁路维护,绝不让整理服务
// 退出、也绝不影响任何一条**未到期**的记忆(与 `preinsert_gc.go` 的 fail-safe
// 同口径:已删的保持删除、未删的留给下一轮)。
//
// 日志口径(`design/23 §19.4`):**一轮只写一行汇总**,逐条的那一行进审计出口
// (见 auditExpiredMemory)。这里绝不逐条把正文刷进日志。
func (s *TidyService) expirySweepOnce(ctx context.Context) {
store := s.tidier.deps.Store
if store == nil {
return
}
stats, err := store.SweepExpiredMemories(ctx, s.now(), ExpirySweepPolicy{}, s.auditExpiredMemory)
if err != nil {
// ctx 取消(关闭中)不是故障:安静退出,不制造一条误导性的 WARN。
if ctx.Err() != nil {
return
}
s.log.Warn("memory expiry sweep failed, expired rows left for the next round",
"error", err, "deleted", stats.Deleted, "restricted_skipped", stats.Restricted)
return
}
if stats.Deleted == 0 && stats.Restricted == 0 {
s.log.Debug("memory expiry sweep: nothing expired")
return
}
s.log.Info("memory expiry sweep",
"deleted", stats.Deleted, "restricted_skipped", stats.Restricted, "batches", stats.Batches,
"interval", DefaultExpireSweepInterval.String())
}
// auditExpiredMemory 是每行删除前的审计出口(`design/23 §19.4`):
// `who = system/memorytidy`、`what = memory.tidy.expire`,result 见 expireAuditResult。
//
// 没有审计出口时**只记 id,不记摘要**:摘要仍是内容片段,把它逐条刷进日志正是
// 本设计要避免的那件事(「一轮汇总只进日志」)。
func (s *TidyService) auditExpiredMemory(row ExpiredMemory) {
hook := s.tidier.deps.Audit
if hook == nil {
s.log.Debug("memory expiry audit dropped: no sink configured",
"what", expireAuditVerb, "id", row.ID)
return
}
hook(tidyAuditWho, expireAuditVerb, expireAuditResult(row))
}
+216
View File
@@ -0,0 +1,216 @@
package memory
import (
"bytes"
"context"
"log/slog"
"strings"
"sync"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector"
)
// 本文件锁定期清扫的**接线与后台形态**(design/23 §19):挂点(AttachExpirySweep)、
// 随 Stop 可停、一轮只写一行汇总日志,以及「删行时同步清掉向量」这条 finding 的
// 集成断言。清扫自身的边界(哪些行删、哪些行留、审计长什么样)在 expire_test.go。
// TestExpirySweepOnce_一轮只写一行汇总:逐条的那一行进审计,日志里一轮只有一行
// (§19.4「一轮汇总只进日志,绝不逐条把正文刷进日志」)。
func TestExpirySweepOnce_一轮只写一行汇总(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
var buf bytes.Buffer
var mu sync.Mutex
audit := func(who, what, result string) {
mu.Lock()
defer mu.Unlock()
_ = who
_ = what
_ = result
}
svc, err := NewTidyService(tidyTestCfg(0.80), TidyDeps{
Store: store, Log: slog.New(slog.NewTextHandler(&buf, nil)),
Now: fixedTidyClock(now), Audit: audit,
})
if err != nil {
t.Fatalf("NewTidyService: %v", err)
}
for i := 0; i < 3; i++ {
addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "日志里不该出现的正文片段", ExpiresAt: now.Add(-time.Minute),
})
}
svc.expirySweepOnce(ctx)
lines := strings.Split(strings.TrimSpace(buf.String()), "\n")
if len(lines) != 1 {
t.Fatalf("一轮写入了 %d 行日志,want 1(只允许一行汇总):\n%s", len(lines), buf.String())
}
if !strings.Contains(lines[0], "deleted=3") {
t.Fatalf("汇总行没有计数:%q", lines[0])
}
if strings.Contains(buf.String(), "日志里不该出现的正文片段") {
t.Fatalf("日志里出现了正文 —— 逐条正文只许进审计的截断摘要:%q", buf.String())
}
}
// TestAttachExpirySweep_挂上即扫_随Stop可停 钉住**接线**:清扫挂到整理服务的
// 心跳 ctx 上(AttachExpirySweep),第一次心跳等待就启动、启动即扫一轮;ctx 取消
// 后它的 goroutine 必须退出 —— `Stop` 走的就是「cancel + wg.Wait」,这里用同一个
// 手法验证「可停」(§19.3-3)。
func TestAttachExpirySweep_挂上即扫_随Stop可停(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
row := addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "在心跳上被扫掉的到期行", ExpiresAt: now.Add(-time.Minute),
})
var (
mu sync.Mutex
audits []string
)
svc := newTidyServiceForTest(t, store, tidyTestCfg(0.80), nil, fixedTidyClock(now),
func(who, what, result string) {
mu.Lock()
defer mu.Unlock()
audits = append(audits, what+" "+result)
})
// 假等待:不真等(真等会让用例挂 24h)。它只用来模拟「心跳调用了一次 wait」。
svc.wait = func(context.Context, time.Duration) bool { return true }
svc.AttachExpirySweep()
runCtx, cancel := context.WithCancel(ctx)
defer cancel()
if !svc.wait(runCtx, time.Hour) {
t.Fatal("包装后的 wait 必须是透传的(假等待返回 true,包装必须照原样返回)")
}
waitUntilRowGone(t, store, row.ID)
cancel()
done := make(chan struct{})
go func() {
svc.wg.Wait()
close(done)
}()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("清扫 goroutine 没有随 ctx 取消退出 —— 「可停」不成立(Stop 会卡在 wg.Wait)")
}
mu.Lock()
defer mu.Unlock()
if len(audits) != 1 || !strings.HasPrefix(audits[0], expireAuditVerb+" ") {
t.Fatalf("审计 = %v, want 一条 what=%s", audits, expireAuditVerb)
}
}
// TestSweepExpiredMemories_同步清掉向量 是切片 M1 的 reviewer 交上来的那条 finding
// 的断言:过期行被物理删除时,它的**向量也必须一起清掉** —— 否则向量留在索引里
// 占掉 idx.Search 的 top-K 预算,召回质量静默下降(行本身已被读侧闸挡住,不会泄漏)。
func TestSweepExpiredMemories_同步清掉向量(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
idx := newTestVectorIndex(t, 64)
store.attachVectorIndex(idx)
now := tidyTestTime(t)
expired := addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "到期的向量条目", ExpiresAt: now.Add(-time.Minute),
})
kept := addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "长期留着的向量条目",
})
if !vectorIndexHas(t, ctx, idx, expired.ID) {
t.Fatal("前置条件不成立:到期行的向量本来就该在索引里(写入侧没喂进去)")
}
if !vectorIndexHas(t, ctx, idx, kept.ID) {
t.Fatal("前置条件不成立:保留行的向量本来就该在索引里")
}
if _, err := store.SweepExpiredMemories(ctx, now, ExpirySweepPolicy{}, nil); err != nil {
t.Fatalf("SweepExpiredMemories: %v", err)
}
if vectorIndexHas(t, ctx, idx, expired.ID) {
t.Fatal("到期行删了、向量还在索引里 —— 它会占掉 top-K 预算(M1 reviewer 的 finding)")
}
if !vectorIndexHas(t, ctx, idx, kept.ID) {
t.Fatal("保留行的向量被误清了(清算成了通杀)")
}
}
// vectorIndexHas 用 n-gram 索引自带的 embedder 查一次,报告某 id 是否还在索引里。
func vectorIndexHas(t *testing.T, ctx context.Context, idx vector.Index, id string) bool {
t.Helper()
vecs, err := idx.Embed(ctx, []string{"到期的向量条目 长期留着的向量条目"})
if err != nil {
t.Fatalf("embed query: %v", err)
}
hits, err := idx.Search(ctx, vecs[0], 50)
if err != nil {
t.Fatalf("search index: %v", err)
}
for _, h := range hits {
if h.ID == id {
return true
}
}
return false
}
// TestClipExpireSummary 单钉摘要裁剪这件小事:空白归一(含换行)、按 rune 截断、
// 截断才加省略号。
func TestClipExpireSummary(t *testing.T) {
t.Parallel()
tests := []struct {
name string
in string
want string
}{
{name: "短文本原样", in: "谢谢啦", want: "谢谢啦"},
{name: "空白归一", in: "一\n\t二 三", want: "一 二 三"},
{name: "空文本", in: " ", want: ""},
{
name: "超长按 rune 截断并省略号",
in: strings.Repeat("很", expireAuditSummaryRunes+10),
want: strings.Repeat("很", expireAuditSummaryRunes-1) + "…",
},
}
for _, tc := range tests {
tc := tc
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
if got := clipExpireSummary(tc.in); got != tc.want {
t.Fatalf("clipExpireSummary(%q) = %q, want %q", tc.in, got, tc.want)
}
})
}
}
// waitUntilRowGone 轮询等一行被删掉(清扫跑在另一个 goroutine 上)。
func waitUntilRowGone(t *testing.T, store *Store, id string) {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
if !memoryRowExists(t, store, id) {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("等不到行 %s 被清扫删除(清扫没跑起来?)", id)
}
+311
View File
@@ -0,0 +1,311 @@
package memory
import (
"context"
"errors"
"strings"
"testing"
"time"
"unicode/utf8"
)
// 本文件锁定**临时期限到期处置的第 ② / ③ 步**(design/23 §19,切片 M3):
// 到期行的物理删除、删前审计、以及四条边界。第 ① 步(读侧隐藏)由
// timegate_test.go 覆盖,本文件只断言「真删掉」这一半。
//
// 四条边界逐条有断言,因为它们各自对应一次「删错就再也回不来」的事故:
//
// ① 空 `expires_at`(长期 / 日常记忆,含 M2 那条 INSERT-与-补写之间的中断窗口)
// **绝不删** —— 判据是 SQL 里的 `expires_at <> ''`;
// ② 未到期 **绝不删**;
// ③ 受限来源(群聊 / 陌生人)**不参与**,且不为它写审计;
// ④ 审计带的是**按 rune 截断的摘要**,绝不是正文全文(本仓唯一一处允许携带
// 内容片段的审计,边界限 `memory.tidy.expire` 一个动作)。
// memoryRowExists 报告一条记忆行是否还在。Store.Get 刻意不过任何读侧闸
// (store.go),因此它是「行在不在」的直接探针:过期行与已软删行都读得到。
func memoryRowExists(t *testing.T, store *Store, id string) bool {
t.Helper()
if _, err := store.Get(context.Background(), id); errors.Is(err, ErrNotFound) {
return false
} else if err != nil {
t.Fatalf("Get(%s): %v", id, err)
}
return true
}
// expiryFixture 建三条典型行:一条已到期、一条未到期、一条**不设过期**
// (`expires_at=”`,即长期 / 日常记忆的形态)。
type expiryFixture struct {
expired MemoryItem
future MemoryItem
longTerm MemoryItem
}
func seedExpiryFixture(t *testing.T, store *Store, now time.Time) expiryFixture {
t.Helper()
base := func() MemoryItem {
return MemoryItem{PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship}
}
expired := base()
expired.Content = "临时记忆:明天要去买咖啡豆"
expired.ExpiresAt = now.Add(-time.Hour)
future := base()
future.Content = "临时记忆:还没到期的那条"
future.ExpiresAt = now.Add(time.Hour)
longTerm := base()
longTerm.Content = "长期记忆:主人喜欢冰美式"
return expiryFixture{
expired: addItem(t, store, expired),
future: addItem(t, store, future),
longTerm: addItem(t, store, longTerm),
}
}
// TestSweepExpiredMemories_过期删_未到期与长期保留 是设计 §19.2 扫描集合的正例与
// 两条反例:只有「非空且已过去的 expires_at」的行被删,未到期行与不设过期的行
// 一律保留。
func TestSweepExpiredMemories_过期删_未到期与长期保留(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
fx := seedExpiryFixture(t, store, now)
var audited []ExpiredMemory
stats, err := store.SweepExpiredMemories(ctx, now, ExpirySweepPolicy{}, func(row ExpiredMemory) {
audited = append(audited, row)
})
if err != nil {
t.Fatalf("SweepExpiredMemories: %v", err)
}
if stats.Deleted != 1 {
t.Fatalf("deleted = %d, want 1(只有那一条到期的该删)", stats.Deleted)
}
if memoryRowExists(t, store, fx.expired.ID) {
t.Fatal("到期行没有被物理删除 —— 读侧闸只负责「看不见」,删除是这一步的事")
}
if !memoryRowExists(t, store, fx.future.ID) {
t.Fatal("未到期行被删了(扫描集合写错)")
}
if !memoryRowExists(t, store, fx.longTerm.ID) {
t.Fatal("空 expires_at(= 不设过期)的行被删了 —— 长期 / 日常记忆必须结构性进不了删除集合")
}
if len(audited) != 1 || audited[0].ID != fx.expired.ID {
t.Fatalf("审计行 = %+v, want 只这一条到期行 %s", audited, fx.expired.ID)
}
}
// TestSweepExpiredMemories_空expires_at绝不删 是**单独**钉住那条判据的用例:
// 它是 M2 登记的「INSERT 与补写 expires_at 之间若进程中断,该行会以 expires_at=”
// 留存」那条窗口的护栏 —— 方向是「宁可多留」,因此那种行**永远**不该被清扫删掉。
//
// 断言刻意写成「哪怕它是 active 且 created_at 很旧」也一样保留:清扫不看创建时间。
func TestSweepExpiredMemories_空expires_at绝不删(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
orphan := addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "进程在补写 expires_at 之前中断留下的行",
// 刻意落在很早以前,证明清扫不看 created_at。
CreatedAt: now.Add(-1000 * time.Hour),
})
stats, err := store.SweepExpiredMemories(ctx, now, ExpirySweepPolicy{}, nil)
if err != nil {
t.Fatalf("SweepExpiredMemories: %v", err)
}
if stats.Deleted != 0 {
t.Fatalf("deleted = %d, want 0(空 expires_at = 永不过期)", stats.Deleted)
}
if !memoryRowExists(t, store, orphan.ID) {
t.Fatal("空 expires_at 的行被删了 —— 那是「永不过期」,不是「过期很久」")
}
}
// TestSweepExpiredMemories_受限来源保留 是 §19.2 的第三条边界:群聊 / 陌生人来
// 源的临时记忆**不参与**到期删除(判据复用 nonOwnerSource,与整理 / 审查同一处
// 实现),且**不为它写审计** —— 删都没删,凭什么留一条「删了」的痕。
func TestSweepExpiredMemories_受限来源保留(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
restricted := addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "群聊里冒出来的一条(历史遗留)",
SourceConversation: FormatSourceConversation("sess-telegram:group1", "m1"),
ExpiresAt: now.Add(-time.Hour),
})
owner := addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "主人私聊里那条到期的",
SourceConversation: FormatSourceConversation("web", "m2"),
ExpiresAt: now.Add(-time.Hour),
})
var audited []string
stats, err := store.SweepExpiredMemories(ctx, now, ExpirySweepPolicy{}, func(row ExpiredMemory) {
audited = append(audited, row.ID)
})
if err != nil {
t.Fatalf("SweepExpiredMemories: %v", err)
}
if stats.Restricted != 1 {
t.Fatalf("restricted = %d, want 1(受限来源要被识别并跳过)", stats.Restricted)
}
if stats.Deleted != 1 {
t.Fatalf("deleted = %d, want 1(只删主人链路那条)", stats.Deleted)
}
if !memoryRowExists(t, store, restricted.ID) {
t.Fatal("受限来源(群聊)的到期行被删了 —— §19.2 要求它不参与到期删除")
}
if memoryRowExists(t, store, owner.ID) {
t.Fatal("主人链路的到期行没有被删(受限判据把正常行也误伤了)")
}
if len(audited) != 1 || audited[0] != owner.ID {
t.Fatalf("审计 = %v, want 只有 %s(受限行不该留「删了」的痕)", audited, owner.ID)
}
}
// TestSweepExpiredMemories_只删在册行:已软删 / 已归档的过期行不参与 —— 它们不是
// 「到期的临时记忆」,是整理软动作留下的行,该不该清是另一件事(§19.2 第一条)。
func TestSweepExpiredMemories_只删在册行(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
merged := addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "被软合并掉的过期行", ExpiresAt: now.Add(-time.Hour),
})
if _, err := store.db.ExecContext(ctx,
`UPDATE memory_items SET status = ? WHERE id = ?`, ingestStatusMerged, merged.ID); err != nil {
t.Fatalf("update status: %v", err)
}
stats, err := store.SweepExpiredMemories(ctx, now, ExpirySweepPolicy{}, nil)
if err != nil {
t.Fatalf("SweepExpiredMemories: %v", err)
}
if stats.Deleted != 0 {
t.Fatalf("deleted = %d, want 0(非在册行不参与)", stats.Deleted)
}
if !memoryRowExists(t, store, merged.ID) {
t.Fatal("已软删(status='merged')的行被到期清扫删了")
}
}
// TestSweepExpiredMemories_有界批次:一轮清扫的批数 × 单批上限是有界的,积压时
// 剩下的留给下一轮 —— 绝不「一次 DELETE 全表」(§19.3-3 的有界)。
func TestSweepExpiredMemories_有界批次(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
const rows = 5
ids := make([]string, 0, rows)
for i := 0; i < rows; i++ {
ids = append(ids, addItem(t, store, MemoryItem{
PersonaID: "persona.a", Scope: ScopePersona, Kind: KindRelationship,
Content: "过期行", ExpiresAt: now.Add(-time.Duration(i+1) * time.Minute),
}).ID)
}
policy := ExpirySweepPolicy{BatchSize: 2, MaxBatches: 2}
first, err := store.SweepExpiredMemories(ctx, now, policy, nil)
if err != nil {
t.Fatalf("first sweep: %v", err)
}
if first.Deleted != 2*2 || first.Batches != 2 {
t.Fatalf("一轮 = deleted %d / batches %d, want 4 / 2(有界:2 批 × 每批 2 行)",
first.Deleted, first.Batches)
}
second, err := store.SweepExpiredMemories(ctx, now, policy, nil)
if err != nil {
t.Fatalf("second sweep: %v", err)
}
if second.Deleted != 1 {
t.Fatalf("第二轮 deleted = %d, want 1(剩下的留给下一轮)", second.Deleted)
}
for _, id := range ids {
if memoryRowExists(t, store, id) {
t.Fatalf("两轮之后仍有到期行 %s 未删(有界不该变成永远删不完)", id)
}
}
}
// TestSweepExpiredMemories_审计在删之前且带截断摘要 是 §19.4 的核心断言:
//
// 1. **删前留痕**:审计回调被调用时那一行**还在**(顺序反了就事后无从追责);
// 2. **摘要按 rune 截断**:不超过 expireAuditSummaryRunes,且截断处留省略号;
// 3. **绝不含正文全文**:审计里找不到那条长正文;
// 4. 正文里的换行被归一成空格(审计行恒为一行,可解析)。
func TestSweepExpiredMemories_审计在删之前且带截断摘要(t *testing.T) {
t.Parallel()
ctx := context.Background()
store := newTestStore(t)
now := tidyTestTime(t)
longText := strings.Repeat("很长的临时记忆正文", 30) // 300 rune
row := addItem(t, store, MemoryItem{
PersonaID: "persona.b", Scope: ScopeProject, Kind: KindWorldEvent,
Content: "第一行\n\t第二行 " + longText,
ExpiresAt: now.Add(-time.Minute),
})
var (
results []string
existedAtAudit []bool
)
_, err := store.SweepExpiredMemories(ctx, now, ExpirySweepPolicy{}, func(r ExpiredMemory) {
existedAtAudit = append(existedAtAudit, memoryRowExists(t, store, r.ID))
results = append(results, expireAuditResult(r))
})
if err != nil {
t.Fatalf("SweepExpiredMemories: %v", err)
}
if len(results) != 1 {
t.Fatalf("审计行数 = %d, want 1", len(results))
}
if !existedAtAudit[0] {
t.Fatal("审计发生时那一行已经被删了 —— §19.3-3 要求「每行删除**前**先写审计」")
}
got := results[0]
if !strings.HasPrefix(got, "id="+row.ID+" persona_id=persona.b scope=project expires_at=") {
t.Fatalf("审计串缺字段:%q", got)
}
if strings.Contains(got, "\n") || strings.Contains(got, "\t") {
t.Fatalf("审计带换行 / 制表符,会破坏「一行一条」:%q", got)
}
if strings.Contains(got, longText) {
t.Fatalf("审计里出现了正文全文 —— 摘要必须被截断:%q", got)
}
_, summary, ok := strings.Cut(got, "summary=")
if !ok {
t.Fatalf("审计串没有 summary 段:%q", got)
}
if n := utf8.RuneCountInString(summary); n > expireAuditSummaryRunes {
t.Fatalf("摘要 %d rune,超过上限 %d:%q", n, expireAuditSummaryRunes, summary)
}
if !strings.HasSuffix(summary, "…") {
t.Fatalf("截断处应当留省略号(让人一眼看出被截过):%q", summary)
}
// 摘要必须是**归一化之后**正文的前缀(证明它是「截断」而不是别的东西)。
flat := strings.Join(strings.Fields(row.Content), " ")
if !strings.HasPrefix(flat, strings.TrimSuffix(summary, "…")) {
t.Fatalf("摘要不是(空白归一后的)正文前缀:%q", summary)
}
}
+90
View File
@@ -0,0 +1,90 @@
package vector
import (
"context"
"fmt"
"strings"
)
// 本文件是**向量删除**(索引的清除侧)。动因是一条实测 finding(切片 M1 的
// 读侧时效闸交下来的):临时记忆到期后**行不在了、向量还在** —— 过期行的向量
// 仍留在 vec_items 里,占掉 idx.Search 的 top-K 预算,于是「语义检索前 K 名」
// 里有一半是永远不会被返回的死条目(读侧闸已挡住那些行),**召回质量静默下降**。
// 到期清扫(internal/runtime/memory/expire.go)在物理删行时同步调这里,把
// 向量一并清掉。
//
// 🔴 **为什么不是 Index 的必需方法**(Deleter 与 Index 并列、由调用方类型断言):
// Index 是本包对外的稠密索引抽象,加一个必需方法会**立刻打破所有既有实现** ——
// 包括别的包里的测试桩(例如 memory 包的 mockVectorIndex)。清除能力因此按
// **可选扩展**建模:两种生产后端(SQLiteIndex / PGVectorIndex)都实现它,
// 只读的桩可以不实现,调用方断言失败时退化为「不清向量」并留一条日志。
// 这与本仓既有的可选能力手法一致(例如 `a.vindex.(*vector.PGVectorIndex)`)。
// Deleter 是 Index 的**可选扩展**:把已不存在的 id 从索引里清掉。id 不存在时
// 视为成功(删不存在的向量本就是空操作),因此重复调用是幂等的。
//
// 调用方(memory 包)在物理删行之后调用它;**不是**读路径的一部分。
type Deleter interface {
Delete(ctx context.Context, ids ...string) error
}
// deleteVecItemsSQLTemplate 是两种后端共用的删除语句骨架:`%s` 是占位串
// (SQLite `?` / pgvector `$1,$2,...`)。表名是编译期常量,绝不来自输入。
const deleteVecItemsSQLTemplate = `DELETE FROM vec_items WHERE id IN (%s)`
// vecDeletePlaceholders 为 ids 生成删除语句的占位串与参数表。`dollar` 为 true
// 时用 Postgres 的 `$n` 形式,否则用 SQLite 的 `?`。
//
// 空白 id 被**跳过**而不是报错:删一个不存在的向量是空操作,不该让整批失败
// (Add 拒绝空 id,是因为那会写出一行不可寻址的数据;删除没有这个问题)。
// 全部为空时返回空占位串,调用方据此直接返回。
func vecDeletePlaceholders(ids []string, dollar bool) (string, []any) {
args := make([]any, 0, len(ids))
for _, id := range ids {
if strings.TrimSpace(id) == "" {
continue
}
args = append(args, id)
}
if len(args) == 0 {
return "", nil
}
var b strings.Builder
for i := range args {
if i > 0 {
b.WriteByte(',')
}
if dollar {
fmt.Fprintf(&b, "$%d", i+1)
} else {
b.WriteByte('?')
}
}
return b.String(), args
}
// Delete 把 ids 对应的向量从 SQLite 索引里删掉。id 不存在时无事发生(幂等);
// 空 ids 是空操作。ctx 被取消时中止。
func (s *SQLiteIndex) Delete(ctx context.Context, ids ...string) error {
placeholders, args := vecDeletePlaceholders(ids, false)
if placeholders == "" {
return nil
}
if _, err := s.db.ExecContext(ctx, fmt.Sprintf(deleteVecItemsSQLTemplate, placeholders), args...); err != nil {
return fmt.Errorf("memory/vector: delete %d sqlite vectors: %w", len(args), err)
}
return nil
}
// Delete 把 ids 对应的向量从 pgvector 索引里删掉。语义与 SQLiteIndex.Delete
// 逐字相同(幂等、空 ids 空操作、ctx 取消即中止),只用 `$n` 占位。
func (p *PGVectorIndex) Delete(ctx context.Context, ids ...string) error {
placeholders, args := vecDeletePlaceholders(ids, true)
if placeholders == "" {
return nil
}
if _, err := p.db.ExecContext(ctx, fmt.Sprintf(deleteVecItemsSQLTemplate, placeholders), args...); err != nil {
return fmt.Errorf("memory/vector: delete %d pg vectors: %w", len(args), err)
}
return nil
}
@@ -0,0 +1,184 @@
package vector
import (
"context"
"os"
"testing"
)
// 本文件锁定向量的**清除侧**(Deleter):到期清扫删行时同步删向量,否则过期
// 向量会占掉 idx.Search 的 top-K 预算(切片 M1 的 reviewer 交上来的 finding)。
// TestSQLite_Delete_正常 覆盖删除的正例与三条边界:删掉的查不到、没删的还在、
// 删不存在的 id 幂等成功、空 ids 空操作。
func TestSQLite_Delete_正常(t *testing.T) {
t.Parallel()
ctx := context.Background()
s := newInMemorySQLite(t, 3)
for _, id := range []string{"cat", "dog", "bird"} {
if err := s.Add(ctx, id, Vector{1, 0, 0}); err != nil {
t.Fatalf("Add(%s): %v", id, err)
}
}
if err := s.Delete(ctx, "cat"); err != nil {
t.Fatalf("Delete(cat): %v", err)
}
hits, err := s.Search(ctx, Vector{1, 0, 0}, 10)
if err != nil {
t.Fatalf("Search: %v", err)
}
for _, h := range hits {
if h.ID == "cat" {
t.Fatal("被删的向量还在索引里")
}
}
if len(hits) != 2 {
t.Fatalf("剩余命中 = %d, want 2(只该删掉那一条)", len(hits))
}
// 幂等:再删一次同一个 id 不报错(删不存在的行本就是空操作)。
if err := s.Delete(ctx, "cat"); err != nil {
t.Fatalf("重复 Delete 应当幂等成功: %v", err)
}
// 空 ids / 全是空白的 id:空操作,不报错。
if err := s.Delete(ctx); err != nil {
t.Fatalf("Delete() 空 ids 应当是无操作: %v", err)
}
if err := s.Delete(ctx, " "); err != nil {
t.Fatalf("Delete(空白 id) 应当是无操作: %v", err)
}
// 批删:一次给两个 id 走的是同一条占位串路径。
if err := s.Delete(ctx, "dog", "bird"); err != nil {
t.Fatalf("批量 Delete: %v", err)
}
hits, err = s.Search(ctx, Vector{1, 0, 0}, 10)
if err != nil {
t.Fatalf("Search after batch delete: %v", err)
}
if len(hits) != 0 {
t.Fatalf("批量删后仍有 %d 条命中", len(hits))
}
}
// TestSQLite_Delete_取消的ctx_异常:ctx 已取消时删除直接报错,绝不静默成功。
func TestSQLite_Delete_取消的ctx_异常(t *testing.T) {
t.Parallel()
s := newInMemorySQLite(t, 3)
ctx, cancel := context.WithCancel(context.Background())
cancel()
if err := s.Delete(ctx, "cat"); err == nil {
t.Fatal("已取消的 ctx 下 Delete 应当报错")
}
}
// TestVecDeletePlaceholders 钉住占位串与参数表的形状:SQLite 用 `?`、pgvector
// 用 `$n`,空白 id 被跳过,全空白时返回空占位串。
func TestVecDeletePlaceholders(t *testing.T) {
t.Parallel()
tests := []struct {
name string
ids []string
dollar bool
wantSQL string
wantN int
}{
{name: "sqlite 单个", ids: []string{"a"}, wantSQL: "?", wantN: 1},
{name: "sqlite 三个", ids: []string{"a", "b", "c"}, wantSQL: "?,?,?", wantN: 3},
{name: "pg 两个", ids: []string{"a", "b"}, dollar: true, wantSQL: "$1,$2", wantN: 2},
{name: "跳过空白", ids: []string{"a", "", " ", "b"}, dollar: true, wantSQL: "$1,$2", wantN: 2},
{name: "全空", ids: []string{"", " "}, wantSQL: "", wantN: 0},
{name: "无参数", ids: nil, wantSQL: "", wantN: 0},
}
for _, tc := range tests {
tc := tc
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
sql, args := vecDeletePlaceholders(tc.ids, tc.dollar)
if sql != tc.wantSQL {
t.Fatalf("占位串 = %q, want %q", sql, tc.wantSQL)
}
if len(args) != tc.wantN {
t.Fatalf("参数数 = %d, want %d", len(args), tc.wantN)
}
})
}
}
// TestSQLite_Delete_底层错误_异常:库已关闭时删除如实报错(而不是静默留下一堆
// 永远删不掉的向量)。
func TestSQLite_Delete_底层错误_异常(t *testing.T) {
t.Parallel()
ctx := context.Background()
db := openTestDB(t)
s, err := NewSQLite(ctx, db, 3, testLogger())
if err != nil {
t.Fatalf("NewSQLite: %v", err)
}
if err := db.Close(); err != nil {
t.Fatalf("close db: %v", err)
}
if err := s.Delete(ctx, "cat"); err == nil {
t.Fatal("库已关闭时 Delete 应当报错")
}
}
// TestPG_Delete_RequiresEnv 在设置了 DB_PG_URL 时跑一次真实的 pgvector 删除往返,
// 否则跳过(与 TestPG_RequiresEnv 同一口径:本环境默认没有 PG 服务)。
func TestPG_Delete_RequiresEnv(t *testing.T) {
url := os.Getenv("DB_PG_URL")
if url == "" {
t.Skip("DB_PG_URL not set; skipping real PostgreSQL+pgvector integration test")
}
ctx := context.Background()
idx, err := NewPG(ctx, url, 3, testLogger())
if err != nil {
t.Fatalf("NewPG: %v", err)
}
defer func() { _ = idx.Close() }()
if err := idx.Add(ctx, "delete-me", Vector{1, 0, 0}); err != nil {
t.Fatalf("Add: %v", err)
}
if err := idx.Delete(ctx, "delete-me"); err != nil {
t.Fatalf("Delete: %v", err)
}
hits, err := idx.Search(ctx, Vector{1, 0, 0}, 10)
if err != nil {
t.Fatalf("Search: %v", err)
}
for _, h := range hits {
if h.ID == "delete-me" {
t.Fatal("pgvector 上被删的向量还在")
}
}
// 幂等。
if err := idx.Delete(ctx, "delete-me"); err != nil {
t.Fatalf("重复 Delete 应当幂等: %v", err)
}
// 空 ids / 空白 id 空操作。
if err := idx.Delete(ctx); err != nil {
t.Fatalf("空 ids: %v", err)
}
if err := idx.Delete(ctx, ""); err != nil {
t.Fatalf("空白 id: %v", err)
}
// 反例对照:确认上面的 Search 不是因为「整张表空了」才没有命中。
if err := idx.Add(ctx, "keep-me", Vector{1, 0, 0}); err != nil {
t.Fatalf("Add keep-me: %v", err)
}
hits, err = idx.Search(ctx, Vector{1, 0, 0}, 10)
if err != nil {
t.Fatalf("Search keep-me: %v", err)
}
found := false
for _, h := range hits {
if h.ID == "keep-me" {
found = true
}
}
if !found {
t.Fatal("对照组失败:没删的向量也查不到,说明 Search 本身没工作")
}
}