Merge branch 'feat/memexpire-bounded-expiry-sweep-with-audi' into dev
This commit is contained in:
AskaEth committed
commit
17c41265e8
9 files changed
+1329
-6
No files matched your search
@@ -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），起在 **TidyService.Start** 里、与心跳共用
|
||||
// 同一个 ctx 与 wg，因此它跟整理同生共死（`Stop` 一并停掉它），不新开关闭链、
|
||||
// 不新开端口、不加 env 键 —— 组合根这边**零接线**。
|
||||
//
|
||||
// 刻意**不做**的一件事（design §4.1 F）：
|
||||
// - 不把 sess-memorytidy 加进 flowPub 的图源白名单（默认关：整理是系统任务，
|
||||
@@ -49,6 +54,10 @@ func (a *app) wireMemoryTidy() error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("memory tidy: %w", err)
|
||||
}
|
||||
// 到期清扫（design/23 §19 的第 ② / ③ 步）**不在组合根挂**：它起在
|
||||
// TidyService.Start 里，与整理心跳共用同一个 ctx 与 wg（见 expire_loop.go）。
|
||||
// 因此这里零改动即完成接线 —— 不新起一条关闭链、不新开监听端口、不加任何
|
||||
// env 键，`wire_services.go` 的 a.memoryTidy.Stop 会连同它一起停掉。
|
||||
a.memoryTidy = svc
|
||||
return nil
|
||||
}
|
||||
@@ -58,11 +67,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))
|
||||
|
||||
@@ -69,6 +69,18 @@ func TestMemoryTidyIsWired(t *testing.T) {
|
||||
"main.go": {
|
||||
"a.memoryTidy.Start(ctx)",
|
||||
},
|
||||
// 到期清扫（切片 M3，design/23 §19 的第 ② / ③ 步）：临时记忆到期后要
|
||||
// **真删掉**，而它起在 TidyService.Start 里（与心跳同一个 cctx / wg）。
|
||||
// 少了这一句，过期行只会被读侧时效闸「挡在门外」而**永远不删** ——
|
||||
// 不报错、不红任何别的用例，只是库里的临时层无界留存。因此这里把它
|
||||
// 钉成 seam：`TestTidyServiceStart_清扫随服务起停` 是它的运行时对照。
|
||||
"../../internal/runtime/memory/tidy_service.go": {
|
||||
"go s.expirySweepLoop(cctx)",
|
||||
},
|
||||
"../../internal/runtime/memory/expire_loop.go": {
|
||||
"func (s *TidyService) expirySweepLoop(ctx context.Context)",
|
||||
"DefaultExpireSweepInterval",
|
||||
},
|
||||
"wire_services.go": {
|
||||
// 关库之前先停整理（它每轮要用连接池）。
|
||||
"a.memoryTidy.Stop(ctx)",
|
||||
|
||||
@@ -0,0 +1,320 @@
|
||||
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` 里
|
||||
// `ingestReviewSweepBatchSize` / `maxIngestReviewSweepBatches` 那一对常量的
|
||||
// 批删范式（`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]) + "…"
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 本文件是到期清扫的**后台件形态**：一个独立循环（与 `preinsert_gc.go` 的
|
||||
// `sweepLoop` 同形），启动期先扫一次、此后每 DefaultExpireSweepInterval 扫一次。
|
||||
//
|
||||
// 设计口径（`design/23 §19.3-2`）：删除只由内存模块的后台件串行触发，与整理 /
|
||||
// 审查同一单写者纪律，**不新开并发写口**、不新开监听端口、不加任何 `.env` 键
|
||||
// （`.env` 已冻结，`design/13 §2.1`），也不碰会话 / 历史读数。
|
||||
//
|
||||
// 🔴 **挂点**：清扫与整理**同生共死** —— 它的 ctx 是 TidyService.Start 里建的那个
|
||||
// cctx（就是 `go s.loop(cctx)` 用的同一个），它的 goroutine 登记进同一个 `s.wg`。
|
||||
// 因此 `cmd/cyrene/wire_services.go` 关闭链里的 `a.memoryTidy.Stop(ctx)` 会一并
|
||||
// 取消它并等它退出（§19.3-3 的「可停」），不必再为清扫加一条关闭路径。
|
||||
//
|
||||
// 🔴 挂点**必须**是 `Start` 里那处显式的 `go s.expirySweepLoop(cctx)`，不许改挂到
|
||||
// 某个注入缝上（例如包一层 `s.wait` 这个「心跳等待原语」）。理由是**可证性**：
|
||||
// 包注入缝意味着清扫的存活依赖于「`loop` 每一轮都调 `s.wait`」这个**没有任何契约
|
||||
// 固定**的假设 —— 将来谁换了等待原语、或把循环改成 `select` + 自己的 timer，清扫
|
||||
// 会**静默消失**（不报错、不红测试，只是临时期限再也删不掉）。起在 `Start` 里，
|
||||
// 读者一眼就能看到两个后台件同生共死。
|
||||
|
||||
// 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))
|
||||
}
|
||||
@@ -0,0 +1,277 @@
|
||||
package memory
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory/vector"
|
||||
)
|
||||
|
||||
// 本文件锁定期清扫的**接线与后台形态**（design/23 §19）：挂点（TidyService.Start
|
||||
// 里的 `go s.expirySweepLoop(cctx)`）、随 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())
|
||||
}
|
||||
}
|
||||
|
||||
// TestTidyServiceStart_清扫随服务起停 钉住**接线**（design/23 §19.3-3 的「可停」）：
|
||||
// 清扫起在 TidyService.Start 里，与心跳共用同一个 cctx 与 wg，因此
|
||||
//
|
||||
// ① Start 一跑它就立刻扫一轮（启动补扫，清掉上次退出后到期的行）；
|
||||
// ② Stop 会取消它并等它退出 —— Stop 没超时即证明它真的随 ctx 走了。
|
||||
//
|
||||
// 它同时是 `cmd/cyrene/wire_memory_tidy_test.go` 那条 seam 断言的运行时对照：把
|
||||
// Start 里那行 `go s.expirySweepLoop(cctx)` 摘掉，这条用例会红（行永远不会被删，
|
||||
// 且 Stop 会因为 wg 计数对不上而超时）。
|
||||
func TestTidyServiceStart_清扫随服务起停(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
store := newTestStore(t)
|
||||
// 10:00Z：不在安静窗（23–7）内 ⇒ Start 里的补跑判定为「不该跑」，整理不搅局。
|
||||
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(_, what, result string) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
audits = append(audits, what+" "+result)
|
||||
})
|
||||
// 假等待：把整理心跳停在 ctx 上（真定时器要等 24h 档）。它只影响心跳 ——
|
||||
// 清扫是**另一个** goroutine + 自己的小时级 ticker。
|
||||
svc.wait = func(ctx context.Context, _ time.Duration) bool {
|
||||
<-ctx.Done()
|
||||
return false
|
||||
}
|
||||
|
||||
svc.Start(ctx)
|
||||
|
||||
waitUntilRowGone(t, store, row.ID)
|
||||
|
||||
stopCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
if err := svc.Stop(stopCtx); err != nil {
|
||||
t.Fatalf("Stop 未能在预算内合上 —— 清扫没有随 ctx 退出 / 没登记进同一个 wg: %v", err)
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(audits) != 1 || !strings.HasPrefix(audits[0], expireAuditVerb+" ") {
|
||||
t.Fatalf("审计 = %v, want 一条 what=%s", audits, expireAuditVerb)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTidyServiceStart_整理关掉时清扫不跑 钉住那条**行为契约**（不是断言字段那种
|
||||
// 假牙）：`MEMORY_TIDY_ENABLED=false` ⇒ 到期清扫**不跑**，过期行**一条都不删**。
|
||||
//
|
||||
// 为什么值得一条专门的用例：清扫那行 `go s.expirySweepLoop(cctx)` 就排在
|
||||
// `if !s.cfg.Enabled { … return }` **之后**。将来谁把 Enabled 检查挪到后面、
|
||||
// 或让 Start 在关掉时也走完整条路，过期行就会在「整理被关掉」时被物理删除 ——
|
||||
// 那是与「后台件随宿主启停」（`preinsert_gc` 随审查服务启停，design/23 §19.3-2）
|
||||
// 相反的行为，而且**不会红任何别的用例**（读侧闸照样挡着，表面上一切正常）。
|
||||
//
|
||||
// 断言是「没有删除发生 + 没有审计 + running 保持 false」三件一起 —— 只看
|
||||
// `cfg.Enabled` 字段是假牙：那证明不了 Start 真的按关掉处理。
|
||||
func TestTidyServiceStart_整理关掉时清扫不跑(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
|
||||
)
|
||||
cfg := tidyTestCfg(0.80)
|
||||
cfg.Enabled = false
|
||||
svc := newTidyServiceForTest(t, store, cfg, nil, fixedTidyClock(now),
|
||||
func(_, what, result string) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
audits = append(audits, what+" "+result)
|
||||
})
|
||||
// 假等待：万一清扫/心跳被起起来了，它会停在这里而不是空转（不影响断言）。
|
||||
svc.wait = func(ctx context.Context, _ time.Duration) bool {
|
||||
<-ctx.Done()
|
||||
return false
|
||||
}
|
||||
|
||||
svc.Start(ctx)
|
||||
|
||||
// ① 确定性断言：Start 走的是「关掉 ⇒ 直接返回」那条路。
|
||||
if svc.running {
|
||||
t.Fatal("Enabled=false 时 Start 应当直接返回，running 必须保持 false")
|
||||
}
|
||||
// ② 让「清扫其实被起起来了」有机会显形：清扫是启动即扫一轮，中间只有
|
||||
// 「取行 + 删行」一次 SQLite 往返（同一台机上实测在微秒级，远比这个窗口小）；
|
||||
// 对照组 TestTidyServiceStart_清扫随服务起停 用的就是同一个语料与同一个时钟，
|
||||
// 它在毫秒级内把行删掉。给同一个数量级的窗口再断言「什么也没发生」。
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
if !memoryRowExists(t, store, row.ID) {
|
||||
t.Fatal("整理被关掉了，到期行却被删了 —— 清扫跑在了 Enabled 闸之外")
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if len(audits) != 0 {
|
||||
t.Fatalf("整理被关掉了，却写出了到期审计：%v", audits)
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -111,6 +111,11 @@ func (s *TidyService) Enabled() bool { return s.cfg.Enabled }
|
||||
// Start 启动心跳。它是幂等的：第二次调用是空操作。配置关闭时只记一条日志
|
||||
// 并立即返回（关闭是显式选择，绝不静默）。ctx（可为 nil）是所有内部工作的
|
||||
// 父级：取消它会像 Stop 一样停止服务。
|
||||
//
|
||||
// 它**同时**起临时期限的到期清扫（design/23 §19 的第 ② / ③ 步，见
|
||||
// expire_loop.go）：清扫与心跳共用同一个 cctx 与 wg，因此它跟整理同生共死 ——
|
||||
// 整理被配置关掉时清扫也不跑（与 `preinsert_gc.go` 的清扫随审查服务启停同口径），
|
||||
// `Stop` 会一并取消并 join 它。
|
||||
func (s *TidyService) Start(ctx context.Context) {
|
||||
if !s.cfg.Enabled {
|
||||
s.log.Info("memory tidy disabled by configuration", "env", "MEMORY_TIDY_ENABLED")
|
||||
@@ -128,12 +133,36 @@ func (s *TidyService) Start(ctx context.Context) {
|
||||
cctx, cancel := context.WithCancel(ctx)
|
||||
s.cancel = cancel
|
||||
s.wg.Add(1)
|
||||
// 到期清扫（design/23 §19 的第 ② / ③ 步）与心跳**同生共死**：同一个 cctx、
|
||||
// 同一个 wg，因此 Stop 会一并取消并 join 它（§19.3-3 的「可停」）。
|
||||
//
|
||||
// 🔴 为什么这笔 Add 也在锁内（与上面那一笔同处）：让 `Stop` 的 `running` 判定
|
||||
// 与这两笔 Add **原子化** —— 于是不可能出现「Stop 已看到 `running=false` 早退」
|
||||
// 或「Wait 漏掉一笔尚未 Add 的 goroutine（那会当场泄漏一个清扫 goroutine）」
|
||||
// 这两种窗口。
|
||||
//
|
||||
// ⚠️ 这把锁**不**覆盖 `Stop` 的 `wg.Wait`：在同一个 `func (s *TidyService) Stop`
|
||||
// 里，`s.mu.Unlock()` 在前、`s.wg.Wait()`（另一个 goroutine 里）在后，两者之间
|
||||
// 没有锁。因此 WaitGroup 的复用纪律（「上一轮 Wait 返回前不得再 Add」）**不是**
|
||||
// 由这把锁给的，而是靠「`Start` / `Stop` 在整个进程生命周期各只调用一次」维持：
|
||||
// `Start` 只在组合根启动期调一次（`cmd/cyrene/main.go` 的
|
||||
// `a.memoryTidy.Start(ctx)`）、`Stop` 只在关停链里调一次（`cmd/cyrene/wire_services.go`
|
||||
// 的 `_ = a.memoryTidy.Stop(ctx)`）。
|
||||
// 交叉调用（`Start` 与在飞的 `Stop` 并发）在 `-race` 下会 `panic: sync:
|
||||
// WaitGroup is reused before previous Wait has returned` —— 实测该 panic 在
|
||||
// M3 之前的基线上同样可复现（既有性质，非本切片引入），只是当前调用点**不可达**。
|
||||
s.wg.Add(1)
|
||||
s.mu.Unlock()
|
||||
|
||||
// 持久事实先读进来：判定「要不要补跑」只看它，不看进程内状态。
|
||||
s.loadPersistedLastRun(ctx)
|
||||
|
||||
// 两个后台件在这里一起起：整理心跳，以及临时期限的到期清扫
|
||||
// （expire_loop.go）。清扫跑在**自己的** goroutine 与**自己的**小时级
|
||||
// ticker 上 —— 整理的周期是 24h 档（intervalSec），不能拿来当清扫周期
|
||||
// （§19.1「到期后 ≤ 一个清扫周期，默认每小时」）。
|
||||
go s.loop(cctx)
|
||||
go s.expirySweepLoop(cctx)
|
||||
|
||||
s.log.Info("memory tidy started",
|
||||
"interval", s.cfg.Interval.String(),
|
||||
|
||||
@@ -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 本身没工作")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user