From a017ab6cf8745c283ef989ac0e348cb2f4c730b7 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 13 Oct 2026 21:27:31 +0800 Subject: [PATCH] =?UTF-8?q?feat(memory):=20=E4=B8=B4=E6=97=B6=E8=AE=B0?= =?UTF-8?q?=E5=BF=86=E5=88=B0=E6=9C=9F=E7=89=A9=E7=90=86=E5=88=A0=E9=99=A4?= =?UTF-8?q?=EF=BC=88=E6=9C=89=E7=95=8C=E6=B8=85=E6=89=AB=20+=20=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E6=B8=85=E5=90=91=E9=87=8F=20+=20=E5=88=A0=E5=89=8D?= =?UTF-8?q?=E5=AE=A1=E8=AE=A1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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）。 --- cmd/cyrene/wire_memory_tidy.go | 24 +- internal/runtime/memory/expire.go | 319 ++++++++++++++++++ internal/runtime/memory/expire_loop.go | 124 +++++++ internal/runtime/memory/expire_loop_test.go | 216 ++++++++++++ internal/runtime/memory/expire_test.go | 311 +++++++++++++++++ internal/runtime/memory/vector/delete.go | 90 +++++ internal/runtime/memory/vector/delete_test.go | 184 ++++++++++ 7 files changed, 1262 insertions(+), 6 deletions(-) create mode 100644 internal/runtime/memory/expire.go create mode 100644 internal/runtime/memory/expire_loop.go create mode 100644 internal/runtime/memory/expire_loop_test.go create mode 100644 internal/runtime/memory/expire_test.go create mode 100644 internal/runtime/memory/vector/delete.go create mode 100644 internal/runtime/memory/vector/delete_test.go diff --git a/cmd/cyrene/wire_memory_tidy.go b/cmd/cyrene/wire_memory_tidy.go index 280f719..de02536 100644 --- a/cmd/cyrene/wire_memory_tidy.go +++ b/cmd/cyrene/wire_memory_tidy.go @@ -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)) diff --git a/internal/runtime/memory/expire.go b/internal/runtime/memory/expire.go new file mode 100644 index 0000000..b0e0468 --- /dev/null +++ b/internal/runtime/memory/expire.go @@ -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= 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]) + "…" +} diff --git a/internal/runtime/memory/expire_loop.go b/internal/runtime/memory/expire_loop.go new file mode 100644 index 0000000..77ca177 --- /dev/null +++ b/internal/runtime/memory/expire_loop.go @@ -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)) +} diff --git a/internal/runtime/memory/expire_loop_test.go b/internal/runtime/memory/expire_loop_test.go new file mode 100644 index 0000000..7b6b4a9 --- /dev/null +++ b/internal/runtime/memory/expire_loop_test.go @@ -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) +} diff --git a/internal/runtime/memory/expire_test.go b/internal/runtime/memory/expire_test.go new file mode 100644 index 0000000..6db9171 --- /dev/null +++ b/internal/runtime/memory/expire_test.go @@ -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) + } +} diff --git a/internal/runtime/memory/vector/delete.go b/internal/runtime/memory/vector/delete.go new file mode 100644 index 0000000..8528ed5 --- /dev/null +++ b/internal/runtime/memory/vector/delete.go @@ -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 +} diff --git a/internal/runtime/memory/vector/delete_test.go b/internal/runtime/memory/vector/delete_test.go new file mode 100644 index 0000000..0bcf7ff --- /dev/null +++ b/internal/runtime/memory/vector/delete_test.go @@ -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 本身没工作") + } +}