From ffe2518bf42ca777e1b7e4cbe80bbb0ca37b246b Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 6 Oct 2026 15:59:16 +0800 Subject: [PATCH] =?UTF-8?q?feat(memory):=20=E7=95=99=E7=97=95=E8=A1=A8?= =?UTF-8?q?=E6=AD=A3=E5=BC=8F=E8=BF=81=E7=A7=BB=200025=20+=20=E4=BF=9D?= =?UTF-8?q?=E7=95=99=E6=9C=9F=E6=B8=85=E6=89=AB=EF=BC=88W5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 迁移 0025_memory_ingest_reviews（幂等，既有部署重跑不炸/不丢行） - GC：保留期 720h + 条数上限 10 万行，挂进审查服务生命周期（Start 起 / Stop 停 / 启动先扫） - 过期即硬删（RevertMerged 只读 memory_items，删留痕不损失回滚能力） - 偏离一处并给理由：表只留 created_at（先例 0017；append-only；加 updated_at 需三份 DDL 真源） - 3 个变异自检（边界 <=/元组比较方向/Retention=0） --- .../0025_memory_ingest_reviews.down.sql | 14 + .../0025_memory_ingest_reviews.up.sql | 59 ++++ internal/runtime/memory/preinsert.go | 18 +- internal/runtime/memory/preinsert_gc.go | 259 ++++++++++++++++++ internal/runtime/memory/preinsert_gc_test.go | 251 +++++++++++++++++ .../memory/preinsert_migration_test.go | 257 +++++++++++++++++ internal/runtime/memory/preinsert_schema.go | 12 +- 7 files changed, 861 insertions(+), 9 deletions(-) create mode 100644 internal/db/migrations/0025_memory_ingest_reviews.down.sql create mode 100644 internal/db/migrations/0025_memory_ingest_reviews.up.sql create mode 100644 internal/runtime/memory/preinsert_gc.go create mode 100644 internal/runtime/memory/preinsert_gc_test.go create mode 100644 internal/runtime/memory/preinsert_migration_test.go diff --git a/internal/db/migrations/0025_memory_ingest_reviews.down.sql b/internal/db/migrations/0025_memory_ingest_reviews.down.sql new file mode 100644 index 0000000..301ecca --- /dev/null +++ b/internal/db/migrations/0025_memory_ingest_reviews.down.sql @@ -0,0 +1,14 @@ +-- 0025 的回滚：先删索引、再删表（与 0019 / 0021 的回滚同一顺序口径）。 +-- +-- 删留痕**不影响任何一条记忆的可逆性**：留痕描述的那次软合并早已落在 +-- `memory_items` 上（`status` / `merged_into` / `merged_from` / `merge_count`）， +-- 回滚一次合并（`Store.RevertMerged`）只读 `memory_items`（见 +-- internal/runtime/memory/preinsert_store.go 的 selectMergedIntoSQL）， +-- 从不读本表。代价只是「这次决定当初为什么这么判」的依据丢失，属降级可接受。 +-- +-- 且本表由记忆存储自举（`NewStore` → `ensureIngestReviewSchema`）：回滚删表之后， +-- 下次启动会按同一 DDL 重建一张空表，不需要人工干预（这也是本表与 +-- memory_items 这类「用户数据表」的关键区别 —— 它是派生的观测留痕，不是正文）。 +DROP INDEX IF EXISTS idx_memory_ingest_reviews_created_at; +DROP INDEX IF EXISTS idx_memory_ingest_reviews_item_id; +DROP TABLE IF EXISTS memory_ingest_reviews; diff --git a/internal/db/migrations/0025_memory_ingest_reviews.up.sql b/internal/db/migrations/0025_memory_ingest_reviews.up.sql new file mode 100644 index 0000000..cab0c33 --- /dev/null +++ b/internal/db/migrations/0025_memory_ingest_reviews.up.sql @@ -0,0 +1,59 @@ +-- 0025_memory_ingest_reviews: 入库前审查（design/23 §15.1 D3）留痕表的**正式迁移号** +-- （切片 W5；D3 本体已落地于 6e28b38，当时**刻意没占迁移号**，以免与并行切片抢号）。 +-- +-- 为什么现在补：本表此前**只**由 `memory.NewStore` 的幂等前滚创建 +-- （internal/runtime/memory/preinsert_schema.go）。后果是「库里有这张表」这件事 +-- 分裂成两份真源：新库靠自举、迁移账本却不知道它存在；而清理（保留期清扫）要按 +-- `created_at` 扫表 —— 一个不在迁移账本里的表，没人能回答「它的形状是谁负责的」。 +-- 本迁移把它登记进 schema_migrations（手法照 0019_memory_tidy 对 memory_items +-- 整理列 / 候选表 / 运行表的处理）。 +-- +-- 🔴 **幂等是硬要求**（本迁移的部署面比一般迁移宽）：**所有既有部署的 +-- `memory_ingest_reviews` 早已存在**（NewStore 自举），只是没有 0025 这一行。 +-- 因此 DDL 一律 `IF NOT EXISTS`：重复执行 = 空操作，绝不因为「表已存在」而炸， +-- 也不会因为「表已存在」而漏建索引。形状必须与自举 DDL **逐字同形**（少一列就是 +-- 「新库有、老库无」的漂移），由 `internal/runtime/memory/preinsert_migration_test.go` +-- 的对照测试钉住（迁移建的表 vs 自举建的表，列与索引逐项比对）。 +-- +-- 命名口径：表名复数、主键 `id`、索引 `idx_{table}_{col}`（援引 AGENTS §10 已登记例外， +-- 与既有迁移一致）。 +-- +-- 时间戳：**只有 created_at**（定宽 UTC 纳秒串，由 internal/runtime/memory.formatTime +-- 写入；见该包 timefmt.go 的口径）。本表是 **append-only**：留痕落一行即不再改写， +-- 因此按同类逐条样本表 `0017_llm_usage` 的先例**不引入 updated_at** —— 一个永远等于 +-- created_at 的列只会撒谎，而且要补它就必须为既有部署再加一条幂等 ALTER， +-- 把 DDL 的两份真源变成三份。若将来真出现「留痕被后续动作改写」的需求， +-- 那是 shape 变更，按项目惯例开新迁移号。 +-- +-- 脱敏（design/23 §15 D3-8「记忆正文不进日志」延伸到表结构）：本表**没有**承载 +-- 记忆正文的列 —— 命中的既有 id（target_id / merged_ids）足以让 owner-only 的 +-- 读侧回表取原文，而正文一个字都不进这张表。 + +CREATE TABLE IF NOT EXISTS memory_ingest_reviews ( + id TEXT PRIMARY KEY, + item_id TEXT NOT NULL, + persona_id TEXT NOT NULL, + scope TEXT NOT NULL, + project_id TEXT NOT NULL DEFAULT '', + decision TEXT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + veto TEXT NOT NULL DEFAULT '', + target_id TEXT NOT NULL DEFAULT '', + merged_ids TEXT NOT NULL DEFAULT '[]', + score REAL NOT NULL DEFAULT 0, + recall_mode TEXT NOT NULL DEFAULT '', + degraded INTEGER NOT NULL DEFAULT 0, + rule_version TEXT NOT NULL DEFAULT '', + session_id TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL +); + +-- 按被审条目回溯（「这一条记忆当初被谁合了」）。 +CREATE INDEX IF NOT EXISTS idx_memory_ingest_reviews_item_id + ON memory_ingest_reviews (item_id); + +-- 读侧倒序（ListIngestReviews）与**保留期清扫的唯一依赖**：批删语句带 +-- `WHERE created_at < ? ORDER BY created_at LIMIT ?`，吃得上这条索引， +-- 且每批的扫描面有界（internal/runtime/memory/preinsert_gc.go）。 +CREATE INDEX IF NOT EXISTS idx_memory_ingest_reviews_created_at + ON memory_ingest_reviews (created_at); diff --git a/internal/runtime/memory/preinsert.go b/internal/runtime/memory/preinsert.go index bceb9a2..8cff25d 100644 --- a/internal/runtime/memory/preinsert.go +++ b/internal/runtime/memory/preinsert.go @@ -216,8 +216,8 @@ func (r *PreInsertReviewer) SessionID() string { return TidySessionID } // Enabled 报告审查是否被配置开启。 func (r *PreInsertReviewer) Enabled() bool { return r.cfg.Enabled } -// Start 拉起工作协程。幂等；配置关闭时只记一条日志并立即返回（关闭是显式 -// 选择，绝不静默）。 +// Start 拉起工作协程（审查队列 + 留痕保留期清扫）。幂等；配置关闭时只记一条日志 +// 并立即返回（关闭是显式选择，绝不静默）。 func (r *PreInsertReviewer) Start(ctx context.Context) { if !r.cfg.Enabled { r.log.Info("memory ingest review disabled by configuration", "env", "MEMORY_REVIEW_ENABLED") @@ -234,10 +234,17 @@ func (r *PreInsertReviewer) Start(ctx context.Context) { r.running = true cctx, cancel := context.WithCancel(ctx) r.cancel = cancel - r.wg.Add(1) + // 两个协程（审查工作协程 + 留痕清扫循环）**在释放锁之前**一次性登记， + // 避免 Stop 的 wg.Wait 与随后的 Add 并发（WaitGroup 的「计数归零后不许再 Add」 + // 是硬约束，不只是风格）。 + r.wg.Add(2) r.mu.Unlock() go r.worker(cctx) + // 留痕清扫（design/23 §15 D3-8 的 memory_ingest_reviews）与审查工作协程同生共死： + // 启动期扫一次、此后按间隔定时扫（口径见 preinsert_gc.go）。它删的是**已经过期 + // 的观测留痕**，与审查本身的判定/记忆行无关，因此失败只记 WARN、绝不退出。 + go r.sweepLoop(cctx) r.log.Info("memory ingest review started", "session_id", TidySessionID, @@ -245,7 +252,10 @@ func (r *PreInsertReviewer) Start(ctx context.Context) { "scan_limit", r.cfg.ScanLimit, "timeout", r.cfg.Timeout.String(), "recall_threshold", r.cfg.Threshold, - "rule_version", IngestRuleVersion) + "rule_version", IngestRuleVersion, + "sweep_interval", DefaultIngestReviewGCInterval.String(), + "retention", DefaultIngestReviewRetention.String(), + "max_reviews", DefaultIngestReviewMaxRows) } // Stop 停止工作协程并等待正在审的那一条跑完。幂等；ctx 在排空前被取消时返回 diff --git a/internal/runtime/memory/preinsert_gc.go b/internal/runtime/memory/preinsert_gc.go new file mode 100644 index 0000000..dabffe4 --- /dev/null +++ b/internal/runtime/memory/preinsert_gc.go @@ -0,0 +1,259 @@ +package memory + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" +) + +// 本文件是入库前审查留痕（design/23 §15 D3-8 的 `memory_ingest_reviews`）的 +// **保留期清扫（Sweep）**（切片 W5）。动因是一条实测口径：审查**每写一条新记忆 +// 就落一行留痕**（现网节奏 ≈ 700–1000 行/天），而 MTB 落地时这张表既没有迁移号、 +// 也没有 TTL ⇒ 无界增长（DECISIONS-2026-10-06 §D 已裁「补」）。 +// +// 范式逐条照 `internal/runtime/events/gc.go` 与 `internal/runtime/flowgraph/gc.go` +// （design/26 §4.4 明写「不另造」），三条取舍完全一致： +// +// 1. **启动期扫一次、此后按间隔定时扫**（间隔默认 DefaultIngestReviewGCInterval）。 +// 循环住在**入库前审查服务**的生命周期里（Start 起来、Stop 停掉）：它是本包内 +// 唯一既持有 context、又有确定启停点的记忆后台件。不新开监听端口、不新起 +// 独立的组合根接线（AGENTS §9 单二进制、单端口）。 +// 2. **失败只记 WARN、绝不让进程退出**：Sweep 把查询/删除失败**上抛**给调用方 +// （调用方记 WARN 后继续等下一轮）；已删的行不会因为后续批次失败而「回滚」， +// 未删的行也不会被清空 —— 一轮失败就只是「这一轮没扫干净」。 +// 3. **批数有界**：一轮清扫最多 maxIngestReviewSweepBatches × +// ingestReviewSweepBatchSize 行（积压时不一次 DELETE 全表，剩下的留给下一轮）， +// 与 AGENTS §10 的压测纪律同口径：有界，不靠堆资源。 +// +// 两条清理判据（**过期即删**，理由见常量注释）： +// +// - **保留期**：`created_at < now − 保留期` 的行逐批删除（默认 720h）； +// - **条数上限**：超过 DefaultIngestReviewMaxRows 行时，按 `(created_at, rowid)` +// 元组裁掉最旧的尾巴 —— 保留期管「日常节奏」，上限管「异常爆发」 +// （一次回填/一个 bug 可以在保留期内灌进百万行，只有条数能把它挡住）。 +// +// **为什么是硬删而不是软删**：留痕描述的那次软合并**早已落在 `memory_items` 上** +// （`status`/`merged_into`/`merged_from`/`merge_count`），`RevertMerged` 只读 +// `memory_items`（preinsert_store.go 的 selectMergedIntoSQL），从不读本表 +// ⇒ 删留痕不损失任何回滚能力。反过来，软删既要加列（shape 变更）又保留了这份 +// 无界增长本身 —— 等于把问题换个名字留着。故过期即**物理删除**（design/23 §8 的 +// 「软删除」是对 `memory_items` 正文的口径，不是对观测留痕的口径）。 + +const ( + // DefaultIngestReviewRetention 是留痕的保留期默认值：30 天（720h）。 + // 依据（同为 720h 的两张表）：① `llm_usage`（0017）是**同类逐条样本表**， + // 同为 append-only + 按 created_at 清扫，保留期 720h；② `channel_events` + // （0021）的口径是「事后定案证据」，也取 720h。反例是 `flow_graphs`（168h）—— + // 那是「回看最近若干轮」的可视化窗口；而留痕要答的问题是 + // 「上周那条记忆**当初为什么**被合掉/放行」，一周往往不够（事故复盘常在事后 + // 一周以上）。体量上 30 天 × 1000 行/天 ≈ 3 万行（每行几十字节，无正文），可忽略。 + DefaultIngestReviewRetention = 720 * time.Hour + + // DefaultIngestReviewGCInterval 是定时清扫间隔默认值：每小时（与 events / + // flowgraph 的 DefaultGCInterval 同值）。清扫只是删几行，频率再高也无收益。 + DefaultIngestReviewGCInterval = time.Hour + + // DefaultIngestReviewMaxRows 是留痕的条数上限默认值：10 万行。 + // 按现网节奏（700–1000 行/天）约合 100–140 天，是保留期（30 天）的 3 倍以上 + // ⇒ **正常节奏下永不触发**，只在异常爆发（一次性回填、写路径 bug 循环落痕） + // 时兜底，保证「无论如何库里的留痕行数都有上界」。数量级参考：10 万行 + // × 约 150 字节 ≈ 15 MB。 + DefaultIngestReviewMaxRows = 100_000 + + // ingestReviewSweepBatchSize 是单批删除的行数上限；maxIngestReviewSweepBatches + // 是一轮清扫的批数上限。两者一起把「一轮清扫」的时长、单次内存占用与 + // **连接池占用**（本项目的 SQLite 池是单连接）都变得有界。 + ingestReviewSweepBatchSize = 500 + maxIngestReviewSweepBatches = 100 +) + +// IngestReviewGCPolicy 是一轮清扫的口径。两个字段各自独立生效： +// `Retention <= 0` 只关掉按时间的一档、`MaxRows <= 0` 只关掉条数上限那一档 +// （不是总体关闭 —— 本表没有「关掉清扫」的开关是**刻意**的：它的无界增长是 +// 结构性的，留一个能关掉 GC 的开关就等于留一个能让库无限长大的开关。 +// 真要不留痕，该关掉的是审查本身 `MEMORY_REVIEW_ENABLED=false` —— 那样也就 +// 不再产生新行，且组合根根本不会构造审查服务）。 +// +// 保留期与上限的默认值见本文件顶部的常量：组合根不传任何配置键，`.env` 已冻结 +// （design/13 §2.1），将来若要面板可调，落点是 `$DATA_DIR/config/**`， +// 由 internal/config 侧读出后传进本结构 —— 本包不读任何配置源。 +type IngestReviewGCPolicy struct { + // Retention 是留痕保留期；<=0 表示不做按时间的清扫。 + Retention time.Duration + // MaxRows 是留痕条数上限；<=0 表示不做条数裁剪。 + MaxRows int +} + +// IngestReviewSweepStats 是一轮清扫的汇总（成功件只记总数，绝不记任何行内容）。 +type IngestReviewSweepStats struct { + // Batches 是实际执行的删除批数（过期档 + 上限档之和）。 + Batches int + // Deleted 是按保留期删掉的过期行数。 + Deleted int64 + // Truncated 是按条数上限裁掉的（**尚未过期**的）行数。它非零即说明发生了 + // 异常爆发 —— 是运维信号，不该被当成日常数字。 + Truncated int64 +} + +// sweepIngestReviewsExpiredSQL 按时间戳删一批过期行。子查询带 +// `ORDER BY created_at LIMIT ?`，因此它吃得上 idx_memory_ingest_reviews_created_at， +// 且每批的扫描面有界。时间戳用定宽 UTC 串比较：字典序等价于时间序 +// （见 timefmt.go 的口径）。边界取**严格小于**：正好落在 cutoff 上的行保留， +// 于是「保留期 = 恰好 30 天」是可断言的（不随执行时刻抖动）。 +const sweepIngestReviewsExpiredSQL = `DELETE FROM memory_ingest_reviews WHERE id IN ( + SELECT id FROM memory_ingest_reviews WHERE created_at < ? ORDER BY created_at LIMIT ?)` + +// selectIngestReviewCapKeySQL 取「要保留的最后一行」的键：按 `(created_at, rowid)` +// 倒序数的第 MaxRows 行。`rowid` 是本表的隐式主键行号（id 是 TEXT 主键，因此 +// 本表是普通 rowid 表；读侧 ListIngestReviews 的 `ORDER BY created_at DESC, rowid DESC` +// 已经在用同一口径）。 +const selectIngestReviewCapKeySQL = `SELECT created_at, rowid FROM memory_ingest_reviews + ORDER BY created_at DESC, rowid DESC LIMIT 1 OFFSET ?` + +// sweepIngestReviewsCapSQL 按上一步取到的键裁一批：`(created_at, rowid)` 严格小于 +// 该键的行即「最旧的尾巴」。用**元组比较**而不是只比 created_at，是为了在 +// 同一时间戳的批量写入（注入固定时钟的回填）下也能精确裁到 MaxRows —— 只比 +// 时间戳时那批同值行谁也删不掉，上限恰好会在最需要它的时候失效。 +const sweepIngestReviewsCapSQL = `DELETE FROM memory_ingest_reviews WHERE rowid IN ( + SELECT rowid FROM memory_ingest_reviews + WHERE created_at < ? OR (created_at = ? AND rowid < ?) + ORDER BY created_at, rowid LIMIT ?)` + +// SweepIngestReviews 按 policy 清扫留痕表，返回一轮汇总。now 由调用方给出 +// （便于测试注入固定时钟；本包不在 Store 里藏墙钟）。 +// +// 失败语义：查询/删除失败返回错误并**带上已完成的统计**（调用方据此知道这一轮 +// 走到了哪一步，记 WARN 后等下一轮）。ctx 取消时立刻停下：**已删的保持删除、 +// 未删的绝不因此被清空**。 +func (s *Store) SweepIngestReviews(ctx context.Context, now time.Time, policy IngestReviewGCPolicy) (IngestReviewSweepStats, error) { + var stats IngestReviewSweepStats + if policy.Retention > 0 { + cutoff := formatTime(now.UTC().Add(-policy.Retention)) + deleted, batches, err := s.deleteIngestReviews(ctx, func(ctx context.Context) (int64, error) { + return s.execIngestReviewSweep(ctx, sweepIngestReviewsExpiredSQL, cutoff, ingestReviewSweepBatchSize) + }) + stats.Deleted += deleted + stats.Batches += batches + if err != nil { + return stats, err + } + } + if policy.MaxRows > 0 { + key, ok, err := s.readIngestReviewCapKey(ctx, policy.MaxRows) + if err != nil { + return stats, err + } + if ok { + deleted, batches, err := s.deleteIngestReviews(ctx, func(ctx context.Context) (int64, error) { + return s.execIngestReviewSweep(ctx, sweepIngestReviewsCapSQL, + key.createdAt, key.createdAt, key.rowid, ingestReviewSweepBatchSize) + }) + stats.Truncated += deleted + stats.Batches += batches + if err != nil { + return stats, err + } + } + } + if stats.Deleted > 0 || stats.Truncated > 0 { + s.log.Info("memory ingest review sweep", + "deleted", stats.Deleted, "truncated", stats.Truncated, "batches", stats.Batches, + "retention", policy.Retention.String(), "max_rows", policy.MaxRows) + } + return stats, nil +} + +// ingestReviewCapKey 是「要保留的最后一行」的键。 +type ingestReviewCapKey struct { + createdAt string + rowid int64 +} + +// readIngestReviewCapKey 取第 maxRows 新的一行的键（maxRows <= 1 时即最新那一行）。 +// 表里行数不足 maxRows 时返回 ok=false（`sql.ErrNoRows`）—— 那是「无需裁剪」， +// 不是错误。 +func (s *Store) readIngestReviewCapKey(ctx context.Context, maxRows int) (ingestReviewCapKey, bool, error) { + var key ingestReviewCapKey + err := s.db.QueryRowContext(ctx, selectIngestReviewCapKeySQL, maxRows-1). + Scan(&key.createdAt, &key.rowid) + if errors.Is(err, sql.ErrNoRows) { + return ingestReviewCapKey{}, false, nil + } + if err != nil { + return ingestReviewCapKey{}, false, fmt.Errorf("memory: read ingest review cap key: %w", err) + } + return key, true, nil +} + +// deleteIngestReviews 跑有界的批删循环：sweep 每批返回删掉的行数，某批不满一批 +// 即收手（说明已经删干净）。批数上限 maxIngestReviewSweepBatches 保证一轮清扫的 +// 时长与连接占用有界；ctx 取消时立刻返回（已完成的部分保持删除）。 +func (s *Store) deleteIngestReviews(ctx context.Context, sweep func(context.Context) (int64, error)) (int64, int, error) { + var ( + total int64 + batches int + ) + for i := 0; i < maxIngestReviewSweepBatches; i++ { + if err := ctx.Err(); err != nil { + return total, batches, fmt.Errorf("memory: sweep ingest reviews: %w", err) + } + deleted, err := sweep(ctx) + if err != nil { + return total, batches, err + } + batches++ + total += deleted + if deleted < int64(ingestReviewSweepBatchSize) { + break + } + } + return total, batches, nil +} + +// execIngestReviewSweep 执行一条批删语句并返回本批删掉的行数。 +func (s *Store) execIngestReviewSweep(ctx context.Context, query string, args ...any) (int64, error) { + res, err := s.db.ExecContext(ctx, query, args...) + if err != nil { + return 0, fmt.Errorf("memory: sweep ingest reviews 删留痕行: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return 0, fmt.Errorf("memory: sweep ingest reviews 读取删除行数: %w", err) + } + return n, nil +} + +// sweepLoop 是留痕清扫的常驻循环：启动期先扫一次（清掉上次退出后累积的过期行）， +// 此后每 DefaultIngestReviewGCInterval 扫一次。它由 PreInsertReviewer.Start 拉起、 +// 随 ctx 取消退出（Stop 会 join 它）。 +// +// 它**不**碰审查队列，也不读记忆正文：清扫只按 created_at 与 rowid 删行。 +func (r *PreInsertReviewer) sweepLoop(ctx context.Context) { + defer r.wg.Done() + r.sweepOnce(ctx) + ticker := time.NewTicker(DefaultIngestReviewGCInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.sweepOnce(ctx) + } + } +} + +// sweepOnce 跑一轮留痕清扫。失败只记 WARN：清扫是旁路维护，绝不让审查服务退出、 +// 也绝不影响任何一条记忆（与 design/23 §15 D3-5 的 fail-safe 同口径）。 +func (r *PreInsertReviewer) sweepOnce(ctx context.Context) { + stats, err := r.deps.Store.SweepIngestReviews(ctx, r.now(), IngestReviewGCPolicy{ + Retention: DefaultIngestReviewRetention, + MaxRows: DefaultIngestReviewMaxRows, + }) + if err != nil { + r.log.Warn("memory ingest review sweep failed, expired rows left for the next round", + "error", err, "deleted", stats.Deleted, "truncated", stats.Truncated) + } +} diff --git a/internal/runtime/memory/preinsert_gc_test.go b/internal/runtime/memory/preinsert_gc_test.go new file mode 100644 index 0000000..b0168ca --- /dev/null +++ b/internal/runtime/memory/preinsert_gc_test.go @@ -0,0 +1,251 @@ +package memory + +import ( + "context" + "fmt" + "slices" + "testing" + "time" +) + +// 本文件是入库前审查留痕（design/23 §15 D3-8）的**保留期清扫**测试（切片 W5）： +// 只删过期的 / 边界严格小于 / 条数上限精确且对同时间戳爆发也生效 / 批数有界且 +// 可重复 / 启动期真的扫过一次。口径与 internal/runtime/events 的 gc_test.go 同构。 + +// countReviewRows 数留痕表里还剩几行。 +func countReviewRows(t *testing.T, store *Store) int { + t.Helper() + var n int + if err := store.db.QueryRowContext(context.Background(), + `SELECT COUNT(*) FROM memory_ingest_reviews`).Scan(&n); err != nil { + t.Fatalf("count ingest reviews: %v", err) + } + return n +} + +// remainingReviewIDs 按 `(created_at, rowid)` 升序返回留痕 id（最旧在前）。 +func remainingReviewIDs(t *testing.T, store *Store) []string { + t.Helper() + rows, err := store.db.QueryContext(context.Background(), + `SELECT id FROM memory_ingest_reviews ORDER BY created_at, rowid`) + if err != nil { + t.Fatalf("list ingest review ids: %v", err) + } + defer func() { _ = rows.Close() }() + out := make([]string, 0) + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + t.Fatalf("scan ingest review id: %v", err) + } + out = append(out, id) + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate ingest review ids: %v", err) + } + return out +} + +// seedReviewRow 落一条留痕：走生产插入路径（insertIngestReview）， +// 因此 created_at 的格式与真实写入**逐字一致**（定宽 UTC 纳秒串）。 +func seedReviewRow(t *testing.T, store *Store, id string, at time.Time) { + t.Helper() + if err := store.insertIngestReview(context.Background(), IngestReview{ + ID: id, + ItemID: "item-" + id, + PersonaID: "persona.a", + Scope: ScopePersona, + Decision: IngestCreate, + CreatedAt: at, + }); err != nil { + t.Fatalf("insert ingest review %q: %v", id, err) + } +} + +// TestSweepIngestReviewsDeletesOnlyExpired 是保留期判据的钉死：只删**严格早于** +// `now − 保留期` 的行；正好落在 cutoff 上的行保留（否则「保留期恰好 30 天」会随 +// 执行时刻抖动，无法断言）。 +func TestSweepIngestReviewsDeletesOnlyExpired(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + now := tidyTestTime(t) + retention := DefaultIngestReviewRetention + seedReviewRow(t, store, "expired", now.Add(-retention-time.Second)) + seedReviewRow(t, store, "boundary", now.Add(-retention)) + seedReviewRow(t, store, "recent", now.Add(-time.Hour)) + seedReviewRow(t, store, "now", now) + + stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{Retention: retention}) + if err != nil { + t.Fatalf("SweepIngestReviews: %v", err) + } + if stats.Deleted != 1 { + t.Fatalf("Deleted = %d, want 1（只有 expired 过期）", stats.Deleted) + } + if stats.Truncated != 0 { + t.Fatalf("Truncated = %d, want 0（MaxRows=0 ⇒ 不做条数裁剪）", stats.Truncated) + } + if got, want := remainingReviewIDs(t, store), []string{"boundary", "recent", "now"}; !slices.Equal(got, want) { + t.Fatalf("剩余留痕 = %v, want %v", got, want) + } +} + +// TestSweepIngestReviewsKeepsNewestWithinMaxRows 是条数上限的钉死：保留期那一档 +// 关掉（Retention=0）时，只按条数裁掉最旧的尾巴，**恰好**保留最新 MaxRows 条。 +func TestSweepIngestReviewsKeepsNewestWithinMaxRows(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + now := tidyTestTime(t) + for i := 0; i < 12; i++ { + seedReviewRow(t, store, fmt.Sprintf("r%02d", i), now.Add(time.Duration(i)*time.Minute)) + } + + stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{MaxRows: 5}) + if err != nil { + t.Fatalf("SweepIngestReviews: %v", err) + } + if stats.Truncated != 7 || stats.Deleted != 0 { + t.Fatalf("裁剪统计 = (truncated=%d, deleted=%d), want (7, 0)", stats.Truncated, stats.Deleted) + } + if got := countReviewRows(t, store); got != 5 { + t.Fatalf("剩余行数 = %d, want 5", got) + } + want := []string{"r07", "r08", "r09", "r10", "r11"} + if got := remainingReviewIDs(t, store); !slices.Equal(got, want) { + t.Fatalf("保留的是 %v, want 最新 5 条 %v", got, want) + } +} + +// TestSweepIngestReviewsCapIsExactOnSameTimestampBurst 是上限判据的**最强一格**： +// 明细全在同一 created_at（注入固定时钟的一次性回填）时，只比时间戳的实现会 +// 一条也删不掉 —— 本实现按 `(created_at, rowid)` 元组比较，因此仍然精确裁到上限。 +func TestSweepIngestReviewsCapIsExactOnSameTimestampBurst(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + now := tidyTestTime(t) + for i := 0; i < 10; i++ { + seedReviewRow(t, store, fmt.Sprintf("burst-%02d", i), now) + } + + stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{MaxRows: 3}) + if err != nil { + t.Fatalf("SweepIngestReviews: %v", err) + } + if stats.Truncated != 7 { + t.Fatalf("Truncated = %d, want 7（同一时间戳也必须裁得动）", stats.Truncated) + } + if got := countReviewRows(t, store); got != 3 { + t.Fatalf("剩余行数 = %d, want 3", got) + } + want := []string{"burst-07", "burst-08", "burst-09"} + if got := remainingReviewIDs(t, store); !slices.Equal(got, want) { + t.Fatalf("保留的是 %v, want 最后插入的三条 %v", got, want) + } +} + +// TestSweepIngestReviewsBatchesAreBoundedAndRepeatable 钉住两点： +// ① 一轮清扫**分批**收手（1200 条 = 500+500+200，第三批不满即停，不一次删全表）； +// ② 清扫可重复：再扫一轮删 0 行，绝不误删。 +func TestSweepIngestReviewsBatchesAreBoundedAndRepeatable(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + now := tidyTestTime(t) + retention := DefaultIngestReviewRetention + const expiredRows = 3*ingestReviewSweepBatchSize - 300 // 1200 = 500 + 500 + 200 + for i := 0; i < expiredRows; i++ { + seedReviewRow(t, store, fmt.Sprintf("old-%04d", i), now.Add(-retention-time.Duration(i+1)*time.Second)) + } + + stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{Retention: retention}) + if err != nil { + t.Fatalf("SweepIngestReviews: %v", err) + } + if stats.Deleted != expiredRows { + t.Fatalf("Deleted = %d, want %d", stats.Deleted, expiredRows) + } + if stats.Batches != 3 { + t.Fatalf("Batches = %d, want 3（500 + 500 + 200）", stats.Batches) + } + if got := countReviewRows(t, store); got != 0 { + t.Fatalf("剩余行数 = %d, want 0", got) + } + + again, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{Retention: retention}) + if err != nil { + t.Fatalf("second SweepIngestReviews: %v", err) + } + if again.Deleted != 0 || again.Truncated != 0 { + t.Fatalf("重复清扫 = (deleted=%d, truncated=%d), want (0, 0)", again.Deleted, again.Truncated) + } +} + +// TestSweepIngestReviewsZeroPolicyIsNoop 钉住「档位各自独立」的边界：两档都 ≤0 时 +// 一轮清扫**一行都不动**（不是「偷偷按默认值删」——默认值由调用方显式传入）。 +func TestSweepIngestReviewsZeroPolicyIsNoop(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + now := tidyTestTime(t) + seedReviewRow(t, store, "ancient", now.Add(-10000*time.Hour)) + seedReviewRow(t, store, "fresh", now) + + stats, err := store.SweepIngestReviews(ctx, now, IngestReviewGCPolicy{}) + if err != nil { + t.Fatalf("SweepIngestReviews: %v", err) + } + if stats.Deleted != 0 || stats.Truncated != 0 || stats.Batches != 0 { + t.Fatalf("零值 policy = %+v, want 全零（一行都不动）", stats) + } + if got := countReviewRows(t, store); got != 2 { + t.Fatalf("剩余行数 = %d, want 2", got) + } +} + +// TestReviewerStartSweepsExpiredIngestReviews 是**接线**的证据：清扫不是躺在包里的 +// 死代码 —— Start 拉起它、启动期真的扫过一次（过期行消失、不过期的行原样保留）。 +func TestReviewerStartSweepsExpiredIngestReviews(t *testing.T) { + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + seedReviewRow(t, store, "expired", base.Add(-DefaultIngestReviewRetention-time.Hour)) + seedReviewRow(t, store, "fresh", base) + + r := newTestReviewer(t, store, nil, 0.80) + r.Start(ctx) + t.Cleanup(func() { + stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = r.Stop(stopCtx) + }) + + waitFor(t, func() bool { return countReviewRows(t, store) == 1 }, "启动期清扫删掉过期留痕") + if got := remainingReviewIDs(t, store); !slices.Equal(got, []string{"fresh"}) { + t.Fatalf("剩余留痕 = %v, want [fresh]（只删过期的）", got) + } +} + +// TestReviewerDisabledDoesNotSweep 钉住关闭语义不漂移：`Enabled=false` 时 Start +// 只记一条日志就返回（既有测试已钉「不起任何协程」），因此**不会**顺带清留痕。 +// 这是如实登记的口径而非缺陷：关闭审查后不再产生新留痕，且组合根在关闭时 +// 根本不构造审查服务（cmd/cyrene/wire_memory_review.go）。 +func TestReviewerDisabledDoesNotSweep(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + seedReviewRow(t, store, "expired", base.Add(-DefaultIngestReviewRetention-time.Hour)) + + off, err := NewPreInsertReviewer(ReviewConfig{Enabled: false}, ReviewDeps{Store: store, Log: testLogger()}) + if err != nil { + t.Fatalf("NewPreInsertReviewer: %v", err) + } + off.Start(ctx) + + if got := countReviewRows(t, store); got != 1 { + t.Fatalf("关闭时留痕行数 = %d, want 1（关闭 ⇒ 不扫、也不新增）", got) + } +} diff --git a/internal/runtime/memory/preinsert_migration_test.go b/internal/runtime/memory/preinsert_migration_test.go new file mode 100644 index 0000000..4580d47 --- /dev/null +++ b/internal/runtime/memory/preinsert_migration_test.go @@ -0,0 +1,257 @@ +package memory + +import ( + "context" + "database/sql" + "fmt" + "slices" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// 本文件钉住**迁移 0025 与 Store 自举 DDL 的同形与幂等**（切片 W5）。 +// +// 背景：`memory_ingest_reviews` 原来只由 `memory.NewStore` 的幂等前滚创建，2026-10-06 +// 补登记为迁移 `0025_memory_ingest_reviews`。于是 DDL 有了**两份真源** +// （迁移 + preinsert_schema.go 的自举），而真实启动顺序是「先 MigrateUp、后 NewStore」； +// 既有部署更特殊：**表早就存在**（自举建的），只是迁移账本里没有 0025 —— 那一跳必须 +// 不炸、不丢数据、不改形状。三条判据都在这里： +// +// 1. TestMigration0025CreatesAndRollsBackIngestReviewTable —— 新库上迁移真的建出 +// 表与两条索引，且可回滚（down 删表 ⇒ up 重建空表）； +// 2. TestMigration0025IsIdempotentOverStoreBootstrapTable —— **既有部署**：表已存在 +// 且已有数据时重跑 0025，不报错、不丢行、形状不变； +// 3. TestMigration0025ShapeMatchesStoreBootstrap —— 两份 DDL **逐列逐索引同形** +// （少一列就是「新库有、老库无」的漂移，`CREATE TABLE IF NOT EXISTS` 救不了）。 + +// ingestReviewTable 是被测表名（与迁移、自举 DDL 里的字面量一致）。 +const ingestReviewTable = "memory_ingest_reviews" + +// ingestReviewMigrationID 是 0025 在 schema_migrations 里的记账 id（= 文件名去掉 .up.sql）。 +const ingestReviewMigrationID = "0025_memory_ingest_reviews" + +// openRawMemoryDB 开一个私有的内存库：不迁移、不建 Store，由调用方自己决定走 +// 迁移还是自举（两条路径的对照是本文件的目的）。 +func openRawMemoryDB(t *testing.T) *sql.DB { + t.Helper() + d, err := db.Open(context.Background(), ":memory:", testLogger()) + if err != nil { + t.Fatalf("open in-memory db: %v", err) + } + t.Cleanup(func() { _ = d.Close() }) + return d +} + +// migrateUp 跑全量迁移（含本片新增的 0025）。 +func migrateUp(t *testing.T, d *sql.DB) { + t.Helper() + if err := db.MigrateUp(context.Background(), d, testLogger()); err != nil { + t.Fatalf("MigrateUp: %v", err) + } +} + +// tableExists 报告表是否存在。 +func tableExists(t *testing.T, d *sql.DB, name string) bool { + t.Helper() + var n int + if err := d.QueryRowContext(context.Background(), + `SELECT count(*) FROM sqlite_master WHERE type='table' AND name = ?`, name).Scan(&n); err != nil { + t.Fatalf("check table %s: %v", name, err) + } + return n == 1 +} + +// tableColumns 把表结构渲染成可比较的字符串切片（name|type|notnull|default|pk）， +// 顺序即建表顺序。 +func tableColumns(t *testing.T, d *sql.DB, table string) []string { + t.Helper() + rows, err := d.QueryContext(context.Background(), "PRAGMA table_info("+table+")") + if err != nil { + t.Fatalf("table_info(%s): %v", table, err) + } + defer func() { _ = rows.Close() }() + out := make([]string, 0) + for rows.Next() { + var ( + cid int + name, typ string + notNull int + dflt sql.NullString + pk int + ) + if err := rows.Scan(&cid, &name, &typ, ¬Null, &dflt, &pk); err != nil { + t.Fatalf("scan table_info(%s): %v", table, err) + } + out = append(out, fmt.Sprintf("%d|%s|%s|%d|%s|%d", cid, name, typ, notNull, dflt.String, pk)) + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate table_info(%s): %v", table, err) + } + return out +} + +// tableIndexes 把表自己建的索引渲染成 `name|unique`（按名字排序，忽略建索引顺序）。 +// `sqlite_autoindex_*` 是 TEXT PRIMARY KEY 的隐式索引，不由我们的 DDL 创建， +// 故排除 —— 主键形状已由 tableColumns 的 pk 列覆盖。 +func tableIndexes(t *testing.T, d *sql.DB, table string) []string { + t.Helper() + rows, err := d.QueryContext(context.Background(), "PRAGMA index_list("+table+")") + if err != nil { + t.Fatalf("index_list(%s): %v", table, err) + } + defer func() { _ = rows.Close() }() + out := make([]string, 0) + for rows.Next() { + var ( + seq int + name string + unique int + origin string + partial int + ) + if err := rows.Scan(&seq, &name, &unique, &origin, &partial); err != nil { + t.Fatalf("scan index_list(%s): %v", table, err) + } + if strings.HasPrefix(name, "sqlite_autoindex_") { + continue + } + out = append(out, fmt.Sprintf("%s|%d", name, unique)) + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate index_list(%s): %v", table, err) + } + slices.Sort(out) + return out +} + +// countRawReviewRows 直读留痕行数（不经 Store）。 +func countRawReviewRows(t *testing.T, d *sql.DB) int { + t.Helper() + var n int + if err := d.QueryRowContext(context.Background(), + "SELECT count(*) FROM "+ingestReviewTable).Scan(&n); err != nil { + t.Fatalf("count rows in %s: %v", ingestReviewTable, err) + } + return n +} + +// TestMigration0025CreatesAndRollsBackIngestReviewTable 钉住迁移本身：up 建表 + 两条 +// 索引、down 删表、再 up 能重建（回滚不留半状态）。 +func TestMigration0025CreatesAndRollsBackIngestReviewTable(t *testing.T) { + t.Parallel() + ctx := context.Background() + d := openRawMemoryDB(t) + migrateUp(t, d) + + if !tableExists(t, d, ingestReviewTable) { + t.Fatalf("迁移后应存在 %s", ingestReviewTable) + } + wantIndexes := []string{ + "idx_memory_ingest_reviews_created_at|0", + "idx_memory_ingest_reviews_item_id|0", + } + if got := tableIndexes(t, d, ingestReviewTable); !slices.Equal(got, wantIndexes) { + t.Fatalf("索引 = %v, want %v", got, wantIndexes) + } + cols := tableColumns(t, d, ingestReviewTable) + for _, want := range []string{"id", "item_id", "created_at"} { + if !slices.ContainsFunc(cols, func(c string) bool { return strings.Contains(c, "|"+want+"|") }) { + t.Fatalf("列 %s 缺席：%v", want, cols) + } + } + + // 回滚：只回滚最高的一条（0025），表随之消失。 + if err := db.MigrateDown(ctx, d, 1, testLogger()); err != nil { + t.Fatalf("MigrateDown(1): %v", err) + } + if tableExists(t, d, ingestReviewTable) { + t.Fatalf("回滚后 %s 应已被删除", ingestReviewTable) + } + + // 再 up：表重建且为空（down/up 前后不留残留）。 + migrateUp(t, d) + if !tableExists(t, d, ingestReviewTable) { + t.Fatalf("再次迁移后应存在 %s", ingestReviewTable) + } + if got := countRawReviewRows(t, d); got != 0 { + t.Fatalf("重建后的行数 = %d, want 0", got) + } + if got := tableColumns(t, d, ingestReviewTable); !slices.Equal(got, cols) { + t.Fatalf("重建后的表结构 = %v, want %v", got, cols) + } +} + +// TestMigration0025IsIdempotentOverStoreBootstrapTable 模拟**既有部署**的真实状态： +// 表已由自举建好、里面已有留痕，而迁移账本里没有 0025（本次升级的全部增量就是 +// 重跑这一条）。判据：不报错、数据不丢、形状不变。 +func TestMigration0025IsIdempotentOverStoreBootstrapTable(t *testing.T) { + t.Parallel() + ctx := context.Background() + d := openRawMemoryDB(t) + migrateUp(t, d) + + // 造出「既有部署」：把 0025 的成果抹掉（表 + 账本记录），让自举重新建表。 + if _, err := d.ExecContext(ctx, "DROP TABLE "+ingestReviewTable); err != nil { + t.Fatalf("drop table: %v", err) + } + if _, err := d.ExecContext(ctx, + "DELETE FROM schema_migrations WHERE id = ?", ingestReviewMigrationID); err != nil { + t.Fatalf("unrecord migration: %v", err) + } + + store, err := NewStore(ctx, d, testLogger()) + if err != nil { + t.Fatalf("NewStore（自举建表）: %v", err) + } + seedReviewRow(t, store, "kept", tidyTestTime(t)) + beforeCols := tableColumns(t, d, ingestReviewTable) + beforeIndexes := tableIndexes(t, d, ingestReviewTable) + + // 升级路径：只有 0025 待应用（表已存在）。 + migrateUp(t, d) + + if got := countRawReviewRows(t, d); got != 1 { + t.Fatalf("迁移后行数 = %d, want 1（既有留痕一行都不能丢）", got) + } + if got := tableColumns(t, d, ingestReviewTable); !slices.Equal(got, beforeCols) { + t.Fatalf("迁移改变了表结构：\n got=%v\nwant=%v", got, beforeCols) + } + if got := tableIndexes(t, d, ingestReviewTable); !slices.Equal(got, beforeIndexes) { + t.Fatalf("迁移改变了索引：\n got=%v\nwant=%v", got, beforeIndexes) + } + + // 账本幂等：已记账后再跑一次是纯空转。 + migrateUp(t, d) + if got := countRawReviewRows(t, d); got != 1 { + t.Fatalf("重复迁移后行数 = %d, want 1", got) + } +} + +// TestMigration0025ShapeMatchesStoreBootstrap 是**漂移护栏**：迁移建出的表与自举 +// 建出的表必须逐列、逐索引同形。它红的时候，正确的修法是让两份 DDL 对齐， +// 而不是放宽这条断言。 +func TestMigration0025ShapeMatchesStoreBootstrap(t *testing.T) { + t.Parallel() + ctx := context.Background() + + migrated := openRawMemoryDB(t) + migrateUp(t, migrated) + + bootstrapped := openRawMemoryDB(t) + if _, err := NewStore(ctx, bootstrapped, testLogger()); err != nil { + t.Fatalf("NewStore: %v", err) + } + + gotCols := tableColumns(t, migrated, ingestReviewTable) + wantCols := tableColumns(t, bootstrapped, ingestReviewTable) + if !slices.Equal(gotCols, wantCols) { + t.Fatalf("迁移与自举的表结构不一致（两份 DDL 必须逐字同形）：\n迁移 =%v\n自举 =%v", gotCols, wantCols) + } + gotIdx := tableIndexes(t, migrated, ingestReviewTable) + wantIdx := tableIndexes(t, bootstrapped, ingestReviewTable) + if !slices.Equal(gotIdx, wantIdx) { + t.Fatalf("迁移与自举的索引不一致：\n迁移 =%v\n自举 =%v", gotIdx, wantIdx) + } +} diff --git a/internal/runtime/memory/preinsert_schema.go b/internal/runtime/memory/preinsert_schema.go index 0f50073..70de369 100644 --- a/internal/runtime/memory/preinsert_schema.go +++ b/internal/runtime/memory/preinsert_schema.go @@ -14,11 +14,13 @@ import ( // 语义不同、生命周期也不同 —— 审查是已经生效的事实留痕，不是待批准的提议。 // 两表共用同一套 id/时间戳口径（主键 id、created_at），词表口径见各列注释。 // -// 表结构的权威机制仍是 internal/db 的迁移（tidy 的 0019 是范例）；本表**尚未 -// 登记迁移**，改由 NewStore 幂等前滚，与 memory_items / tidy 两表同一手法 -// （store_schema.go 的口径：记忆存储必须自包含、可在任意 *sql.DB 上使用）。 -// 迁移登记留给主会话（见 MTB-report.md 的「未做」一节），以免与并行切片 -// 抢迁移编号。 +// 表结构的权威机制是 internal/db 的迁移（`0025_memory_ingest_reviews`，切片 W5 补登记）； +// 本处仍保留幂等前滚，与 memory_items / tidy 两表同一手法（store_schema.go 的口径： +// 记忆存储必须自包含、可在任意 *sql.DB 上使用）。两份 DDL **必须逐字同形**： +// 少一列就是「新库有、老库无」的漂移（既有部署的表是这里建的，`CREATE TABLE +// IF NOT EXISTS` 不会给它们补列），由 preinsert_migration_test.go 逐列/逐索引比对钉住。 +// +// 保留期清扫（按 created_at 删过期行 + 条数上限）见 preinsert_gc.go。 // createMemoryIngestReviewsTable 是入库前审查的结构化留痕（design/23 §15 D3-8）。 // 刻意**不存记忆正文**：命中的既有 id 足以让 owner-only 的读侧回表取正文，