From cc6904b1b4782d7281c17cd3333b5ee8ebbf636b Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 26 Oct 2026 13:09:30 +0800 Subject: [PATCH] =?UTF-8?q?feat(memory):=20retention=5Ftier=20=E6=A1=A3?= =?UTF-8?q?=E4=BD=8D=E5=88=97=20+=20defer=20=E4=B8=8E=E6=A1=A3=E4=BD=8D?= =?UTF-8?q?=E5=90=88=E5=B9=B6=E4=B8=BA=E5=90=8C=E4=B8=80=E5=88=A4=E5=AE=9A?= =?UTF-8?q?=EF=BC=88M37=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 主人 2026-10-26 对「记忆分型」裁定「两列」：kind（关于什么）保留，新增 retention_tier（留多久 / 干什么用）。本片把这条轴从列、判定、写入、读侧过滤 一路打通，并把「判第五态 defer」与「判 retention_tier=temp」合并成一个判定。 结构（迁移 0026 + 自举前滚）： - memory_items 新增 retention_tier TEXT NOT NULL DEFAULT ''（空串 = 尚未分型， 历史行与受限来源行），索引 idx_memory_items_retention_tier；up/down 均实跑过 （在隔离的生产库副本上 migrate → 建 Store → down → 再 up，5073 行零丢失）。 - NewStore 幂等前滚该列 + 建索引（ensureRetentionSchema），与迁移同形由测试钉住。 判定（retention.go，全仓唯一入口）： - RetentionTier 闭集 temp/daily/work/long/knowledge + 空串未分型；与 kind 正交。 - resolveRetention 的优先级：调用方显式档位 > 显式有效期（⇒ temp）> 受限来源 不分型 > 留存规则 N1~N7 ⇒ temp > 默认 long。判据靠元数据，规则不许推翻显式档位。 - tempDeadline 是「档位 → expires_at」的唯一换算：temp ⇔ 有效期非空，两列同一笔 INSERT 落下（M2 登记过的「INSERT 与补写之间的短窗口」就此关闭）。 合并（defer ⇔ temp）： - judgeIngest 不再跑留存规则，只读 Add 落定的 item.RetentionTier；规则调用点因此 全仓唯一（源码级断言钉住）。 - 判定不再依赖同步闸是否开启：闸未接/关着时档位与有效期照样落库。 - 异步层对已是 temp 的行不重复改写（避免 TTL 随审查延迟漂移）。 - IngestRuleVersion d3.3 → d3.4（判据作用范围变了，按既有硬规矩升版）。 读侧： - 四条列表读路径的 SELECT 与 scanItem 读出 retention_tier（14 列）；Get 同样读回。 - 新增 ListByPersonaTiers / ListScopeTiers / ListTiers / ListProjectTiers（可选档位 过滤，缺省不过滤 ⇒ 历史行照常返回；显式传空串只取未分型；非法档位报错）。 原方法签名逐字不变 —— 网关 MemoryStore 接口钉着它们，可变参数会编译不过。 - store.go 的读路径拆到 store_list.go，守 500 行上限。 回归：go build ./... + go vet ./... + go test ./... 全绿（内存包另有 -race）。 --- .../0026_memory_retention_tier.down.sql | 17 + .../0026_memory_retention_tier.up.sql | 43 ++ internal/runtime/memory/extract_retention.go | 15 +- .../runtime/memory/extract_retention_test.go | 20 +- internal/runtime/memory/preinsert.go | 27 +- .../runtime/memory/preinsert_defer_test.go | 32 +- internal/runtime/memory/preinsert_gate.go | 61 +-- internal/runtime/memory/preinsert_judge.go | 23 +- .../memory/preinsert_migration_test.go | 7 +- internal/runtime/memory/preinsert_store.go | 47 +- internal/runtime/memory/retention.go | 187 ++++++++ .../runtime/memory/retention_judge_test.go | 180 +++++++ .../memory/retention_migration_test.go | 241 ++++++++++ internal/runtime/memory/retention_schema.go | 55 +++ internal/runtime/memory/retention_test.go | 447 ++++++++++++++++++ internal/runtime/memory/rules.go | 52 +- internal/runtime/memory/store.go | 144 +++--- internal/runtime/memory/store_list.go | 144 ++++++ internal/runtime/memory/store_rows.go | 9 +- internal/runtime/memory/store_schema.go | 12 +- internal/runtime/memory/tidy_dedupe.go | 4 +- internal/runtime/memory/tidy_store.go | 2 +- internal/runtime/memory/timegate.go | 4 +- 23 files changed, 1592 insertions(+), 181 deletions(-) create mode 100644 internal/db/migrations/0026_memory_retention_tier.down.sql create mode 100644 internal/db/migrations/0026_memory_retention_tier.up.sql create mode 100644 internal/runtime/memory/retention.go create mode 100644 internal/runtime/memory/retention_judge_test.go create mode 100644 internal/runtime/memory/retention_migration_test.go create mode 100644 internal/runtime/memory/retention_schema.go create mode 100644 internal/runtime/memory/retention_test.go create mode 100644 internal/runtime/memory/store_list.go diff --git a/internal/db/migrations/0026_memory_retention_tier.down.sql b/internal/db/migrations/0026_memory_retention_tier.down.sql new file mode 100644 index 0000000..5a6b517 --- /dev/null +++ b/internal/db/migrations/0026_memory_retention_tier.down.sql @@ -0,0 +1,17 @@ +-- 0026 的回滚：先删索引、再删列（与 0019 同一顺序口径 —— SQLite 拒绝 DROP COLUMN +-- 删除一个仍被索引引用的列）。 +-- +-- 回滚的代价（如实登记）：`retention_tier` 这一列的值**全部丢失**。它对本迁移之前 +-- 写入的行本来就恒为 `''`（未分型），因此丢的只是「本迁移之后被判出来的档位」。 +-- 真正不可逆的只有一条：**判成 temp 的行不会被自动改回长期层** —— 它们的 +-- `expires_at` 是另一列（0019 就存在），本回滚**刻意不动它**： +-- +-- - 若回滚时顺手清空 `expires_at`，一次 schema 回滚就会把已经判定的临时记忆 +-- 全部升回永久（那是改数据，不是回滚结构）； +-- - 若不清空，回滚后这些行仍是「有时效的行」，读侧时效闸（纯 SQL，不依赖本列） +-- 照旧隐藏它们，后台清扫照旧删除它们 —— 即**行为不变**。 +-- +-- 因此本回滚只撤结构、绝不改写任何一行的时效数据；要重判档位，重跑判定即可 +-- （`retention_tier` 列会以默认值 `''` 重建，历史行回到「未分型」）。 +DROP INDEX IF EXISTS idx_memory_items_retention_tier; +ALTER TABLE memory_items DROP COLUMN retention_tier; diff --git a/internal/db/migrations/0026_memory_retention_tier.up.sql b/internal/db/migrations/0026_memory_retention_tier.up.sql new file mode 100644 index 0000000..a0e3d6e --- /dev/null +++ b/internal/db/migrations/0026_memory_retention_tier.up.sql @@ -0,0 +1,43 @@ +-- 0026_memory_retention_tier：「留多久 / 干什么用」这根轴（主人 2026-10-26 裁定「两列」）。 +-- +-- 为什么是**新列**而不是复用 kind：`kind`（relationship / world_event）回答的是 +-- 「关于什么」，`scope`（user / persona / project）回答的是「谁的」，本条新增的 +-- `retention_tier` 回答的是「留多久 / 干什么用」。三根轴**正交、互不推导**： +-- 一条关系记忆可以是临时记忆，也可以是长期记忆；把「临时性」塞进 `kind` 会让 +-- 「一条关系记忆又是临时的」无法表达（审计 `记忆存量体检-2026-10-26.md` 卷四的 +-- 三条路对比，主人选了第 1 条）。 +-- +-- 🔴 默认值 `''` = **尚未分型**，不是第六档业务档位： +-- ① 本迁移之前的历史行全部落在 `''`（本迁移**刻意不做回填**：那是一次写全表的 +-- 判定动作，属于整理侧的第二现场，不该藏在迁移里，也不该在启动路径上静默跑）； +-- ② 受限来源（群聊 / 陌生人）的新行同样不分型（它们的留存由来源面决定，后台到期 +-- 清扫也会跳过它们 —— 判成 temp 就是造一条「对读面隐藏却永不删除」的行）。 +-- 读侧因此**必须**能区分「未分型」与「已分型」：缺省不过滤（历史行照常返回）， +-- 显式按 `''` 过滤即「只看未分型的行」（回填分型的入口）。 +-- +-- 档位闭集（代码侧的定义是 internal/runtime/memory/retention.go 的 RetentionTier）： +-- temp 临时层：唯一带时效的档位，`retention_tier='temp'` ⇔ `expires_at` +-- 非空且落在临时 TTL 内（两列由同一个判定、同一笔 INSERT 落下） +-- daily 日常层（一天之内的琐事 / 生活节奏；**不是** scope=persona 的「角色日常」） +-- work 工作层（任务推进 / 工单 / 项目过程） +-- long 长期层（稳定的偏好 / 关系 / 事实；入库判定的默认落点） +-- knowledge 知识库层（外部资料 / 规格 / 文档性质的记忆行） +-- '' 尚未分型（见上） +-- +-- 幂等性：`ALTER TABLE ADD COLUMN` 在 SQLite 里**没有** IF NOT EXISTS 形式，因此本 +-- 迁移与 0019_memory_tidy 同一手法 —— 依赖启动顺序「先 MigrateUp、后 NewStore」 +-- （cmd/cyrene/wire.go）：生产库里这一步先跑，随后 NewStore 的自举前滚 +-- （ensureRetentionTierColumn）走到 "duplicate column name" 是预期路径。反过来 +-- （先自举、后迁移）也只在同一条账本跳一次。列与索引的形状必须与自举 DDL +-- （internal/runtime/memory/store_schema.go）**逐字同形**，由 +-- internal/runtime/memory/retention_migration_test.go 逐列/逐索引钉住。 +-- +-- 命名口径：表名复数、主键 `id`、索引 `idx_{table}_{col}`（援引 AGENTS §10 已登记 +-- 例外，与既有迁移一致）。 +ALTER TABLE memory_items ADD COLUMN retention_tier TEXT NOT NULL DEFAULT ''; + +-- 三条列表读路径的可选档位过滤（`retention_tier IN (...)`）与「只看未分型的行」 +-- （回填入口）都吃这条索引。单列而不是复合列：这一轴的查询形状是 +-- 「跨角色按档位聚合」（面板 / 回填），不绑定 persona。 +CREATE INDEX IF NOT EXISTS idx_memory_items_retention_tier + ON memory_items (retention_tier); diff --git a/internal/runtime/memory/extract_retention.go b/internal/runtime/memory/extract_retention.go index af506f4..aab2a11 100644 --- a/internal/runtime/memory/extract_retention.go +++ b/internal/runtime/memory/extract_retention.go @@ -12,11 +12,11 @@ import ( // 不能想到什么都往记忆里写**」与「**临时记忆到期删除**」。 // // 🔴 落点硬约束（design/03 §10.3）：**不许新造第二条写路径**。本闸**不新开入口**， -// 而是接进既有的唯一写路径 —— `judgeIngest`（preinsert_judge.go）在**候选比对 -// 之前**先过一遍这里，命中即返回与 design/23 §18.1 第五态**同一枚举 `defer`、 -// 同一落点 `expires_at`**。于是「提取得宽」由**同一条**入库路径收敛，不存在 -// 绕过闸的写口；命中者**不进 T1**（§10.6：规则闸拦下的不进 T1，避免「模型推翻 -// 规则」的架构回归）。 +// 它是 `resolveRetention`（retention.go）里的一步 —— 那是「留多久」轴的**唯一**判定： +// 命中即判 `retention_tier=temp`，而第五态 `defer` 只是「档位是 temp ∧ 关系判为 +// 真新」的另一种说法（preinsert_judge.go 只读档位，不再跑本文件）。于是「提取得宽」 +// 由**同一条**入库路径收敛，不存在绕过闸的写口；命中者**不进 T1**（§10.6：规则闸 +// 拦下的不进 T1，避免「模型推翻规则」的架构回归）。 // // 判据为什么必须是纯函数而不是提示词（design/03 §10.2）：现行全仓唯一的留存判据 // 是一句自然语言（`llmextractor.go` 的 extractSystemPrompt），那**违反 AGENTS §9 @@ -98,11 +98,14 @@ const ( ) // judgeExtractionRetention 对一条新记忆的正文施加提取留存规则（design/03 §10.5 -// 的 N1~N5 与 M34 新增的 N6/N7）。命中返回该规则码（调用方据此判 `defer`）， +// 的 N1~N5 与 M34 新增的 N6/N7）。命中返回该规则码（调用方据此定 temp 档）， // 否则返回 retentionNone。 // // 它**不读任何状态、不写任何东西、零模型调用**：入参只有正文，出参只有规则码， // 因此在写路径上是微秒级且可完全单测（design/03 §10.3 A 级的「零成本」要求）。 +// 🔴 全仓**唯一**调用点是 `resolveRetention`（retention.go）：任何地方再跑一遍 +// 就是「两套规则各判一遍」—— 那正是 M27 体检发现缺口（1082 次只判出 1 条）的成因， +// retention_test.go 有一条源码级断言钉住这个唯一性。 func judgeExtractionRetention(content string) retentionRule { text := normalizeRetentionText(content) if text == "" { diff --git a/internal/runtime/memory/extract_retention_test.go b/internal/runtime/memory/extract_retention_test.go index 7a3444e..4ebe6f4 100644 --- a/internal/runtime/memory/extract_retention_test.go +++ b/internal/runtime/memory/extract_retention_test.go @@ -195,13 +195,27 @@ func TestExtractionRetention_UpperCaseASCII(t *testing.T) { } } -// TestJudgeIngestDefersLowValueItem 是**接线**断言：留存规则闸真的接在判定链上， -// 命中的新条目拿到第五态 `defer` + 留痕原因码 `t0_low_value`，且**不进灰区** +// TestJudgeIngestDefersLowValueItem 是**接线**断言：档位判定真的接在判定链上， +// 被判成 temp 的新条目拿到第五态 `defer` + 留痕原因码 `t0_low_value`，且**不进灰区** // （design/03 §10.6：规则闸拦下的不进 T1）。 +// +// 🔴 切片 M37 起，`judgeIngest` **不再自己跑留存规则** —— 档位由 `resolveRetention` +// 一次判完（commit 前的写法是这里再调一次 `judgeExtractionRetention`，那正是 +// 「两套规则各判一遍」）。因此本用例先把档位判出来放进条目（与 `Store.Add` 的 +// 做法逐字一致），再断言判定链**只读**这一位。变异（把规则调用搬回 judgeIngest） +// 会让 retention_test.go 的唯一性断言先红。 func TestJudgeIngestDefersLowValueItem(t *testing.T) { t.Parallel() item := habitItem("谢谢啦", tidyTestTime(t), 0.9) - // 故意给一条「相似候选」：它本该进灰区，但留存规则先命中 ⇒ 灰区必须为空， + verdict, err := resolveRetention(item) + if err != nil { + t.Fatalf("resolveRetention: %v", err) + } + if verdict.Tier != RetentionTemp { + t.Fatalf("低价值条目档位 = %q, want %q", verdict.Tier, RetentionTemp) + } + item.RetentionTier = verdict.Tier + // 故意给一条「相似候选」：它本该进灰区，但 temp 档先命中 ⇒ 灰区必须为空， // 于是 T1 一个 token 都花不出去。 cands := []ingestCandidate{{item: habitItem("谢谢", tidyTestTime(t), 0.9), score: 0.9, mode: tidyRecallLexical}} v := judgeIngest(item, cands) diff --git a/internal/runtime/memory/preinsert.go b/internal/runtime/memory/preinsert.go index 9acbbff..ea554ba 100644 --- a/internal/runtime/memory/preinsert.go +++ b/internal/runtime/memory/preinsert.go @@ -90,7 +90,14 @@ const ( // （`t0_low_value`）与 T1 的同一次调用（`t1_defer`）都能给出它，落点 = // 照常插入 + 写短 `expires_at`。**升版是硬要求**：同一对条目在 d3.2 下是 // `create`、在 d3.3 下可能是 `defer`，不升版新旧留痕无法区分（§18.4）。 -const IngestRuleVersion = "d3.3" +// - `d3.4`（切片 M37）把「判 `defer`」与「判 `retention_tier=temp`」**合并成一个 +// 判定**（迁移 0026）：档位由 `Store.Add` 的 `resolveRetention` 一次落定 +// （调用方显式档位 > 显式有效期 > 受限来源不分型 > 留存规则 → temp > 默认 long）， +// `defer` 从此只是「档位是 temp ∧ 关系判为真新」，闸与异步审查都不再各跑一遍 +// 规则。**规则内容与阈值未改**（同一份 N1~N7），但**判据的作用范围变了**： +// 档位与 `expires_at` 现在对**每一行**在同一笔 INSERT 里落下，且不再依赖同步闸 +// 是否开启 —— 同一对条目在 d3.3 与 d3.4 下的落库形态因此可能不同，按规矩升版。 +const IngestRuleVersion = "d3.4" // 调优旋钮（ReviewConfig）与它们的默认值常量在 preinsert_config.go。 @@ -464,10 +471,20 @@ func (r *PreInsertReviewer) applyVerdict(ctx context.Context, v ingestVerdict, i } if v.Decision == IngestDefer { // 第五态（design/23 §18.1）：条目**早已落库**（异步审查跑在写入之后）， - // 这里只把 `expires_at` 写成「现在 + 临时 TTL」把它放进临时层。 - // **不删行、不改正文、不碰任何其它行** ⇒ 回滚 = 清空该列（§18.4）。 - // 写不进去就如实返回错误（fail-safe：行留在长期层 = 宁可多留）。 - if err := r.deps.Store.setIngestExpiry(ctx, item.ID, expiryText(r.deferDeadline(now))); err != nil { + // 这里把「档位 + 有效期」改成临时层（**同一条语句**写，两者不可能分叉）。 + // 候选是 T1 在同一次调用里判的 defer（`t1_defer`）。 + // + // 🔴 **已经落成临时层的行不重复改写**：`Store.Add` 的单一判定已经写过这两列 + // （`item.RetentionTier` 随队列传到这里），再写一次只会把有效期从 + // 「判定时刻 + TTL」推迟到「审查时刻 + TTL」—— 那是一次会漂移的口径。 + // + // **不删行、不改正文、不碰任何其它行** ⇒ 回滚 = 改回 long + 清空有效期 + // （§18.4）。写不进去就如实返回错误（fail-safe：行留在长期层 = 宁可多留）。 + if item.RetentionTier == RetentionTemp { + return nil, nil + } + if err := r.deps.Store.setIngestRetention(ctx, item.ID, + RetentionTemp, expiryText(r.deferDeadline(now))); err != nil { return nil, err } return nil, nil diff --git a/internal/runtime/memory/preinsert_defer_test.go b/internal/runtime/memory/preinsert_defer_test.go index ec03015..1727335 100644 --- a/internal/runtime/memory/preinsert_defer_test.go +++ b/internal/runtime/memory/preinsert_defer_test.go @@ -19,7 +19,7 @@ import ( // 2. 写进去的 `expires_at` 与读侧时效闸**同一套规范文本** // （定宽 30 字节 UTC —— 写错布局会让闸静默判错，本仓有历史事故）； // 3. TTL 坏值（≤0）**绝不放行**：回落代码默认 7 天 + WARN； -// 4. `IngestRuleVersion` 升到 `d3.3`（改判定规则必升版，否则新旧留痕无法区分）； +// 4. `IngestRuleVersion` 升到 `d3.4`（改判定规则必升版，否则新旧留痕无法区分）； // 5. defer 是**软动作**：回滚 = 清空 `expires_at`，行与正文一字不动。 // readIngestExpiry 直读一行记忆的 `expires_at`（该列不在 Store.Get 的读侧面上： @@ -79,7 +79,7 @@ func ttlPanelPtr(v int) *int { return &v } // TestIngestGateDeferWritesShortExpiresAt 是任务书验收点①的**闸路径**： // 低价值条目照常 active 落库（一个字不丢），`expires_at` = 判定时刻 + 临时 TTL， -// 留痕是 `defer` / `t0_low_value` / 规则版本 `d3.3`。 +// 留痕是 `defer` / `t0_low_value` / 规则版本 `d3.4`。 func TestIngestGateDeferWritesShortExpiresAt(t *testing.T) { t.Parallel() store := newTestStore(t) @@ -107,8 +107,8 @@ func TestIngestGateDeferWritesShortExpiresAt(t *testing.T) { if rev.Decision != IngestDefer || rev.Reason != retentionLowValueReason { t.Fatalf("闸留痕 = %+v, want decision=defer reason=%q", rev, retentionLowValueReason) } - if rev.RuleVer != "d3.3" { - t.Fatalf("闸留痕规则版本 = %q, want d3.3", rev.RuleVer) + if rev.RuleVer != "d3.4" { + t.Fatalf("闸留痕规则版本 = %q, want d3.4", rev.RuleVer) } if s := r.Stats(); s.GateDeferred != 1 { t.Fatalf("闸计数 = %+v, want gate_deferred=1", s) @@ -211,8 +211,8 @@ func TestDeferExpiryHiddenByReadGateAndRollbackRestores(t *testing.T) { t.Fatal("刚落的临时记忆应当在有效期内可见（TTL = 1h）") } - if err := store.setIngestExpiry(ctx, got.ID, expiryText(time.Now().UTC().Add(-time.Minute))); err != nil { - t.Fatalf("setIngestExpiry: %v", err) + if err := store.setIngestRetention(ctx, got.ID, RetentionTemp, expiryText(time.Now().UTC().Add(-time.Minute))); err != nil { + t.Fatalf("setIngestRetention: %v", err) } if personaListHas(t, store, got.PersonaID, got.ID) { t.Fatal("已到期的临时记忆不该出现在读路径上（时效闸失效）") @@ -222,8 +222,8 @@ func TestDeferExpiryHiddenByReadGateAndRollbackRestores(t *testing.T) { } // 回滚（design/23 §18.4）：清空 expires_at ⇒ 升回长期，立即可见。 - if err := store.setIngestExpiry(ctx, got.ID, ""); err != nil { - t.Fatalf("setIngestExpiry(清空): %v", err) + if err := store.setIngestRetention(ctx, got.ID, RetentionLong, ""); err != nil { + t.Fatalf("setIngestRetention(回滚): %v", err) } if !personaListHas(t, store, got.PersonaID, got.ID) { t.Fatal("清空 expires_at 后应当升回长期（回滚是无损的）") @@ -268,12 +268,16 @@ func TestIngestGatePlanExpiresAtUsesSameLayoutAsTimeGate(t *testing.T) { } } -// TestIngestRuleVersionIsD33 钉住任务书验收点④：改判定规则必升版 +// TestIngestRuleVersionIsD34 钉住任务书验收点④：改判定规则必升版 // （design/23 §18.4），否则新旧留痕无法区分。 -func TestIngestRuleVersionIsD33(t *testing.T) { +// +// d3.4（切片 M37）把「判 defer」与「判 retention_tier=temp」合并成同一个判定， +// 规则内容未改但判据的作用范围变了（每一行都在同一笔 INSERT 里落档位与有效期， +// 且不再依赖同步闸是否开启），因此按规矩升版。 +func TestIngestRuleVersionIsD34(t *testing.T) { t.Parallel() - if IngestRuleVersion != "d3.3" { - t.Fatalf("IngestRuleVersion = %q, want d3.3（第五态改变了判定闭集）", IngestRuleVersion) + if IngestRuleVersion != "d3.4" { + t.Fatalf("IngestRuleVersion = %q, want d3.4（档位与 defer 合并为同一判定）", IngestRuleVersion) } } @@ -347,12 +351,12 @@ func TestNewPreInsertReviewerNormalizesBadTempTTL(t *testing.T) { } // TestIngestRuleVersionIsNotInTheTidyIdempotencyKey 是**防漂移缝断言**，回答一条 -// 跨机制的实体问题：`IngestRuleVersion` 由 d3.2 升到 d3.3 之后，**既有 pending 整理 +// 跨机制的实体问题：`IngestRuleVersion` 由 d3.2 升到 d3.3（现为 d3.4）之后，**既有 pending 整理 // 候选会不会凭幂等键复制出重复行**？ // // 答案：**不会**，因为这是两个独立的常量 —— // -// IngestRuleVersion ("d3.3") 入库审查（本切片）→ 只进 memory_ingest_reviews.rule_version +// IngestRuleVersion ("d3.4") 入库审查（本切片）→ 只进 memory_ingest_reviews.rule_version // TidyRuleVersion ("t0.1") 整理（design/23 §17.6）→ **进整理候选的幂等键** // // 幂等键的算式是 `tidyIdempotencyKey`（tidy_candidate.go）：hash(persona_id, scope, diff --git a/internal/runtime/memory/preinsert_gate.go b/internal/runtime/memory/preinsert_gate.go index efd9bc5..dc43d89 100644 --- a/internal/runtime/memory/preinsert_gate.go +++ b/internal/runtime/memory/preinsert_gate.go @@ -116,6 +116,10 @@ type ingestGatePlan struct { // 它是**时刻**而不是「TTL 秒数」：判定的那一刻就把 TTL 换算掉，落库侧因此 // 不需要认识任何配置，也不会出现「写的时候又算一次 TTL」的双口径。 ExpiresAt time.Time + // Retention 是这一行落库时写进 `retention_tier` 的档位（迁移 0026）。 + // 它与 ExpiresAt 由 planIngest 的**同一处换算**同时填（temp ⇔ 有到期时刻）， + // 因此两列不可能分叉 —— 闸的任何分支都不需要自己管这一位。 + Retention RetentionTier } // expiresAt 返回新行落库时的 `expires_at` 文本，与 `insertStatus()` / @@ -129,6 +133,13 @@ func (p ingestGatePlan) expiresAt() string { return expiryText(p.ExpiresAt) } +// retention 返回新行落库时的 `retention_tier` 文本（迁移 0026 的「留多久」轴）。 +// 与 expiresAt() 是同一组的两个落库值：`Retention == temp` ⇔ `ExpiresAt` 非零， +// 由 planIngest 的同一处换算保证（见 retention.go 的 tempDeadline）。 +func (p ingestGatePlan) retention() string { + return string(p.Retention) +} + // absorbedID 返回应该写进新行 `merged_into` 的存活行 id。 func (p ingestGatePlan) absorbedID() string { if p.Action == ingestPlanAbsorbNew { @@ -148,12 +159,21 @@ func (p ingestGatePlan) insertStatus() string { // planIngest 在写入路径上跑同步入库闸。未接线审查（store 上没有 reviewer）、 // 总开关关闭、或闸被显式关掉时，它返回「只插不审」的 passthrough 计划 —— // 写入路径因此逐字回到本次改动之前。 +// +// 🔴 **档位与有效期在这里统一落到计划上**（唯一换算点，见 retention.go 的 +// tempDeadline）：无论走闸还是走 passthrough，落库值都从这一处出，因此 +// 「闸关闭时档位就丢失」这种分叉在结构上不可能发生 —— M27 体检里 +// 「`expires_at` 全表为空」的那条缺口，一半原因就是这条换算曾经只挂在闸的分支里。 func (s *Store) planIngest(ctx context.Context, item MemoryItem) ingestGatePlan { - r := s.ingestReviewer() - if !r.gateEnabled() { - return ingestGatePlan{Action: ingestPlanInsert, Decision: IngestCreate, Async: true} + var plan ingestGatePlan + if r := s.ingestReviewer(); !r.gateEnabled() { + plan = ingestGatePlan{Action: ingestPlanInsert, Decision: IngestCreate, Async: true} + } else { + plan = r.gate(ctx, item) } - return r.gate(ctx, item) + plan.Retention = item.RetentionTier + plan.ExpiresAt = s.tempDeadline(item) + return plan } // gate 是闸本体：在**有界**预算内召回同分区候选、用确定性规则判定，并给出 @@ -215,8 +235,10 @@ func (r *PreInsertReviewer) planFromVerdict(item MemoryItem, v ingestVerdict) in // 第五态（design/23 §18.1）：行照常 active 落库（一个字不丢、可 revert）， // 只把 expires_at 写成「现在 + 临时 TTL」；**不投异步队列**（defer 已是 // 终局判断，再送语义审查只会把临时性抹掉）。 + // + // 到期时刻**不在这里算**：planIngest 用 tempDeadline 统一换算（唯一口径）， + // 因此本分支只负责「动作」与「留痕原因码」。 plan.Action = ingestPlanDefer - plan.ExpiresAt = r.deferDeadline(r.now().UTC()) plan.Trace = gateTraceDeferred default: // 真新 / 拿不准：正常 insert，并投既有异步队列（用语义召回再判一次）。 @@ -350,29 +372,14 @@ func (s *Store) finishIngest(ctx context.Context, item MemoryItem, plan ingestGa return } if plan.Action == ingestPlanDefer { - // 第五态（design/23 §18.1）：新行已经在 Add 里照常插好了（active、正文 - // 一字不差），这里只把 `expires_at` 补成计划给出的时刻 —— 与 - // absorb_new / absorb_old 是**同一范式**：INSERT 落行、finishIngest 落 - // 「那一行的派生列」。 + // 第五态（design/23 §18.1）：新行已经在 Add 里**一笔**落好 —— 档位与 + // `expires_at` 都随 INSERT 写下（plan.expiresAt() / plan.retention()）， + // 这里**不再有第二笔 UPDATE**。 // - // 为什么不是「INSERT 直接带上 expires_at」：store.go 的 INSERT 读的是 - // `expiryText(item.ExpiresAt)`（由**调用方**置位），而 defer 的判断发生在 - // 闸里（`planIngest` 拿的是 item 的**副本**），计划的值够不到那条 INSERT。 - // - // 🔴 **代价与短暂窗口（如实登记，不静默）**：这一态因此多出一笔按主键的 - // UPDATE（微秒级），且 INSERT 与这笔补写之间有一个**极短的窗口** —— - // 若进程恰在此刻中断，该行会以 `expires_at=''`（**永不过期**）留在库里。 - // 方向是刻意的：它正是 fail-safe 的「**宁可多留，不可误弃**」 - // （design/23 §18.3）—— 停在长期层只是多占一条记忆，而反向失败 - // （把该留的判成临时）会丢信息。写失败同样只记 WARN、**不让 Add 失败**。 - // - // 非 defer 的三态**一笔都不多写**：absorb_new / absorb_old 本来就有它们的 - // 第二笔（合并登记 / 置为被合并），insert 在函数开头就返回了。 - if err := s.setIngestExpiry(ctx, item.ID, plan.expiresAt()); err != nil { - s.log.Warn("memory ingest gate defer failed, item left long-lived", - "id", item.ID, "scope", string(item.Scope), "error", err) - return - } + // 这是本切片相对 M2 的一处结构性改进：那时 defer 的 `expires_at` 靠 + // INSERT 之后的补写，注释里如实登记过「进程恰在此刻中断 ⇒ 该行以 + // expires_at="" 永不过期留在库里」的短窗口。现在两列同笔写入，那个窗口 + // 不存在了；这里剩下的事情只有计数。 if r := s.ingestReviewer(); r != nil { r.bump(func(st *ReviewStats) { st.GateDeferred++ }) } diff --git a/internal/runtime/memory/preinsert_judge.go b/internal/runtime/memory/preinsert_judge.go index 1fd145f..93e5dab 100644 --- a/internal/runtime/memory/preinsert_judge.go +++ b/internal/runtime/memory/preinsert_judge.go @@ -16,10 +16,10 @@ package memory // `ingestVerdict.Gray`，交**异步审查层**的 T1 再问一次 // （preinsert_t1.go，design/23 §15.1 D3-2）。**本文件恒为零模型调用**。 // -// 另外在**候选比对之前**插一道**提取留存规则闸**（extract_retention.go， -// design/03 §10.5 的 N1~N5）：新条目自己就「显然不值得长期留存」时直接给第五态 -// `defer`（`reason=t0_low_value`），**不进候选比对、不进 T1**。它是纯规则、零模型 -// 调用，因此不改变上面那句「本文件恒为零模型调用」。 +// 另外在**候选比对之前**先看**已判定的档位**（`item.RetentionTier`，由 +// `Store.Add` 的 `resolveRetention` 落定，见 retention.go）：档位是 temp 时直接给 +// 第五态 `defer`（`reason=t0_low_value`），**不进候选比对、不进 T1**。 +// 本文件因此**不再跑任何留存规则**，也仍然恒为零模型调用。 // // 两条不可走样： // - **相似度只用来召回候选**，永不参与判定（design/23 §1.2「单阈值双向失效」）； @@ -63,6 +63,12 @@ type ingestVerdict struct { // judgeIngest 对一条新落库的记忆做四态判定。cands 已按召回分数降序 // （collectTidyPairs 的确定性排序），因此同分时的选择可复现。 // +// 🔴 **第五态 `defer` 不再在这里跑规则**（迁移 0026 / 切片 M37）：档位「留多久」 +// 由 `Store.Add` 的**唯一判定**（resolveRetention → retention.go）落成 +// `item.RetentionTier`，本函数只读这一位 —— `tier == temp` ⇔ `defer`。 +// 于是「判 defer」与「判 retention_tier=temp」是**同一个判定**，不存在两套规则 +// 各判一遍（M27 体检的缺口正是两处判据各说各话、且几乎都不触发）。 +// // 无论结论是什么，留痕都要带上**本轮召回来源**（§15 D3-8 的「依据」）： // 「没有命中任何包含关系」与「根本没召回到东西」是两件不同的事。 func judgeIngest(item MemoryItem, cands []ingestCandidate) ingestVerdict { @@ -73,13 +79,14 @@ func judgeIngest(item MemoryItem, cands []ingestCandidate) ingestVerdict { if v, ok := judgeExactDuplicate(item, cands); ok { return v } - // 提取留存规则闸（design/03 §10.5 的 N1~N5）：新条目**自己**就「显然不值得 - // 长期留存」时直接给第五态 `defer`，**不进候选比对、不进 T1** - // （design/23 §10.6：规则闸拦下的不进 T1，避免「模型推翻规则」）。 + // 档位已判临时（`retention_tier=temp`）⇒ 第五态 `defer`，**不进候选比对、 + // 不进 T1**（design/23 §10.6：规则闸拦下的不进 T1，避免「模型推翻规则」）。 // // 它排在**精确重复之后**是刻意的：已有等价条目时 `skip` 只更新既有行、 // **一行都不新增**，比 `defer`（新增一行临时记忆）更省也更干净。 - if rule := judgeExtractionRetention(item.Content); rule != retentionNone { + // 档位与 `defer` 因此不是同一件事：`defer` = 「关系判为真新 ∧ 档位是 temp」， + // 而档位本身对**每一行**都成立（被 skip 掉的那行也有档位，只是它不在册）。 + if item.RetentionTier == RetentionTemp { return ingestVerdict{ Decision: IngestDefer, Reason: retentionLowValueReason, Mode: mode, Veto: VetoNone, Gray: nil, diff --git a/internal/runtime/memory/preinsert_migration_test.go b/internal/runtime/memory/preinsert_migration_test.go index 4580d47..dcb2da7 100644 --- a/internal/runtime/memory/preinsert_migration_test.go +++ b/internal/runtime/memory/preinsert_migration_test.go @@ -162,9 +162,10 @@ func TestMigration0025CreatesAndRollsBackIngestReviewTable(t *testing.T) { } } - // 回滚：只回滚最高的一条（0025），表随之消失。 - if err := db.MigrateDown(ctx, d, 1, testLogger()); err != nil { - t.Fatalf("MigrateDown(1): %v", err) + // 回滚：本表之上还压着一条 0026（`memory_retention_tier`，切片 M37），因此这里 + // 回滚**两条**才轮到 0025 —— 迁移账本按 id 升序、down 按降序撤，这是既有语义。 + if err := db.MigrateDown(ctx, d, 2, testLogger()); err != nil { + t.Fatalf("MigrateDown(2): %v", err) } if tableExists(t, d, ingestReviewTable) { t.Fatalf("回滚后 %s 应已被删除", ingestReviewTable) diff --git a/internal/runtime/memory/preinsert_store.go b/internal/runtime/memory/preinsert_store.go index 45b8fac..cf5c50c 100644 --- a/internal/runtime/memory/preinsert_store.go +++ b/internal/runtime/memory/preinsert_store.go @@ -32,7 +32,7 @@ const ingestStatusMerged = "merged" // （design/23 §19），**不会复活**。姊妹注释在 tidy_store.go 的 listTidyPartitionItemsSQL。 const listReviewCandidatesSQL = `SELECT id, persona_id, role_id, project_id, scope, kind, content, tags, confidence, source_conversation, created_at, updated_at, - decay_weight FROM memory_items + decay_weight, retention_tier FROM memory_items WHERE persona_id = ? AND scope = ? AND project_id = ? AND status = ? AND id <> ? ORDER BY created_at DESC, rowid DESC LIMIT ?` @@ -46,18 +46,22 @@ const updateSurvivorSQL = `UPDATE memory_items SET merged_from = ?, merge_count = ?, confidence = ?, updated_at = ? WHERE id = ?` const revertMergedSQL = `UPDATE memory_items - SET status = ?, merged_into = '', updated_at = ? WHERE id = ? AND status = ?` + SET status = ?, merged_into = "", updated_at = ? WHERE id = ? AND status = ?` const selectMergedIntoSQL = `SELECT merged_into FROM memory_items WHERE id = ? AND status = ?` const selectItemStatusSQL = `SELECT status FROM memory_items WHERE id = ?` -// setIngestExpirySQL 是第五态（design/23 §18.1）的落点：**只写 expires_at 一列**。 -// 不碰 status / merged_* / content，因此它天然是「软动作」——回滚 = 把该列清空 -// （§18.4）；行一行不删（临时层的物理删除是 design/23 §19 的另一条有审计的 -// 清扫路径，不在这里）。 -const setIngestExpirySQL = `UPDATE memory_items SET expires_at = ? WHERE id = ?` +// setIngestRetentionSQL 是「留多久」轴在**异步审查层**的落点：把一条已落库记忆 +// 改成临时层（档位 + 有效期**同一条语句**写，两者因此不可能分叉），或按回滚口径 +// 改回长期层（档位 long + 有效期清空）。 +// +// 不碰 status / merged_* / content，因此它天然是「软动作」——回滚 = 把两列改回去 +// （design/23 §18.4）；行一行不删（临时层的物理删除是 design/23 §19 的另一条 +// 有审计的清扫路径，不在这里）。 +const setIngestRetentionSQL = `UPDATE memory_items + SET retention_tier = ?, expires_at = ? WHERE id = ?` const insertIngestReviewSQL = `INSERT INTO memory_ingest_reviews (id, item_id, persona_id, scope, project_id, decision, reason, veto, target_id, @@ -146,22 +150,37 @@ func (s *Store) listReviewCandidates(ctx context.Context, p tidyPartition, exclu return scanItems(rows, "review candidate") } -// setIngestExpiry 把一条记忆的 `expires_at` 写成 expiryText（**定宽 30 字节 UTC**）， -// 空白串表示「不设过期」。它是第五态 defer 的**唯一**写口，两处调用： +// setIngestRetention 把一条记忆改成临时层（`retention_tier='temp'` + +// `expires_at=expiryText(...)`）或改回长期层（`long` + 空串）。 // -// - 同步入库闸（preinsert_gate.go 的 finishIngest）：INSERT 之后补这一列； -// - 异步审查层（preinsert.go 的 applyVerdict）：T1 判 defer 时改这一列。 +// 它是**档位与有效期同一条语句**的落点：两列因此不可能分叉（「temp 却没有有效期」 +// 或「有有效期却标着长期」都写不出来）。三类调用： +// +// - 异步审查层（preinsert.go 的 applyVerdict）：T1 在同一次调用里判 defer 时， +// 把一条已经落库的长期记忆改判成临时层； +// - 回滚（design/23 §18.4）：把两列改回 `long` / `""` —— 清空有效期即升回长期； +// - 测试夹具需要「已经过期」的行时也走它（显式时刻由 expiryText 编码）。 +// +// 同步入库路径**不调它**：`Store.Add` 已经把两列同一笔 INSERT 写下（见 retention.go +// 的 tempDeadline），所以那条路径上没有第二笔 UPDATE、也没有中间窗口。 // // 🔴 **编码必须与读侧同源**：读侧时效闸是纯文本比较（timegate.go），因此这里 // 只许走 expiryText —— 写成别的布局（RFC3339 带 `+08:00`、秒精度、空格分隔） // 会让闸**静默**判错（早八小时的列值可能被判成「大于当下」，B 仓历史事故， // timegate_test.go 钉着那条反证）。 -func (s *Store) setIngestExpiry(ctx context.Context, id, expiresAt string) error { +func (s *Store) setIngestRetention(ctx context.Context, id string, tier RetentionTier, expiresAt string) error { if strings.TrimSpace(id) == "" { return ErrNotFound } - if _, err := s.db.ExecContext(ctx, setIngestExpirySQL, expiresAt, id); err != nil { - return fmt.Errorf("memory: set ingest expiry: %w", err) + if !tier.valid() { + return fmt.Errorf("%w: %q", ErrInvalidRetentionTier, tier) + } + if !tier.assigned() && expiresAt != "" { + // 未分型 + 非空有效期是自相矛盾的落点（会造出「没有档位却会到期」的行）。 + return fmt.Errorf("%w: unset tier with expires_at", ErrRetentionConflict) + } + if _, err := s.db.ExecContext(ctx, setIngestRetentionSQL, string(tier), expiresAt, id); err != nil { + return fmt.Errorf("memory: set ingest retention: %w", err) } return nil } diff --git a/internal/runtime/memory/retention.go b/internal/runtime/memory/retention.go new file mode 100644 index 0000000..5d3bb90 --- /dev/null +++ b/internal/runtime/memory/retention.go @@ -0,0 +1,187 @@ +package memory + +import ( + "fmt" + "strings" + "time" +) + +// 本文件是 `memory_items` 的第三根轴「留多久 / 干什么用」：档位闭集、**唯一**的 +// 判定入口、以及「档位 → 有效期」的**唯一**换算（迁移 0026、审计 +// `DevDocs/audit/记忆存量体检-2026-10-26.md` 卷四）。 +// +// 🔴 正交性（本片最容易写错的一条）：`kind` 是「关于什么」 +// （relationship / world_event），`scope` 是「谁的」（user / persona / project）， +// `retention_tier` 是「留多久 / 干什么用」。三根轴**互不推导**：一条关系记忆可以 +// 是临时，也可以是长期；一条世界事件也可以是知识库条目。任何「由 kind 推 tier」 +// （或反过来）的写法都是把两根轴压回一根，那正是本片要避免的设计回归。 +// +// 🔴 空串不是第六档，是「**尚未分型**」。它只出现在两类行上： +// +// ① 0026 迁移之前的**历史行**（列默认值为空串，本片刻意不做回填判定）； +// ② **受限来源**（群聊 / 陌生人）的新行 —— 见 resolveRetention：它们的留存由 +// 来源面决定，后台到期清扫也会跳过它们（expire.go 的 `nonOwnerSource` 谓词）， +// 判成 temp 就是造一条「对读面隐藏、却永远不会被删」的行。 +// +// 因此读侧**必须**能区分「未分型」与「已分型」：三条列表读路径缺省不过滤（历史行照常 +// 返回），显式传 `RetentionTierUnset` 时**只**返回未分型的行。 + +// RetentionTier 是「留多久 / 干什么用」轴的档位。零值 = 尚未分型。 +type RetentionTier string + +const ( + // RetentionTierUnset 是「尚未分型」：历史行与受限来源行。**它不是一档业务档位** + // —— 读侧绝不许把它当成 temp / long 之类的某一档（那会让「未分型」这个事实 + // 当场丢失，而它正是将来回填分型的抓手）。 + RetentionTierUnset RetentionTier = "" + + // RetentionTemp 是临时层：到期即从读面消失（读侧时效闸，timegate.go），随后被 + // 后台清扫物理删除（design/23 §19）。它是**唯一带时效**的档位， + // 且 `retention_tier='temp'` ⇔ `expires_at` 非空 —— 两者由同一个判定、同一处换算 + // 落库（见 resolveRetention / tempDeadline），**不许分叉**。 + RetentionTemp RetentionTier = "temp" + + // RetentionDaily 是日常层：一天之内的生活琐事 / 节奏。它与 `scope=persona` 的 + // 「角色日常」**不是一回事** —— 后者回答「谁的记忆」，这一档回答「留多久 / 干什么用」。 + RetentionDaily RetentionTier = "daily" + + // RetentionWork 是工作层：任务推进、工单、项目过程。 + RetentionWork RetentionTier = "work" + + // RetentionLong 是长期层：稳定的偏好 / 关系 / 事实。它也是入库判定的**默认落点** + // （判据没命中「显然不值得长期留」时，落到长期层 —— 后果只是多留一条，方向是 + // fail-safe 的「宁可多留，不可误弃」，design/23 §18.3）。 + RetentionLong RetentionTier = "long" + + // RetentionKnowledge 是知识库层：外部资料 / 规格 / 文档类条目。它与 + // `internal/runtime/knowledge` 的独立存储无关（那一层不归记忆管），这里只说 + // 「这条**记忆行**是知识库性质的」。 + RetentionKnowledge RetentionTier = "knowledge" +) + +// valid 报告 t 是否落在闭集内（**含**未分型的空串：它是合法的列值，不是一个错误）。 +func (t RetentionTier) valid() bool { + switch t { + case RetentionTierUnset, RetentionTemp, RetentionDaily, RetentionWork, RetentionLong, RetentionKnowledge: + return true + default: + return false + } +} + +// assigned 报告 t 是否**已分型**（即非空串）。它是「未分型」的唯一判据， +// 调用方不许用 `tier == RetentionLong` 之类的比较去替代它。 +func (t RetentionTier) assigned() bool { return t != RetentionTierUnset } + +// retentionVerdict 是「留多久」轴的判定结果：档位 + 命中的反例规则码。 +// +// 规则码只进日志、不进 `memory_ingest_reviews.reason`（那一列的闭集是 +// `{t0_low_value, t1_defer, …}`，见 extract_retention.go 的口径）—— 但它必须有 +// 一个落点，否则「n1~n7 到底谁命中过」只能靠猜（M27 体检就是这么发现闸形同虚设的）。 +type retentionVerdict struct { + Tier RetentionTier + Rule retentionRule +} + +// resolveRetention 决定一条待落库条目最终落哪一档 —— **全仓唯一的档位判定入口**。 +// +// 判定顺序（每一档都有不可走样的理由）： +// +// 1. **调用方显式给了档位** ⇒ 以调用方为准，规则/模型一律不许推翻它。这条对齐 +// AGENTS §9「判据靠元数据，不靠提示词」：写记忆的调用方（手动「记住」、 +// 知识库导入、整理侧回填）比一条词项规则更知道「这条该留多久」。 +// 显式非 temp 档 + 显式有效期是**自相矛盾**的输入（会造出一行「标着长期却会 +// 到期消失」的记忆），直接报错而不是替它挑一个。 +// 2. **显式给了有效期**（`item.ExpiresAt` 非零）⇒ 这一行会到期 ⇒ 它就是临时记忆。 +// 两者是同一根轴的两种表达，因此这里不是「推导」，而是同一个事实的补全。 +// 3. **受限来源**（群聊 / 陌生人）⇒ 不分型（空串）。见文件头注 ②。 +// 4. **留存规则命中**（design/03 §10.5 的 N1~N7）⇒ temp，并带回规则码。 +// 5. 其余 ⇒ long（默认落点，fail-safe）。 +// +// 🔴 「判第五态 `defer`」与「判 `retention_tier=temp`」**就是这一步**：闸与异步审查 +// 都只读 `item.RetentionTier`（preinsert_judge.go），不再各自跑一遍规则。M27/M37 的 +// 实测缺口正是「两处各判一遍、上游几乎不触发」，合并之后只有一个判据。 +func resolveRetention(item MemoryItem) (retentionVerdict, error) { + if tier := item.RetentionTier; tier.assigned() { + if tier != RetentionTemp && !item.ExpiresAt.IsZero() { + return retentionVerdict{}, fmt.Errorf( + "memory: retention_tier=%q with a non-empty expires_at: %w", tier, ErrRetentionConflict) + } + return retentionVerdict{Tier: tier}, nil + } + if !item.ExpiresAt.IsZero() { + return retentionVerdict{Tier: RetentionTemp}, nil + } + if nonOwnerSource(item.SourceConversation) { + return retentionVerdict{Tier: RetentionTierUnset}, nil + } + if rule := judgeExtractionRetention(item.Content); rule != retentionNone { + return retentionVerdict{Tier: RetentionTemp, Rule: rule}, nil + } + return retentionVerdict{Tier: RetentionLong}, nil +} + +// tempDeadline 给出落库时要写进 `expires_at` 的时刻，是「档位 → 有效期」的 +// **唯一**换算点：temp ⇒ 一个时刻，其余档位 ⇒ 零值（落库为空串 ⇒ 永不过期）。 +// +// 因此「temp ⇔ expires_at 非空」不是一条靠自觉维持的约定，而是这条换算的结构性结果； +// 任何新写路径只要拿这一行去 INSERT，就不可能写出「temp 却没有有效期」的行。 +// +// 时刻的三种来源（优先级从高到低）： +// +// 1. 调用方显式给的 `item.ExpiresAt`（回填 / 测试夹具需要「已经过期」的行）； +// 2. 接了入库审查服务时用**它的**可注入时钟与面板 TTL（`deferDeadline`）—— +// 闸、异步审查与写入侧因此用的是同一个算式； +// 3. 没接审查服务时用墙钟 + 代码默认 `tempTtlSec`。 +func (s *Store) tempDeadline(item MemoryItem) time.Time { + if item.RetentionTier != RetentionTemp { + return time.Time{} + } + if !item.ExpiresAt.IsZero() { + return item.ExpiresAt + } + if r := s.ingestReviewer(); r != nil { + return r.deferDeadline(r.now().UTC()) + } + return time.Now().UTC().Add(time.Duration(tempTtlSec) * time.Second) +} + +// retentionTierClause 把可选的档位过滤渲染成参数化 WHERE 片段（SQL 全参数化）。 +// +// 缺省（未传任何档位）= **不过滤**：既有调用方零改动，历史行（空串）与其他档位一样 +// 照常返回。传了档位就只返回这些档位；显式传 `RetentionTierUnset`（空串）是 +// 「只看未分型的行」—— 这是回填分型的入口，必须与「不过滤」可区分。 +func retentionTierClause(tiers []RetentionTier) (string, []any, error) { + if len(tiers) == 0 { + return "", nil, nil + } + seen := make(map[RetentionTier]bool, len(tiers)) + args := make([]any, 0, len(tiers)) + marks := make([]string, 0, len(tiers)) + for _, tier := range tiers { + if !tier.valid() { + return "", nil, fmt.Errorf("%w: %q", ErrInvalidRetentionTier, tier) + } + if seen[tier] { + continue + } + seen[tier] = true + marks = append(marks, "?") + args = append(args, string(tier)) + } + return " AND retention_tier IN (" + strings.Join(marks, ", ") + ")", args, nil +} + +// retentionTierList 是三条列表读路径共用的拼装：把可选的档位过滤接到既有 WHERE +// 之后、时效闸与 ORDER BY 之前，并按**子句出现顺序**排出参数。 +// +// 参数顺序是纯文本比较之外的唯一坑：SQLite 的 `?` 按出现位置绑定，档位片段的 +// 占位符夹在 status 与时效闸之间，因此档位参数也必须夹在那两组之间 —— 拼错不会 +// 报错，只会**静默按错误的档位过滤**（timegate 的历史事故同型）。 +func retentionTierList(where string, args []any, tiers []RetentionTier) (string, []any, error) { + clause, tierArgs, err := retentionTierClause(tiers) + if err != nil { + return "", nil, err + } + return where + clause, append(args, tierArgs...), nil +} diff --git a/internal/runtime/memory/retention_judge_test.go b/internal/runtime/memory/retention_judge_test.go new file mode 100644 index 0000000..2297a46 --- /dev/null +++ b/internal/runtime/memory/retention_judge_test.go @@ -0,0 +1,180 @@ +package memory + +import ( + "context" + "testing" + "time" +) + +// 本文件用**固定夹具语料**（不是生产库）钉住「判 temp 的判定本身」与 +// 「`defer` 与 `retention_tier=temp` 是同一个判定」这两条。 +// +// 语料分两半，方向是刻意相反的（任务书要求的「正例判得到 / 反例不会被判」）： +// +// 正例 = design/03 §10.5 的 N1~N7 形态（寒暄、一次性指令、时效查询、指示性时点、 +// 进行中状态）⇒ 必须判 temp； +// 反例 = 长期事实 / 稳定偏好 + 约束 / 显式「记住」/ 习惯 ⇒ 必须**不**判 temp +// （判错的后果不可逆：7 天后从所有读路径消失，随后被物理删除）。 +// +// 变异方向：把 resolveRetention 的 temp 分支删掉、或把规则改宽到「反例也命中」， +// 这两组断言会分别变红 —— 它们不是「跑通就行」的烟雾测试。 + +// retentionFixture 是一条夹具：内容 + 期望落点 + 为什么。 +type retentionFixture struct { + content string + want RetentionTier + why string + // rule 非空时一并断言命中的反例规则码（只有正例才需要，反例恒为 retentionNone）。 + rule retentionRule +} + +// retentionFixtures 是正反例语料。正例的每一条都能在 design/03 §10.5（N1~N7）找到 +// 出处；反例的每一条都是「判成临时就会丢信息」的真实形态。 +func retentionFixtures() []retentionFixture { + return []retentionFixture{ + // ── 正例（必须判 temp）──────────────────────────────────────────── + {"谢谢啦", RetentionTemp, "N1 寒暄/致谢", retentionChatter}, + {"早上好呀", RetentionTemp, "N1 寒暄", retentionChatter}, + {"帮我订个 7 点的闹钟", RetentionTemp, "N3 一次性祈使（含数值也被祈使前缀先吃掉）", retentionNone}, + {"现在几点了", RetentionTemp, "N4 时效性查询", retentionTimelyData}, + {"用户昨天发来一卷验收纸条", RetentionTemp, "N6 指示性时点在句首", retentionTimepoint}, + {"主人有一件正在慢慢考虑、还未决定的事", RetentionTemp, "N7 进行中状态（生产语料原样取回）", retentionInProgress}, + // ── 反例（必须判 long）────────────────────────────────────────── + {fixtureHabitShort, RetentionLong, "习惯陈述（熬夜）是长期事实", retentionNone}, + {"记住：主人对花生过敏，点餐要避开", RetentionLong, "显式「记住」是指令，闸不许改判", retentionNone}, + {"主人每天只喂小茂密一次猫粮，一次 2 勺", RetentionLong, "偏好 + 具体约束（数值/频次）", retentionNone}, + {"主人被称作「叶酱」，不许用「喵喵七号」", RetentionLong, "称呼是硬性长期事实", retentionNone}, + {"主人是上班族，工作日中午在食堂吃饭", RetentionLong, "职业/作息是长期事实", retentionNone}, + } +} + +// TestResolveRetentionFixtureCorpus 逐条钉住档位判定（正例判得到 / 反例不会被判）。 +func TestResolveRetentionFixtureCorpus(t *testing.T) { + t.Parallel() + for _, tc := range retentionFixtures() { + tc := tc + t.Run(tc.content, func(t *testing.T) { + t.Parallel() + item := habitItem(tc.content, tidyTestTime(t), 0.9) + verdict, err := resolveRetention(item) + if err != nil { + t.Fatalf("resolveRetention: %v", err) + } + if verdict.Tier != tc.want { + t.Fatalf("%q 判成 %q, want %q（%s）", tc.content, verdict.Tier, tc.want, tc.why) + } + if tc.rule != retentionNone && verdict.Rule != tc.rule { + t.Fatalf("%q 命中的规则 = %q, want %q（%s）", tc.content, verdict.Rule, tc.rule, tc.why) + } + }) + } +} + +// TestResolveRetentionWritePathMatchesFixture 把同一份语料压过**真实写入路径** +// （Store.Add），断言落库的两列与判定一致 —— 判据与落点之间不许有第二套逻辑。 +func TestResolveRetentionWritePathMatchesFixture(t *testing.T) { + t.Parallel() + store := newTestStore(t) + for _, tc := range retentionFixtures() { + tc := tc + t.Run(tc.content, func(t *testing.T) { + got := seedTidyItem(t, store, gateFixtureItem(tc.content, time.Time{})) + tier, expires := readRetentionColumns(t, store, got.ID) + if tier != tc.want { + t.Fatalf("%q 落库档位 = %q, want %q", tc.content, tier, tc.want) + } + if tc.want == RetentionTemp && expires == "" { + t.Fatalf("%q 判成 temp 却没写 expires_at（两列分叉）", tc.content) + } + if tc.want != RetentionTemp && expires != "" { + t.Fatalf("%q 判成 %q 却写了 expires_at = %q", tc.content, tc.want, expires) + } + }) + } +} + +// TestDeferDecisionIsTheTempTier 钉住「`defer` 与 `retention_tier=temp` 是同一个判定」 +// 在**闸开着**时成立：temp 档的条目拿到第五态 `defer` + `t0_low_value`，长期档的 +// 条目走既有四态（create），且留痕的规则版本是 d3.4。 +func TestDeferDecisionIsTheTempTier(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + ttl := time.Hour + r := newGateReviewer(t, store, ReviewConfig{TempTTL: ttl}, nil) + store.AttachIngestReviewer(r) + + tempRow := seedTidyItem(t, store, gateFixtureItem("谢谢啦", base)) + rev := gateRecordFor(t, store, tempRow.ID) + if rev.Decision != IngestDefer || rev.Reason != retentionLowValueReason { + t.Fatalf("temp 档留痕 = %+v, want defer + %q", rev, retentionLowValueReason) + } + if rev.RuleVer != IngestRuleVersion { + t.Fatalf("留痕规则版本 = %q, want %q", rev.RuleVer, IngestRuleVersion) + } + tier, expires := readRetentionColumns(t, store, tempRow.ID) + if tier != RetentionTemp { + t.Fatalf("defer 行的档位 = %q, want %q", tier, RetentionTemp) + } + // 到期时刻来自**同一个判定**与**同一个时钟**（闸注入的 now + 面板 TTL）。 + if want := expiryText(base.Add(ttl)); expires != want { + t.Fatalf("defer 行 expires_at = %q, want %q（判定时刻 + TTL）", expires, want) + } + + longRow := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base.Add(time.Minute))) + if rev := gateRecordFor(t, store, longRow.ID); rev.Decision == IngestDefer { + t.Fatalf("长期事实被判成 defer：%+v", rev) + } +} + +// TestTempTierDoesNotDependOnTheGate 钉住**档位判定不依赖同步闸**：闸没接（或关着） +// 时，档位与有效期照样在同一笔 INSERT 落下。 +// +// M27 体检的缺口之一是「档位/时效只在闸的分支里换算」，于是一旦闸关闭， +// `expires_at` 就再也没有生产者。本用例是那条缺口的回归钉子。 +func TestTempTierDoesNotDependOnTheGate(t *testing.T) { + t.Parallel() + store := newTestStore(t) // 未接审查服务 ⇒ planIngest 走 passthrough + got := seedTidyItem(t, store, gateFixtureItem("谢谢啦", time.Time{})) + + tier, expires := readRetentionColumns(t, store, got.ID) + if tier != RetentionTemp { + t.Fatalf("闸未接时档位 = %q, want %q", tier, RetentionTemp) + } + if expires == "" { + t.Fatal("闸未接时没写 expires_at —— 档位判定又退回到「只有闸才算」了") + } + // 这条行不会被投进任何队列、也不会有留痕：档位是写入侧的事实，不是审查的产物。 + if got.RetentionTier != RetentionTemp { + t.Fatalf("Add 返回的档位 = %q, want %q", got.RetentionTier, RetentionTemp) + } +} + +// TestAsyncReviewDoesNotMoveAnAlreadyTempRow 钉住异步审查层的**不漂移**： +// 一条已经由 Add 判成 temp 的行被审查再判一次时，`expires_at` 不会被 +// 「审查时刻 + TTL」重新写一遍（那会让同一条记忆的有效期随审查延迟漂移）。 +func TestAsyncReviewDoesNotMoveAnAlreadyTempRow(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{TempTTL: time.Hour, DisableIngestGate: true}, nil) + store.AttachIngestReviewer(r) + + got := seedTidyItem(t, store, gateFixtureItem("谢谢啦", base)) + _, before := readRetentionColumns(t, store, got.ID) + if before == "" { + t.Fatal("Add 没写 expires_at（档位判定没生效）") + } + + rev, err := r.Review(ctx, got) + if err != nil { + t.Fatalf("Review: %v", err) + } + if rev.Decision != IngestDefer { + t.Fatalf("异步层结论 = %q, want %q（档位是 temp ⇒ defer）", rev.Decision, IngestDefer) + } + if _, after := readRetentionColumns(t, store, got.ID); after != before { + t.Fatalf("异步审查把 expires_at 从 %q 改成了 %q —— 同一条记忆的有效期不该漂移", before, after) + } +} diff --git a/internal/runtime/memory/retention_migration_test.go b/internal/runtime/memory/retention_migration_test.go new file mode 100644 index 0000000..3d71531 --- /dev/null +++ b/internal/runtime/memory/retention_migration_test.go @@ -0,0 +1,241 @@ +package memory + +import ( + "context" + "database/sql" + "slices" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" +) + +// 本文件钉住**迁移 0026（`memory_retention_tier`）与 Store 自举 DDL 的同形与可回滚**， +// 手法照 preinsert_migration_test.go 对 0025 的那一套（同一批 helpers）。 +// +// 真实启动顺序是「先 MigrateUp、后 NewStore」（cmd/cyrene/wire.go），因此新列有两份 +// 真源：迁移（权威）与 NewStore 的幂等前滚（任意 *sql.DB 上的自举，retention_schema.go）。 +// 少一列就是「新库有、老库无」的漂移，因此两条判据都必须有： +// +// 1. TestMigration0026AddsColumnOverExistingRows —— 迁移落在**已有数据**上： +// 行不丢、列补上、默认值是空串（未分型）；down 真的把列删掉且不改任何一行的 +// 其它数据；再 up 能重建。 +// 2. TestMigration0026ShapeMatchesStoreBootstrap —— 两份 DDL **逐列逐索引同形** +// （它红的时候，正确的修法是让两份对齐，而不是放宽断言）。 + +// retentionTierColumn 是被测列名（迁移与自举 DDL 里的字面量）。 +const retentionTierColumn = "retention_tier" + +// retentionTierIndex 是被测索引名。 +const retentionTierIndex = "idx_memory_items_retention_tier" + +// retentionMigrationID 是 0026 在 schema_migrations 里的记账 id（= 文件名去掉 .up.sql）。 +const retentionMigrationID = "0026_memory_retention_tier" + +// hasMemoryColumn 报告 memory_items 上是否存在某列。 +func hasMemoryColumn(t *testing.T, d *sql.DB, name string) bool { + t.Helper() + return slices.ContainsFunc(tableColumns(t, d, "memory_items"), + func(c string) bool { return strings.Contains(c, "|"+name+"|") }) +} + +// hasMemoryIndex 报告 memory_items 上是否存在某条索引。 +func hasMemoryIndex(t *testing.T, d *sql.DB, name string) bool { + t.Helper() + return slices.ContainsFunc(tableIndexes(t, d, "memory_items"), + func(i string) bool { return strings.HasPrefix(i, name+"|") }) +} + +// insertLegacyMemoryRow 以**0026 之前的形态**写一行（不列 retention_tier）。 +func insertLegacyMemoryRow(t *testing.T, d *sql.DB, id, content string) { + t.Helper() + if _, err := d.ExecContext(context.Background(), `INSERT INTO memory_items + (id, persona_id, scope, kind, content, created_at, updated_at) + VALUES (?, 'persona.a', 'persona', 'relationship', ?, `+legacyTimeLiteral+`, `+legacyTimeLiteral+`)`, + id, content); err != nil { + t.Fatalf("insert legacy row %s: %v", id, err) + } +} + +// legacyTimeLiteral 是定宽 UTC 文本（与 formatTime 同形，读侧纯文本比较依赖它）。 +const legacyTimeLiteral = "'2026-10-01T10:00:00.000000000Z'" + +// TestMigration0026AddsColumnOverExistingRows 钉住迁移本体的三条： +// 落在既有数据上不丢行、默认值是空串（未分型）、down 真的删列且不动其它数据。 +func TestMigration0026AddsColumnOverExistingRows(t *testing.T) { + t.Parallel() + ctx := context.Background() + d := openRawMemoryDB(t) + migrateUp(t, d) + + // 造出「既有部署」：把 0026 撤掉（列与索引消失），再按**旧形态**写入两行。 + if err := db.MigrateDown(ctx, d, 1, testLogger()); err != nil { + t.Fatalf("MigrateDown(1): %v", err) + } + if hasMemoryColumn(t, d, retentionTierColumn) || hasMemoryIndex(t, d, retentionTierIndex) { + t.Fatalf("0026 回滚后 %s / %s 应已消失", retentionTierColumn, retentionTierIndex) + } + insertLegacyMemoryRow(t, d, "legacy-1", "一条 0026 之前就存在的记忆") + insertLegacyMemoryRow(t, d, "legacy-2", "另一条 0026 之前就存在的记忆") + + // 升级：0026 落在**已有数据**上。 + migrateUp(t, d) + if !hasMemoryColumn(t, d, retentionTierColumn) { + t.Fatalf("迁移后 memory_items 应带 %s 列", retentionTierColumn) + } + if !hasMemoryIndex(t, d, retentionTierIndex) { + t.Fatalf("迁移后应存在索引 %s", retentionTierIndex) + } + var rows int + if err := d.QueryRowContext(ctx, `SELECT COUNT(*) FROM memory_items`).Scan(&rows); err != nil { + t.Fatalf("count rows: %v", err) + } + if rows != 2 { + t.Fatalf("迁移后行数 = %d, want 2（既有行一行都不能丢）", rows) + } + // 默认值必须是**未分型**（空串），不是某一档 —— 本迁移刻意不做回填。 + for _, id := range []string{"legacy-1", "legacy-2"} { + var tier, content string + if err := d.QueryRowContext(ctx, + `SELECT retention_tier, content FROM memory_items WHERE id = ?`, id). + Scan(&tier, &content); err != nil { + t.Fatalf("read %s: %v", id, err) + } + if tier != "" { + t.Fatalf("既有行 %s 的档位 = %q, want 空串（历史行必须是「未分型」）", id, tier) + } + if content == "" { + t.Fatalf("既有行 %s 的正文丢了", id) + } + } + + // 回滚：只撤结构，不改写任何一行的其它数据。 + if err := db.MigrateDown(ctx, d, 1, testLogger()); err != nil { + t.Fatalf("MigrateDown(1, 二次): %v", err) + } + if hasMemoryColumn(t, d, retentionTierColumn) || hasMemoryIndex(t, d, retentionTierIndex) { + t.Fatalf("二次回滚后 %s / %s 应已消失", retentionTierColumn, retentionTierIndex) + } + if err := d.QueryRowContext(ctx, `SELECT COUNT(*) FROM memory_items`).Scan(&rows); err != nil { + t.Fatalf("count rows after down: %v", err) + } + if rows != 2 { + t.Fatalf("回滚后行数 = %d, want 2（down 只撤结构，不删行）", rows) + } + + // 再 up：列以默认值重建，历史行回到「未分型」。 + migrateUp(t, d) + if !hasMemoryColumn(t, d, retentionTierColumn) { + t.Fatalf("再次迁移后应重建 %s 列", retentionTierColumn) + } + var withTier int + if err := d.QueryRowContext(ctx, + `SELECT COUNT(*) FROM memory_items WHERE retention_tier = ""`).Scan(&withTier); err != nil { + t.Fatalf("count untyped: %v", err) + } + if withTier != 2 { + t.Fatalf("重建后未分型行 = %d, want 2", withTier) + } +} + +// TestMigration0026ShapeMatchesStoreBootstrap 是**漂移护栏**：迁移建出的 +// memory_items（列 + 索引）与自举 DDL 建出的必须逐项同形。 +// +// ⚠️ 比对的是**列集合**而不是列序，理由是既有的既成事实：`project_id` 是自举 +// 前滚补上的（ensureProjectIDColumn），迁移路径上它同样会被追加到末尾，因此两条 +// 路径的列序本来就不同。本片新增的 `retention_tier` 在两条路径上都追加在**当时**的 +// 末尾，不制造新的漂移 —— 集合一致 + 索引一致就是这条护栏的判据。 +func TestMigration0026ShapeMatchesStoreBootstrap(t *testing.T) { + t.Parallel() + ctx := context.Background() + + // 生产顺序：先迁移、后自举。两边都走完这一整套再比。 + migrated := openRawMemoryDB(t) + migrateUp(t, migrated) + if _, err := NewStore(ctx, migrated, testLogger()); err != nil { + t.Fatalf("NewStore(migrated): %v", err) + } + + bootstrapped := openRawMemoryDB(t) + if _, err := NewStore(ctx, bootstrapped, testLogger()); err != nil { + t.Fatalf("NewStore: %v", err) + } + + gotCols := sortedColumns(tableColumns(t, migrated, "memory_items")) + wantCols := sortedColumns(tableColumns(t, bootstrapped, "memory_items")) + if !slices.Equal(gotCols, wantCols) { + t.Fatalf("迁移与自举的 memory_items 列集合不一致（两份 DDL 必须同形）：\n迁移=%v\n自举=%v", + gotCols, wantCols) + } + gotIdx := tableIndexes(t, migrated, "memory_items") + wantIdx := tableIndexes(t, bootstrapped, "memory_items") + if !slices.Equal(gotIdx, wantIdx) { + t.Fatalf("迁移与自举的 memory_items 索引不一致：\n迁移=%v\n自举=%v", gotIdx, wantIdx) + } + // 新列在两份 DDL 里的形状必须逐字相同（TEXT NOT NULL DEFAULT + 空串字面量）。 + for name, cols := range map[string][]string{"迁移": gotCols, "自举": wantCols} { + if !slices.Contains(cols, retentionTierColumn+"|TEXT|1|''|0") { + t.Fatalf("%s 侧的 %s 列形状不是 TEXT NOT NULL DEFAULT 空串：%v", name, retentionTierColumn, cols) + } + } + // 自举 DDL 里新列写在最后一列（新建库的期望形状）——迁移路径上它后面可能还有 + // 自举追加的 project_id，因此只对自举侧断言顺序。 + bootCols := tableColumns(t, bootstrapped, "memory_items") + if last := bootCols[len(bootCols)-1]; !strings.Contains(last, "|"+retentionTierColumn+"|") { + t.Fatalf("自举建的 memory_items 最后一列 = %q, want %s", last, retentionTierColumn) + } +} + +// sortedColumns 把 `cid|name|type|notnull|default|pk` 渲染出的行去掉 cid 后排序， +// 使比对与列序无关（理由见 TestMigration0026ShapeMatchesStoreBootstrap）。 +func sortedColumns(rows []string) []string { + out := make([]string, 0, len(rows)) + for _, r := range rows { + if i := strings.Index(r, "|"); i >= 0 { + r = r[i+1:] + } + out = append(out, r) + } + slices.Sort(out) + return out +} + +// TestMigration0026ThenStoreBootstrapIsARegularPath 覆盖**真实启动顺序** +// （cmd/cyrene/wire.go：先 MigrateUp、后 NewStore）：0026 先建好列，随后 Store 的 +// 自举前滚走到 "duplicate column name" —— 那是**预期路径**，必须无害；重复建 Store +// 同样幂等，且写入路径可用。 +func TestMigration0026ThenStoreBootstrapIsARegularPath(t *testing.T) { + t.Parallel() + ctx := context.Background() + d := openRawMemoryDB(t) + migrateUp(t, d) + if !hasMemoryColumn(t, d, retentionTierColumn) { + t.Fatalf("迁移后应有 %s 列", retentionTierColumn) + } + + store, err := NewStore(ctx, d, testLogger()) + if err != nil { + t.Fatalf("NewStore（列已由迁移建好）: %v", err) + } + after := tableColumns(t, d, "memory_items") + // 列必须还在（自举的 duplicate column 路径不许把它当成失败）。 + if !hasMemoryColumn(t, d, retentionTierColumn) { + t.Fatalf("自举之后 %s 列消失了", retentionTierColumn) + } + if !hasMemoryIndex(t, d, retentionTierIndex) { + t.Fatalf("自举之后应存在索引 %s", retentionTierIndex) + } + // 幂等：再建一次 Store 不报错、不改形状。 + if _, err := NewStore(ctx, d, testLogger()); err != nil { + t.Fatalf("NewStore(第二次): %v", err) + } + if got := tableColumns(t, d, "memory_items"); !slices.Equal(got, after) { + t.Fatalf("第二次自举改变了表结构：\n got=%v\nwant=%v", got, after) + } + // 写入路径可用（档位两列在同一笔 INSERT 落下）。 + got := seedTidyItem(t, store, gateFixtureItem("谢谢啦", time.Time{})) + if tier, _ := readRetentionColumns(t, store, got.ID); tier != RetentionTemp { + t.Fatalf("迁移建表后经 Store 落库的档位 = %q, want %q", tier, RetentionTemp) + } +} diff --git a/internal/runtime/memory/retention_schema.go b/internal/runtime/memory/retention_schema.go new file mode 100644 index 0000000..6eefa29 --- /dev/null +++ b/internal/runtime/memory/retention_schema.go @@ -0,0 +1,55 @@ +package memory + +import ( + "context" + "database/sql" + "fmt" + "strings" +) + +// 本文件是「留多久 / 干什么用」轴（迁移 0026）的 schema 前滚：给**既有**的 +// memory_items 补 `retention_tier` 列，再建读侧索引。 +// +// 为什么需要在这里前滚：`createMemoryItemsTable` 是 `CREATE TABLE IF NOT EXISTS`， +// 它只对**新库**生效 —— 既有部署的表早就建好了，新列不会自己长出来。这与 +// `ensureProjectIDColumn` / `ensureTidyMemoryColumns` 是同一手法、同一原因 +// （记忆存储必须自包含、可在任意 *sql.DB 上使用）。 +// +// 🔴 与迁移的分工（两份真源，必须同形）：权威机制是 +// `internal/db/migrations/0026_memory_retention_tier`；这里是「任意 *sql.DB」的 +// 前滚。启动顺序是「先 MigrateUp、后 NewStore」（cmd/cyrene/wire.go），因此生产 +// 库里 0026 先落地、这里走到 `duplicate column name` 是**预期路径**；反过来 +// （先自举、后迁移）也成立，因为迁移的 ALTER 同样只跑一次。两份 DDL 的列/索引 +// 逐项同形由 retention_migration_test.go 钉住。 +// +// 顺序**必须**是「建表 → 前滚列 → 建索引」：索引依赖的列在既有库上还不存在时， +// 先建索引会报 "no such column: retention_tier"（tidy_schema.go 记过同型的教训）。 + +// ensureRetentionSchema 幂等地给 memory_items 补上档位列与读侧索引。 +func ensureRetentionSchema(ctx context.Context, db *sql.DB) error { + if err := ensureRetentionTierColumn(ctx, db); err != nil { + return err + } + if _, err := db.ExecContext(ctx, createMemoryItemsRetentionTierIndex); err != nil { + return fmt.Errorf("memory: retention tier index: %w", err) + } + return nil +} + +// ensureRetentionTierColumn 在列缺失时补加 `retention_tier`，已存在时忽略 +// SQLite 的 "duplicate column name"（那是预期路径，不是错误）。 +// +// 默认值 = 空串 = 尚未分型：既有行因此**行为逐字不变**（不设过期、读路径照常可见）， +// 本片也刻意**不**在启动时回填它们的档位 —— 回填是一次写全表的动作，该由整理侧 +// 显式做（「按空串过滤」正是那个入口），不该藏在组合根的启动路径里。 +func ensureRetentionTierColumn(ctx context.Context, db *sql.DB) error { + _, err := db.ExecContext(ctx, `ALTER TABLE memory_items + ADD COLUMN retention_tier TEXT NOT NULL DEFAULT ''`) + if err == nil { + return nil + } + if strings.Contains(err.Error(), "duplicate column name") { + return nil + } + return fmt.Errorf("memory: roll forward retention_tier column: %w", err) +} diff --git a/internal/runtime/memory/retention_test.go b/internal/runtime/memory/retention_test.go new file mode 100644 index 0000000..a9fb8a3 --- /dev/null +++ b/internal/runtime/memory/retention_test.go @@ -0,0 +1,447 @@ +package memory + +import ( + "context" + "os" + "strings" + "testing" + "time" +) + +// 本文件是「留多久 / 干什么用」轴（迁移 0026、retention.go）的验收用例集。 +// 它逐条钉住任务书的验收点： +// +// 1. 闭集：temp / daily / work / long / knowledge 五个档位 + 空串「尚未分型」； +// 空串**不是**某一档，读侧能把两者分开（不过滤时原样返回，按空串过滤只取未分型）； +// 2. **正交性**：`kind`（关于什么）与 `retention_tier`（留多久）是两列两根轴， +// relationship+temp / relationship+long / world_event+temp / world_event+knowledge +// 四种组合都能存能读，互不推导； +// 3. **不变量**：`tier=temp` ⇔ `expires_at` 非空且落在 TTL 窗口内；非 temp 恒为空串； +// 4. **一个判定**：`defer` 的判据就是「档位是 temp」，且档位与有效期同一笔 INSERT +// 落下（不依赖同步闸是否开启）。 + +// readRetentionColumns 直读一行的档位与有效期。档位在 Store.Get 的读侧面上，有效期 +// **刻意只写不读**（store.go 的 ExpiresAt 注释），因此要验证落库形态只能直读。 +func readRetentionColumns(t *testing.T, store *Store, id string) (RetentionTier, string) { + t.Helper() + var tier, expires string + if err := store.db.QueryRowContext(context.Background(), + `SELECT retention_tier, expires_at FROM memory_items WHERE id = ?`, id). + Scan(&tier, &expires); err != nil { + t.Fatalf("read retention columns %s: %v", id, err) + } + return RetentionTier(tier), expires +} + +// seedHistoricalItem 造一条**0026 之前**的历史行：档位停在列默认值（空串 = 未分型）。 +// +// 它只能直写 SQL 造出来 —— `Store.Add` 总会判出一个档位，这正是「未分型只属于 +// 历史行与受限来源」这条口径的结构性证据（不是在测试里塞一个假值）。 +func seedHistoricalItem(t *testing.T, store *Store, personaID, content string) string { + t.Helper() + ctx := context.Background() + id, err := newID() + if err != nil { + t.Fatalf("newID: %v", err) + } + now := formatTime(time.Now().UTC()) + if _, err := store.db.ExecContext(ctx, `INSERT INTO memory_items + (id, persona_id, scope, kind, content, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + id, personaID, string(ScopePersona), string(KindRelationship), content, now, now); err != nil { + t.Fatalf("seed historical item: %v", err) + } + return id +} + +// TestRetentionTierClosedSet 钉住闭集本身：五个业务档位 + 未分型都是合法列值， +// 而 `assigned()` 只对非空串为真（这是「未分型」的唯一判据）。 +func TestRetentionTierClosedSet(t *testing.T) { + t.Parallel() + cases := []struct { + tier RetentionTier + valid bool + assigned bool + }{ + {RetentionTierUnset, true, false}, + {RetentionTemp, true, true}, + {RetentionDaily, true, true}, + {RetentionWork, true, true}, + {RetentionLong, true, true}, + {RetentionKnowledge, true, true}, + {"ephemeral", false, true}, + {"TEMP", false, true}, + {"long ", false, true}, + } + for _, tc := range cases { + if got := tc.tier.valid(); got != tc.valid { + t.Fatalf("RetentionTier(%q).valid() = %t, want %t", tc.tier, got, tc.valid) + } + if got := tc.tier.assigned(); got != tc.assigned { + t.Fatalf("RetentionTier(%q).assigned() = %t, want %t", tc.tier, got, tc.assigned) + } + } + // 闭集里的五个业务档位一个都不能少（少了哪一档，「两列」的语义就不完整）。 + for _, want := range []RetentionTier{RetentionTemp, RetentionDaily, RetentionWork, RetentionLong, RetentionKnowledge} { + if !want.valid() || !want.assigned() { + t.Fatalf("档位 %q 不在闭集里", want) + } + } +} + +// TestAddRejectsUnknownRetentionTier 钉住「闭集是硬约束」：非法档位在 Add 入口被拒， +// 不会带着脏值落库（否则读侧的过滤会把 `TEMP` 当成另一种 temp）。 +func TestAddRejectsUnknownRetentionTier(t *testing.T) { + t.Parallel() + store := newTestStore(t) + item := baseItem() + item.RetentionTier = RetentionTier("ephemeral") + if _, err := store.Add(context.Background(), item); err == nil { + t.Fatal("非法档位竟被接受 —— 闭集没有生效") + } +} + +// TestRetentionTierIsOrthogonalToKind 钉住任务书的正交性验收点：`kind`（关于什么） +// 与 `retention_tier`（留多久）是两根轴，四种组合都能存能读，且读回来逐字不变。 +// +// 变异方向：若有人写出「relationship ⇒ long」「world_event ⇒ temp」这类推导， +// 这四条里立刻有两条读回来的档位与写入时不同。 +func TestRetentionTierIsOrthogonalToKind(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + + cases := []struct { + name string + kind Kind + tier RetentionTier + content string + }{ + {"关系+临时", KindRelationship, RetentionTemp, "谢谢啦"}, + {"关系+长期", KindRelationship, RetentionLong, "主人习惯被称作「叶酱」"}, + {"世界事件+临时", KindWorldEvent, RetentionTemp, "早上好呀"}, + {"世界事件+知识库", KindWorldEvent, RetentionKnowledge, "迁移号约定：接着上一个号递增"}, + } + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + item := baseItem() + item.Kind = tc.kind + item.RetentionTier = tc.tier + item.Content = tc.content + got, err := store.Add(ctx, item) + if err != nil { + t.Fatalf("Add: %v", err) + } + if got.RetentionTier != tc.tier { + t.Fatalf("Add 返回的档位 = %q, want %q", got.RetentionTier, tc.tier) + } + loaded, err := store.Get(ctx, got.ID) + if err != nil { + t.Fatalf("Get: %v", err) + } + if loaded.Kind != tc.kind || loaded.RetentionTier != tc.tier { + t.Fatalf("读回 (kind=%q, tier=%q), want (%q, %q) —— 两列是两根独立的轴", + loaded.Kind, loaded.RetentionTier, tc.kind, tc.tier) + } + }) + } +} + +// TestTempTierImpliesExpiryWithinTTLWindow 钉住不变量的一半：**被判成 temp 的行， +// 有效期一定落在 TTL 窗口内**，且由同一笔 INSERT 写下（没有第二笔 UPDATE 的窗口）。 +// +// 这里没有接审查服务 ⇒ 走「墙钟 + 代码默认 tempTtlSec」那一档换算；接了审查服务的 +// 情况由 TestDeferDecisionIsTheTempTier 用固定时钟钉。 +func TestTempTierImpliesExpiryWithinTTLWindow(t *testing.T) { + t.Parallel() + store := newTestStore(t) + ttl := time.Duration(tempTtlSec) * time.Second + before := time.Now().UTC() + + got := seedTidyItem(t, store, gateFixtureItem("谢谢啦", time.Time{})) + after := time.Now().UTC() + + tier, expires := readRetentionColumns(t, store, got.ID) + if tier != RetentionTemp { + t.Fatalf("短寒暄的档位 = %q, want %q（留存规则没接上）", tier, RetentionTemp) + } + at, err := parseTime(expires) + if err != nil { + t.Fatalf("parse expires_at %q: %v", expires, err) + } + if !at.After(before) || at.After(after.Add(ttl)) { + t.Fatalf("expires_at = %v 不在 TTL 窗口 (%v, %v] 内", at, before, after.Add(ttl)) + } +} + +// TestNonTempTierHasNoExpiry 钉住不变量的另一半：**非 temp 档的行 expires_at 恒为空串** +// （因此四条列表读路径恒可见、后台清扫结构性跳过它们）。 +func TestNonTempTierHasNoExpiry(t *testing.T) { + t.Parallel() + store := newTestStore(t) + got := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, time.Time{})) + tier, expires := readRetentionColumns(t, store, got.ID) + if tier != RetentionLong { + t.Fatalf("长期事实的档位 = %q, want %q（默认落点）", tier, RetentionLong) + } + if expires != "" { + t.Fatalf("非 temp 档的 expires_at = %q, want 空串", expires) + } +} + +// TestExplicitTierIsNotOverriddenByRules 钉住 AGENTS §9 的取向：**调用方的显式档位 +// 是元数据，词项规则不许推翻它**。一条「短寒暄」按规则会被判 temp，但调用方明确说 +// 它是知识库条目 ⇒ 它是 knowledge 且没有有效期。 +func TestExplicitTierIsNotOverriddenByRules(t *testing.T) { + t.Parallel() + store := newTestStore(t) + item := gateFixtureItem("谢谢啦", time.Time{}) + item.RetentionTier = RetentionKnowledge + got := seedTidyItem(t, store, item) + + tier, expires := readRetentionColumns(t, store, got.ID) + if tier != RetentionKnowledge { + t.Fatalf("显式档位被规则推翻：tier = %q, want %q", tier, RetentionKnowledge) + } + if expires != "" { + t.Fatalf("knowledge 档却写下了 expires_at = %q", expires) + } +} + +// TestRetentionInputContradictions 钉住两个自相矛盾的输入都不许静默通过： +// +// ① 非 temp 档 + 非零有效期 ⇒ 报错（一行标着长期却会到期消失）； +// ② 未分型 + 非零有效期 ⇒ 判 temp（「会到期」就是临时记忆，这是同一个事实的补全）。 +func TestRetentionInputContradictions(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + + bad := gateFixtureItem(fixtureHabitShort, time.Time{}) + bad.RetentionTier = RetentionLong + bad.ExpiresAt = time.Now().UTC().Add(time.Hour) + if _, err := store.Add(ctx, bad); err == nil { + t.Fatal("「长期档 + 有效期」竟被接受 —— 两列会当场分叉") + } + + implied := gateFixtureItem(fixtureHabitShort, time.Time{}) + implied.ExpiresAt = time.Now().UTC().Add(time.Hour) + got, err := store.Add(ctx, implied) + if err != nil { + t.Fatalf("Add（只给有效期）: %v", err) + } + if tier, _ := readRetentionColumns(t, store, got.ID); tier != RetentionTemp { + t.Fatalf("只给有效期时档位 = %q, want %q（有到期时刻就是临时记忆）", tier, RetentionTemp) + } +} + +// TestRestrictedSourceStaysUntyped 钉住受限来源（群聊 / 陌生人）**不分型**： +// 判它 temp 会造出一条「读面隐藏、后台清扫又跳过（expire.go 的 nonOwnerSource）」 +// 的行 —— 那是一条永远不会被删的幽灵行。 +func TestRestrictedSourceStaysUntyped(t *testing.T) { + t.Parallel() + store := newTestStore(t) + item := gateFixtureItem("谢谢啦", time.Time{}) + item.SourceConversation = FormatSourceConversation("sess-channel-42", "msg-9") + got := seedTidyItem(t, store, item) + + tier, expires := readRetentionColumns(t, store, got.ID) + if tier != RetentionTierUnset { + t.Fatalf("受限来源的档位 = %q, want 未分型（空串）", tier) + } + if expires != "" { + t.Fatalf("受限来源被写了 expires_at = %q（会变成永不删除的隐藏行）", expires) + } +} + +// tierFixtureSet 是档位过滤用例的四条夹具行（四条覆盖四种档位状态）： +// long（长期事实，persona 作用域）、temp（临时，user 作用域）、knowledge +// （知识库，project 作用域）、以及一条**未分型的历史行**（只能直写 SQL 造）。 +type tierFixtureSet struct { + longID, tempID, projectID, historicalID string +} + +// seedTierFixtureSet 造出四条夹具行。 +func seedTierFixtureSet(t *testing.T, store *Store) tierFixtureSet { + t.Helper() + set := tierFixtureSet{} + set.longID = seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, time.Time{})).ID + tempItem := gateFixtureItem("谢谢啦", time.Time{}) + tempItem.Scope = ScopeUser + set.tempID = seedTidyItem(t, store, tempItem).ID + projectItem := gateFixtureItem("迁移号约定：接着上一个号递增", time.Time{}) + projectItem.Scope = ScopeProject + projectItem.ProjectID = "proj-1" + projectItem.RetentionTier = RetentionKnowledge + set.projectID = seedTidyItem(t, store, projectItem).ID + set.historicalID = seedHistoricalItem(t, store, "persona.a", "一条 0026 之前就存在的记忆") + return set +} + +// tierIDs 把条目列表折成 id 集合（断言用）。 +func tierIDs(items []MemoryItem) map[string]bool { + out := make(map[string]bool, len(items)) + for _, it := range items { + out[it.ID] = true + } + return out +} + +// TestListTiersFilter 钉住读侧的**可选**过滤（跨角色的 List / ListTiers）： +// +// - 不传档位 = 不过滤：**历史行（空串）照常返回**（既有调用方零改动）； +// - 传档位 = 只返回这些档位（可多档并集）； +// - 显式传空串 = 只返回**未分型**的行（回填分型的入口，必须与「不过滤」可区分）。 +func TestListTiersFilter(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + set := seedTierFixtureSet(t, store) + + // 不过滤：四类行都在（含未分型的历史行）。 + all, err := store.List(ctx) + if err != nil { + t.Fatalf("List: %v", err) + } + got := tierIDs(all) + for _, id := range []string{set.longID, set.tempID, set.projectID, set.historicalID} { + if !got[id] { + t.Fatalf("不过滤时漏掉了 %s（缺省必须不过滤，历史行照常返回）", id) + } + } + + // 只看 temp：只剩那条临时记忆。 + temps, err := store.ListTiers(ctx, RetentionTemp) + if err != nil { + t.Fatalf("ListTiers(temp): %v", err) + } + if got := tierIDs(temps); len(got) != 1 || !got[set.tempID] { + t.Fatalf("ListTiers(temp) = %v, want 只有 %s", got, set.tempID) + } + + // 只看未分型：只剩那条历史行（与「不过滤」必须不同）。 + untyped, err := store.ListTiers(ctx, RetentionTierUnset) + if err != nil { + t.Fatalf("ListTiers(unset): %v", err) + } + if got := tierIDs(untyped); len(got) != 1 || !got[set.historicalID] { + t.Fatalf("ListTiers('') = %v, want 只有 %s（空串必须能与不过滤区分）", got, set.historicalID) + } + + // 多档位并集。 + two, err := store.ListTiers(ctx, RetentionTemp, RetentionKnowledge) + if err != nil { + t.Fatalf("ListTiers(temp, knowledge): %v", err) + } + if got := tierIDs(two); len(got) != 2 || !got[set.tempID] || !got[set.projectID] { + t.Fatalf("ListTiers(temp, knowledge) = %v, want {%s, %s}", got, set.tempID, set.projectID) + } +} + +// TestListTiersFilterByScope 钉住同一口径在三条**带分区**的读路径上成立 +// （ListByPersonaTiers / ListScopeTiers / ListProjectTiers），并钉住非法档位 +// **必须报错**而不是静默退化成「不过滤」。 +func TestListTiersFilterByScope(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + set := seedTierFixtureSet(t, store) + + personaLong, err := store.ListByPersonaTiers(ctx, "persona.a", RetentionLong) + if err != nil { + t.Fatalf("ListByPersonaTiers: %v", err) + } + if got := tierIDs(personaLong); len(got) != 1 || !got[set.longID] { + t.Fatalf("ListByPersonaTiers(long) = %v, want 只有 %s", got, set.longID) + } + // 历史行也是 persona.a 的，但档位是空串 ⇒ 在 long 过滤里不该出现。 + personaUntyped, err := store.ListByPersonaTiers(ctx, "persona.a", RetentionTierUnset) + if err != nil { + t.Fatalf("ListByPersonaTiers(unset): %v", err) + } + if got := tierIDs(personaUntyped); len(got) != 1 || !got[set.historicalID] { + t.Fatalf("ListByPersonaTiers('') = %v, want 只有 %s", got, set.historicalID) + } + userTemp, err := store.ListScopeTiers(ctx, ScopeUser, "persona.a", RetentionTemp) + if err != nil { + t.Fatalf("ListScopeTiers: %v", err) + } + if got := tierIDs(userTemp); len(got) != 1 || !got[set.tempID] { + t.Fatalf("ListScopeTiers(user, temp) = %v, want 只有 %s", got, set.tempID) + } + projectKnown, err := store.ListProjectTiers(ctx, "proj-1", RetentionKnowledge) + if err != nil { + t.Fatalf("ListProjectTiers: %v", err) + } + if got := tierIDs(projectKnown); len(got) != 1 || !got[set.projectID] { + t.Fatalf("ListProjectTiers(knowledge) = %v, want 只有 %s", got, set.projectID) + } + + // 非法档位：报错，不许静默退化成「不过滤」。 + calls := []struct { + name string + run func() error + }{ + {"ListByPersonaTiers", func() error { + _, err := store.ListByPersonaTiers(ctx, "persona.a", "ephemeral") + return err + }}, + {"ListTiers", func() error { _, err := store.ListTiers(ctx, "ephemeral"); return err }}, + {"ListScopeTiers", func() error { + _, err := store.ListScopeTiers(ctx, ScopeUser, "persona.a", "ephemeral") + return err + }}, + {"ListProjectTiers", func() error { + _, err := store.ListProjectTiers(ctx, "proj-1", "ephemeral") + return err + }}, + } + for _, call := range calls { + if err := call.run(); err == nil { + t.Fatalf("%s 接受了非法档位（应当报错，而不是当「不过滤」）", call.name) + } + } +} + +// TestRetentionJudgementHasSingleRuleCallSite 是**防回归的源码级断言**：留存规则 +// （`judgeExtractionRetention`）在生产代码里只能有一个调用点 —— `resolveRetention`。 +// +// 它钉的是本切片的核心承诺「**不许两套规则各判一遍**」：M27 体检的缺口（1082 次审查 +// 只判出 1 条 defer、`expires_at` 全表为空）根因就是档位与 defer 各判一次、谁都不完整。 +// 有人把规则调用再塞回 `judgeIngest` / 闸 / 异步审查时，本用例会红。 +func TestRetentionJudgementHasSingleRuleCallSite(t *testing.T) { + t.Parallel() + files, err := os.ReadDir(".") + if err != nil { + t.Fatalf("read package dir: %v", err) + } + var sites []string + for _, f := range files { + name := f.Name() + if f.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") { + continue + } + src, err := os.ReadFile(name) + if err != nil { + t.Fatalf("read %s: %v", name, err) + } + for _, line := range strings.Split(string(src), "\n") { + trimmed := strings.TrimSpace(line) + if !strings.Contains(trimmed, "judgeExtractionRetention(") { + continue + } + if strings.HasPrefix(trimmed, "//") || strings.HasPrefix(trimmed, "func ") { + continue + } + sites = append(sites, name+": "+trimmed) + } + } + if len(sites) != 1 { + t.Fatalf("judgeExtractionRetention 的调用点 = %d 处，want 恰为 1（resolveRetention）：\n%s", + len(sites), strings.Join(sites, "\n")) + } + if !strings.HasPrefix(sites[0], "retention.go:") { + t.Fatalf("唯一调用点应在 retention.go 的 resolveRetention 里，实得 %s", sites[0]) + } +} diff --git a/internal/runtime/memory/rules.go b/internal/runtime/memory/rules.go index 3669cbb..ed85a2a 100644 --- a/internal/runtime/memory/rules.go +++ b/internal/runtime/memory/rules.go @@ -78,7 +78,21 @@ func (s *Store) AddScoped(ctx context.Context, actorID string, item MemoryItem) // （expires_at / valid_to 均未到期）的行 —— 闸只有这里与 ListByPersona / // ListProject / List 四处，检索路径经 Retriever.visibleCandidates 组合调用 // 本方法，因此自动获得同一判据（timegate.go 头注、design/03 §11.2）。 +// +// `tiers` 的档位过滤见同族的 ListScopeTiers（签名不变的理由见 +// store_list.go 的 ListByPersona 注释：网关接口钉着这批精确签名）。 func (s *Store) ListScope(ctx context.Context, scope Scope, personaID string) ([]MemoryItem, error) { + return s.listScope(ctx, scope, personaID, nil) +} + +// ListScopeTiers 是 ListScope 的**可选档位过滤**版本（空 tiers = 不过滤； +// `RetentionTierUnset` = 只看尚未分型的行）。 +func (s *Store) ListScopeTiers(ctx context.Context, scope Scope, personaID string, tiers ...RetentionTier) ([]MemoryItem, error) { + return s.listScope(ctx, scope, personaID, tiers) +} + +// listScope 是 ListScope / ListScopeTiers 的共同实现（tiers 为空不过滤）。 +func (s *Store) listScope(ctx context.Context, scope Scope, personaID string, tiers []RetentionTier) ([]MemoryItem, error) { if strings.TrimSpace(personaID) == "" { return nil, ErrMissingPersona } @@ -86,11 +100,16 @@ func (s *Store) ListScope(ctx context.Context, scope Scope, personaID string) ([ return nil, ErrInvalidScope } now := timeGateNow() - rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind, + query := `SELECT id, persona_id, role_id, project_id, scope, kind, content, tags, confidence, source_conversation, created_at, updated_at, - decay_weight FROM memory_items WHERE scope = ? AND status = ?`+timeGateClause+` - ORDER BY created_at DESC, rowid DESC`, - string(scope), string(tidyStatusActive), now, now) + decay_weight, retention_tier FROM memory_items WHERE scope = ? AND status = ?` + args := []any{string(scope), string(tidyStatusActive)} + query, args, err := retentionTierList(query, args, tiers) + if err != nil { + return nil, err + } + rows, err := s.db.QueryContext(ctx, query+timeGateClause+` + ORDER BY created_at DESC, rowid DESC`, append(args, now, now)...) if err != nil { return nil, fmt.Errorf("memory: list scope items: %w", err) } @@ -105,15 +124,34 @@ func (s *Store) ListScope(ctx context.Context, scope Scope, personaID string) ([ // 记忆所有角色可读，因此不施加角色过滤。projectID 为空 // 返回 ErrMissingProjectID。与其余三条列表读路径同口径：只返回在册且过 // **时效闸**的行（timegate.go、design/03 §9.6/§11.2）。 +// +// `tiers` 的档位过滤见同族的 ListProjectTiers —— 本方法虽然是深检索入口，但 +// 「留多久」这根轴对项目记忆同样成立，因此一并支持。 func (s *Store) ListProject(ctx context.Context, projectID string) ([]MemoryItem, error) { + return s.listProject(ctx, projectID, nil) +} + +// ListProjectTiers 是 ListProject 的**可选档位过滤**版本（空 tiers = 不过滤）。 +func (s *Store) ListProjectTiers(ctx context.Context, projectID string, tiers ...RetentionTier) ([]MemoryItem, error) { + return s.listProject(ctx, projectID, tiers) +} + +// listProject 是 ListProject / ListProjectTiers 的共同实现（tiers 为空不过滤）。 +func (s *Store) listProject(ctx context.Context, projectID string, tiers []RetentionTier) ([]MemoryItem, error) { if strings.TrimSpace(projectID) == "" { return nil, ErrMissingProjectID } now := timeGateNow() - rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind, + query := `SELECT id, persona_id, role_id, project_id, scope, kind, content, tags, confidence, source_conversation, created_at, updated_at, - decay_weight FROM memory_items WHERE project_id = ? AND status = ?`+timeGateClause+` - ORDER BY created_at DESC, rowid DESC`, projectID, string(tidyStatusActive), now, now) + decay_weight, retention_tier FROM memory_items WHERE project_id = ? AND status = ?` + args := []any{projectID, string(tidyStatusActive)} + query, args, err := retentionTierList(query, args, tiers) + if err != nil { + return nil, err + } + rows, err := s.db.QueryContext(ctx, query+timeGateClause+` + ORDER BY created_at DESC, rowid DESC`, append(args, now, now)...) if err != nil { return nil, fmt.Errorf("memory: list project items: %w", err) } diff --git a/internal/runtime/memory/store.go b/internal/runtime/memory/store.go index 2a646d3..35ec82e 100644 --- a/internal/runtime/memory/store.go +++ b/internal/runtime/memory/store.go @@ -113,11 +113,23 @@ type MemoryItem struct { // 零值 = 不设过期，落库为空串，四条列表读路径恒可见；非零值落库为定宽 UTC // 文本，到期后该行即从读路径消失，直到后台清扫物理删除（design/23 §19）。 // + // 它与 RetentionTier 是**同一根轴的两种表达**（会到期 ⇔ temp），因此两者由 + // `Store.Add` 里的同一个判定与同一处换算落库，绝不会分叉：显式给非零值而 + // 档位不是 temp 会被拒（ErrRetentionConflict），给 temp 而不给则按临时 TTL 算。 + // // ⚠️ **本字段目前只写不读**：Add 会把它写进 expires_at 列，但读路径一律不 // SELECT 该列（闸是纯 SQL，见 timegate.go），故 Get/List* 回来的条目 - // ExpiresAt 恒为零值。要读回时效需同步 13 列 SELECT 与 scanItem，并连带 + // ExpiresAt 恒为零值。要读回时效需同步 14 列 SELECT 与 scanItem，并连带 // tidy/preinsert 的内部召回 SQL —— 那是改扫描口径，不在本切片授权范围。 ExpiresAt time.Time + + // RetentionTier 是「留多久 / 干什么用」这根轴（迁移 0026、retention.go）。 + // 它与 Kind（关于什么）、Scope（谁的）**正交，互不推导**。 + // + // 写入侧语义：空串 = 由 `Store.Add` 的单一判定填（见 resolveRetention）； + // 非空 = 调用方的显式断言，判据不许推翻它。读回来时： + // **空串 = 尚未分型**（0026 之前的历史行 / 受限来源新行），绝不是某一档。 + RetentionTier RetentionTier } // 区分 memory 包各类失败的哨兵错误。每个都在 @@ -135,6 +147,15 @@ var ( // ErrInvalidKind 在条目的 Kind 不是已定义 kind 时返回。 ErrInvalidKind = errors.New("memory: invalid kind") + // ErrInvalidRetentionTier 在条目的 RetentionTier（或列表读路径的档位过滤） + // 不是闭集内的值时返回。空串**不在**此列：它是「尚未分型」这一合法状态。 + ErrInvalidRetentionTier = errors.New("memory: invalid retention tier") + + // ErrRetentionConflict 在条目同时给出「非 temp 的档位」与「非零的 expires_at」 + // 时返回 —— 那是自相矛盾的输入（一行标着长期却会到期消失），既不猜也不静默 + // 丢弃其中一半。 + ErrRetentionConflict = errors.New("memory: retention tier conflicts with expires_at") + // ErrNotFound 在条目 id 不存在时返回。 ErrNotFound = errors.New("memory: item not found") @@ -215,6 +236,11 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) if err := ensureIngestReviewSchema(ctx, db); err != nil { return nil, err } + // 「留多久 / 干什么用」轴（迁移 0026）：既有库补列 + 建读侧索引。它排在 + // 建表与整理列之后，因为索引依赖的列必须先存在（见 retention_schema.go）。 + if err := ensureRetentionSchema(ctx, db); err != nil { + return nil, err + } if err := normalizeMemoryTimestamps(ctx, db); err != nil { return nil, err } @@ -246,6 +272,13 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) // `expires_at` 列：Add 是全仓唯一 INSERT 路径（design/23 §15.1 D3-1）， // 因此时效列的真生产者**只能**在这里 —— 读侧时效闸（timegate.go）比对的 // 就是这一列，若写入侧不写出它，闸就退化成恒真的空壳。 +// +// 🔴 **档位判定与有效期换算都在这一处**（迁移 0026、retention.go）： +// `resolveRetention` 是全仓唯一的档位判据（调用方显式档位 > 显式有效期 > +// 受限来源不分型 > 留存规则 → temp > 默认 long），`plan.tempDeadline` 是唯一的 +// 「档位 → expires_at」换算。两列因此**同一个判定、同一笔 INSERT**落下， +// 不存在「判了 temp 却没写有效期」的中间态 —— M2 时代登记过的「INSERT 与补写 +// expires_at 之间的短窗口」在本片被结构性关闭（不再有第二笔 UPDATE）。 func (s *Store) Add(ctx context.Context, item MemoryItem) (MemoryItem, error) { if err := item.normalize(); err != nil { return MemoryItem{}, err @@ -254,21 +287,35 @@ func (s *Store) Add(ctx context.Context, item MemoryItem) (MemoryItem, error) { if item, err = applyTimestamps(item); err != nil { return MemoryItem{}, err } + // 🔴 档位判定（唯一入口）：把结果写回条目，闸与异步审查因此**只读**这一位， + // 不再各自跑一遍留存规则（见 preinsert_judge.go 的 defer 分支）。 + verdict, err := resolveRetention(item) + if err != nil { + return MemoryItem{}, err + } + item.RetentionTier = verdict.Tier + if verdict.Rule != retentionNone { + // 具体命中的反例规则码只进日志（进留痕列的是闭集里的 t0_low_value）： + // 「n1~n7 谁命中过」是 M27 体检缺的那一条观测线。 + s.log.Debug("memory: retention judged temporary", + "id", item.ID, "scope", string(item.Scope), "rule", string(verdict.Rule)) + } tagsJSON, err := json.Marshal(item.Tags) if err != nil { return MemoryItem{}, fmt.Errorf("memory: encode tags: %w", err) } plan := s.planIngest(ctx, item) + item.ExpiresAt = plan.ExpiresAt _, err = s.db.ExecContext(ctx, `INSERT INTO memory_items (id, persona_id, role_id, project_id, scope, kind, content, tags, confidence, source_conversation, created_at, updated_at, decay_weight, status, merged_into, - expires_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + expires_at, retention_tier) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, item.ID, item.PersonaID, item.RoleID, item.ProjectID, string(item.Scope), string(item.Kind), item.Content, string(tagsJSON), item.Confidence, item.SourceConversation, formatTime(item.CreatedAt), formatTime(item.UpdatedAt), item.DecayWeight, - plan.insertStatus(), plan.absorbedID(), expiryText(item.ExpiresAt)) + plan.insertStatus(), plan.absorbedID(), plan.expiresAt(), plan.retention()) if err != nil { return MemoryItem{}, fmt.Errorf("memory: insert item: %w", err) } @@ -290,88 +337,6 @@ func (s *Store) Add(ctx context.Context, item MemoryItem) (MemoryItem, error) { return item, nil } -// Get 返回给定 id 的条目。id 为空或 -// 不存在时返回 ErrNotFound。 -// -// 🔴 **刻意不过任何读侧闸**（status 与时效都不过；design/03 §11.2 第 3 条）： -// 撤销软合并必须先读得到被吸收的行，人工核查 / 清算也要先读得到已过期与 -// 已失效的行 —— 加闸等于把审计与撤销的抓手一并抹掉。要「面向用户的可见集」 -// 请走 ListByPersona / ListScope / ListProject / List 或 Retriever（那些都过闸）。 -func (s *Store) Get(ctx context.Context, id string) (MemoryItem, error) { - if strings.TrimSpace(id) == "" { - return MemoryItem{}, ErrNotFound - } - row := s.db.QueryRowContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind, - content, tags, confidence, source_conversation, created_at, updated_at, - decay_weight FROM memory_items WHERE id = ?`, id) - item, err := scanItem(row) - if errors.Is(err, sql.ErrNoRows) { - return MemoryItem{}, ErrNotFound - } - if err != nil { - return MemoryItem{}, fmt.Errorf("memory: get item: %w", err) - } - return item, nil -} - -// ListByPersona 返回 personaID 的全部条目，最新的在前。rowid 打破 -// created_at 并列，使同一时钟刻内写入的条目保持稳定的 -// 最新优先顺序。它是角色隔离边界：属于 -// 其他角色的条目绝不返回，因此角色之间从不共享各自的角色日常 -// 命名空间（design §4）。它只是检索候选集的*自有* -// 那一半 —— 检索器在排名前把它与共享的用户/项目作用域 -// 取并集（design §3/§4，见 Retriever.visibleCandidates）。 -// -// 只返回**在册**（status='active'）条目：整理与入库前审查都只做软动作， -// 被合并/被归档的行内容原样留在表里，因此「能看到什么记忆」的每条读路径 -// 都必须过 status 闸，否则清理过的行照样会被检索到、照样占满面板。 -// -// 同时过**时效闸**（timeGateClause）：已过期（expires_at ≤ now）与已失效 -// （valid_to ≤ now）的行一并不再返回 —— 判据细节与「为什么必须下沉到这里」 -// 见 timegate.go 与 design/03 §9.6/§11.2。 -func (s *Store) ListByPersona(ctx context.Context, personaID string) ([]MemoryItem, error) { - if strings.TrimSpace(personaID) == "" { - return nil, ErrMissingPersona - } - // 闸的左右操作数同源：两边都由 formatTime 产出定宽 UTC 文本。 - now := timeGateNow() - rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind, - content, tags, confidence, source_conversation, created_at, updated_at, - decay_weight FROM memory_items WHERE persona_id = ? AND status = ?`+timeGateClause+` - ORDER BY created_at DESC, rowid DESC`, - personaID, string(tidyStatusActive), now, now) - if err != nil { - return nil, fmt.Errorf("memory: list persona items: %w", err) - } - return scanItems(rows, "persona item") -} - -// List 返回跨所有角色的每条**在册**记忆条目，最新的在前。rowid -// 打破 created_at 并列，使同一时钟刻内写入的条目保持 -// 稳定的最新优先顺序。它是聚合/管理视图，不受角色隔离边界限制；必须 -// 遵守隔离的调用方应改用 ListByPersona 或 ListScope。返回的 -// 切片非 nil。 -// -// 「在册」= status='active'：被软合并（'merged'）与被归档（'archived'） -// 的行**内容与行都还在**，但不再参与检索、不再进上下文、也不再出现在 -// 管理面板的记忆列表里（design/23 §8/§10 的软动作口径）。要看它们走 -// 审查留痕 `GET /admin/memories/reviews` 或按 id 用 Get —— 后者刻意 -// 不过 status 闸，因为撤销软合并必须先读得到被吸收的行。 -// -// 同样过**时效闸**：临时记忆到期（expires_at ≤ now）与事实失效 -// （valid_to ≤ now）的行一并从这里消失（timegate.go、design/03 §9.6）。 -func (s *Store) List(ctx context.Context) ([]MemoryItem, error) { - now := timeGateNow() - rows, err := s.db.QueryContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind, - content, tags, confidence, source_conversation, created_at, updated_at, - decay_weight FROM memory_items WHERE status = ?`+timeGateClause+` - ORDER BY created_at DESC, rowid DESC`, string(tidyStatusActive), now, now) - if err != nil { - return nil, fmt.Errorf("memory: list items: %w", err) - } - return scanItems(rows, "item") -} - // Delete 删除给定 id 的条目，id 为空或 // 没有删除任何行时返回 ErrNotFound。调用方只应传入经 // 角色作用域读取（ListByPersona/Get）得到的 id，使删除始终留在 @@ -414,6 +379,10 @@ func (s *Store) RememberManual(ctx context.Context, personaID, roleID, text stri // normalize 填充默认的 Scope/Kind/Confidence/DecayWeight，然后校验。 // 它由 Add 调用，因此调用方可以省略取值，同时仍能 // 捕获显式非法值。 +// +// RetentionTier **不在这里填默认值**：空串是「还没分型」这一合法状态，它由 Add 的 +// 单一判定（resolveRetention）落定 —— 在 normalize 里塞一个默认档会把「判定没跑」 +// 与「判定结论是 long」变成同一件事。 func (it *MemoryItem) normalize() error { if it.Scope == "" { it.Scope = ScopePersona @@ -445,6 +414,9 @@ func (it *MemoryItem) validate() error { if !it.Kind.valid() { return ErrInvalidKind } + if !it.RetentionTier.valid() { + return fmt.Errorf("%w: %q", ErrInvalidRetentionTier, it.RetentionTier) + } return nil } diff --git a/internal/runtime/memory/store_list.go b/internal/runtime/memory/store_list.go new file mode 100644 index 0000000..bc0a52f --- /dev/null +++ b/internal/runtime/memory/store_list.go @@ -0,0 +1,144 @@ +package memory + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" +) + +// 本文件是 memory_items 的**读路径**（按 id 取 / 三条列表 + 它们的档位过滤版本）， +// 从 store.go 拆出来只为守 AGENTS §10 的文件行数上限（≤500 行）与「模块按职责拆 +// 源文件」那条规范 —— 口径逐字未变。 +// +// 四条可见性判据都下沉在这里：`status='active'`、时效闸（timeGateClause，见 +// timegate.go）、作用域/角色过滤，以及切片的「留多久 / 干什么用」档位过滤。 +// `Get` 是**唯一刻意不过闸**的读口（撤销软合并与清算要先读得到被吸收/已过期的行）。 +// +// 🔴 档位过滤走**同族方法**（ListByPersonaTiers / ListTiers）而不是把原方法改成 +// 可变参数：`List` / `ListByPersona` / `ListScope` / `ListProject` 的精确签名已被 +// 网关的 `MemoryStore` 接口钉住，可变参数会让 `*Store` 不再实现那个接口。 + +// Get 返回给定 id 的条目。id 为空或 +// 不存在时返回 ErrNotFound。 +// +// 🔴 **刻意不过任何读侧闸**（status 与时效都不过；design/03 §11.2 第 3 条）： +// 撤销软合并必须先读得到被吸收的行，人工核查 / 清算也要先读得到已过期与 +// 已失效的行 —— 加闸等于把审计与撤销的抓手一并抹掉。要「面向用户的可见集」 +// 请走 ListByPersona / ListScope / ListProject / List 或 Retriever（那些都过闸）。 +func (s *Store) Get(ctx context.Context, id string) (MemoryItem, error) { + if strings.TrimSpace(id) == "" { + return MemoryItem{}, ErrNotFound + } + row := s.db.QueryRowContext(ctx, `SELECT id, persona_id, role_id, project_id, scope, kind, + content, tags, confidence, source_conversation, created_at, updated_at, + decay_weight, retention_tier FROM memory_items WHERE id = ?`, id) + item, err := scanItem(row) + if errors.Is(err, sql.ErrNoRows) { + return MemoryItem{}, ErrNotFound + } + if err != nil { + return MemoryItem{}, fmt.Errorf("memory: get item: %w", err) + } + return item, nil +} + +// ListByPersona 返回 personaID 的全部条目，最新的在前。rowid 打破 +// created_at 并列，使同一时钟刻内写入的条目保持稳定的 +// 最新优先顺序。它是角色隔离边界：属于 +// 其他角色的条目绝不返回，因此角色之间从不共享各自的角色日常 +// 命名空间（design §4）。它只是检索候选集的*自有* +// 那一半 —— 检索器在排名前把它与共享的用户/项目作用域 +// 取并集（design §3/§4，见 Retriever.visibleCandidates）。 +// +// 只返回**在册**（status='active'）条目：整理与入库前审查都只做软动作， +// 被合并/被归档的行内容原样留在表里，因此「能看到什么记忆」的每条读路径 +// 都必须过 status 闸，否则清理过的行照样会被检索到、照样占满面板。 +// +// 同时过**时效闸**（timeGateClause）：已过期（expires_at ≤ now）与已失效 +// （valid_to ≤ now）的行一并不再返回 —— 判据细节与「为什么必须下沉到这里」 +// 见 timegate.go 与 design/03 §9.6/§11.2。 +// +// 档位过滤见同族的 ListByPersonaTiers（签名不变的理由见文件头注）。 +func (s *Store) ListByPersona(ctx context.Context, personaID string) ([]MemoryItem, error) { + return s.listByPersona(ctx, personaID, nil) +} + +// ListByPersonaTiers 是 ListByPersona 的**可选档位过滤**版本（迁移 0026）：返回在册 +// 条目中档位落在 tiers 里的那些。tiers 为空 = 与 ListByPersona 逐字等价（不过滤）。 +// +// 它存在的理由是把「留多久 / 干什么用」这根轴变成**可查询**的：面板数「临时层有几条」 +// 与回填分型（`RetentionTierUnset`）都要能按档位取行，而不是读全表再在内存里筛 +// （那会在 4k 行的存量上把一次查询变成一次全表扫描 + 一次全量分配）。 +func (s *Store) ListByPersonaTiers(ctx context.Context, personaID string, tiers ...RetentionTier) ([]MemoryItem, error) { + return s.listByPersona(ctx, personaID, tiers) +} + +// listByPersona 是 ListByPersona / ListByPersonaTiers 的共同实现（tiers 为空不过滤）。 +func (s *Store) listByPersona(ctx context.Context, personaID string, tiers []RetentionTier) ([]MemoryItem, error) { + if strings.TrimSpace(personaID) == "" { + return nil, ErrMissingPersona + } + // 闸的左右操作数同源：两边都由 formatTime 产出定宽 UTC 文本。 + now := timeGateNow() + query := `SELECT id, persona_id, role_id, project_id, scope, kind, + content, tags, confidence, source_conversation, created_at, updated_at, + decay_weight, retention_tier FROM memory_items WHERE persona_id = ? AND status = ?` + args := []any{personaID, string(tidyStatusActive)} + query, args, err := retentionTierList(query, args, tiers) + if err != nil { + return nil, err + } + rows, err := s.db.QueryContext(ctx, query+timeGateClause+` + ORDER BY created_at DESC, rowid DESC`, append(args, now, now)...) + if err != nil { + return nil, fmt.Errorf("memory: list persona items: %w", err) + } + return scanItems(rows, "persona item") +} + +// List 返回跨所有角色的每条**在册**记忆条目，最新的在前。rowid +// 打破 created_at 并列，使同一时钟刻内写入的条目保持 +// 稳定的最新优先顺序。它是聚合/管理视图，不受角色隔离边界限制；必须 +// 遵守隔离的调用方应改用 ListByPersona 或 ListScope。返回的 +// 切片非 nil。 +// +// 「在册」= status='active'：被软合并（'merged'）与被归档（'archived'） +// 的行**内容与行都还在**，但不再参与检索、不再进上下文、也不再出现在 +// 管理面板的记忆列表里（design/23 §8/§10 的软动作口径）。要看它们走 +// 审查留痕 `GET /admin/memories/reviews` 或按 id 用 Get —— 后者刻意 +// 不过 status 闸，因为撤销软合并必须先读得到被吸收的行。 +// +// 同样过**时效闸**：临时记忆到期（expires_at ≤ now）与事实失效 +// （valid_to ≤ now）的行一并从这里消失（timegate.go、design/03 §9.6）。 +// +// 档位过滤走同族的 ListTiers（签名不变的理由见文件头注）。 +func (s *Store) List(ctx context.Context) ([]MemoryItem, error) { + return s.listAll(ctx, nil) +} + +// ListTiers 是 List 的**可选档位过滤**版本（口径与 ListByPersonaTiers 一致： +// 空 tiers = 不过滤；`RetentionTierUnset` = 只看尚未分型的行）。 +func (s *Store) ListTiers(ctx context.Context, tiers ...RetentionTier) ([]MemoryItem, error) { + return s.listAll(ctx, tiers) +} + +// listAll 是 List / ListTiers 的共同实现（tiers 为空不过滤）。 +func (s *Store) listAll(ctx context.Context, tiers []RetentionTier) ([]MemoryItem, error) { + now := timeGateNow() + query := `SELECT id, persona_id, role_id, project_id, scope, kind, + content, tags, confidence, source_conversation, created_at, updated_at, + decay_weight, retention_tier FROM memory_items WHERE status = ?` + args := []any{string(tidyStatusActive)} + query, args, err := retentionTierList(query, args, tiers) + if err != nil { + return nil, err + } + rows, err := s.db.QueryContext(ctx, query+timeGateClause+` + ORDER BY created_at DESC, rowid DESC`, append(args, now, now)...) + if err != nil { + return nil, fmt.Errorf("memory: list items: %w", err) + } + return scanItems(rows, "item") +} diff --git a/internal/runtime/memory/store_rows.go b/internal/runtime/memory/store_rows.go index b507f09..05d3b72 100644 --- a/internal/runtime/memory/store_rows.go +++ b/internal/runtime/memory/store_rows.go @@ -42,20 +42,25 @@ func scanItems(rows *sql.Rows, op string) ([]MemoryItem, error) { // scanItem 读取一行 memory_items。tags 以 JSON 数组字符串存储； // 时间戳以定宽 UTC 字符串（timeLayout）存储，因此文本 // 比较即可按时间排序。 +// +// retention_tier 是**第 14 列**（迁移 0026）：它与 `kind` 是两根独立的轴， +// 因此这里只做「读出来」，不参与任何推导。空串原样成为 +// RetentionTierUnset（= 尚未分型），绝不被折成某一档。 func scanItem(sc rowScanner) (MemoryItem, error) { var ( it MemoryItem - scope, kind string + scope, kind, tier string tagsJSON string createdAt, updatedAt string ) if err := sc.Scan(&it.ID, &it.PersonaID, &it.RoleID, &it.ProjectID, &scope, &kind, &it.Content, &tagsJSON, &it.Confidence, &it.SourceConversation, - &createdAt, &updatedAt, &it.DecayWeight); err != nil { + &createdAt, &updatedAt, &it.DecayWeight, &tier); err != nil { return MemoryItem{}, err } it.Scope = Scope(scope) it.Kind = Kind(kind) + it.RetentionTier = RetentionTier(tier) if err := json.Unmarshal([]byte(tagsJSON), &it.Tags); err != nil { return MemoryItem{}, fmt.Errorf("memory: decode tags: %w", err) } diff --git a/internal/runtime/memory/store_schema.go b/internal/runtime/memory/store_schema.go index 5a21e50..4c3788e 100644 --- a/internal/runtime/memory/store_schema.go +++ b/internal/runtime/memory/store_schema.go @@ -26,7 +26,8 @@ CREATE TABLE IF NOT EXISTS memory_items ( merge_count INTEGER NOT NULL DEFAULT 1, occurred_at TEXT NOT NULL DEFAULT '', valid_to TEXT NOT NULL DEFAULT '', - expires_at TEXT NOT NULL DEFAULT '' + expires_at TEXT NOT NULL DEFAULT '', + retention_tier TEXT NOT NULL DEFAULT '' )` const createMemoryItemsPersonaIndex = ` @@ -36,3 +37,12 @@ CREATE INDEX IF NOT EXISTS idx_memory_items_persona_id const createMemoryItemsPersonaCreatedIndex = ` CREATE INDEX IF NOT EXISTS idx_memory_items_persona_id_created_at ON memory_items (persona_id, created_at)` + +// createMemoryItemsRetentionTierIndex 是「留多久 / 干什么用」轴的读侧索引（迁移 0026）。 +// 三条列表读路径的可选档位过滤（`retention_tier IN (...)`）与「只看未分型的行」 +// （回填分型的入口）都吃它。列**必须**排在 CREATE TABLE 的最后：迁移的 +// `ALTER TABLE ADD COLUMN` 也把新列追加在末尾，两份 DDL 才逐列同形 +// （retention_migration_test.go 钉住这条同形，少一列就是「新库有、老库无」的漂移）。 +const createMemoryItemsRetentionTierIndex = ` +CREATE INDEX IF NOT EXISTS idx_memory_items_retention_tier + ON memory_items (retention_tier)` diff --git a/internal/runtime/memory/tidy_dedupe.go b/internal/runtime/memory/tidy_dedupe.go index 138d7bc..7deaf1f 100644 --- a/internal/runtime/memory/tidy_dedupe.go +++ b/internal/runtime/memory/tidy_dedupe.go @@ -336,9 +336,9 @@ func dedupeGroupEligible(it MemoryItem) bool { // tidyDedupeMember 是分簇用的条目视图：记忆行 + merge_count。 // -// 为什么要单独带 merge_count：`MemoryItem` 的读侧 SELECT（store.go 的 13 列）**不含** +// 为什么要单独带 merge_count：`MemoryItem` 的读侧 SELECT（store_list.go 的 14 列）**不含** // `merge_count`（它只写不读），而「存活行选谁」的次要判据正是它（合并次数多的那条 -// 已经吸收过别的表述，留它信息量最大）。为一个次要判据去改 13 列读面会牵动 +// 已经吸收过别的表述，留它信息量最大）。为一个次要判据去改 14 列读面会牵动 // Get/List*/preinsert/tidy 四处 SQL，风险远大于收益；改为按分区一次轻量查询把 // id → merge_count 取出来（见 Store.dedupeMergeCounts）。 type tidyDedupeMember struct { diff --git a/internal/runtime/memory/tidy_store.go b/internal/runtime/memory/tidy_store.go index 8c0f532..d2f926a 100644 --- a/internal/runtime/memory/tidy_store.go +++ b/internal/runtime/memory/tidy_store.go @@ -50,7 +50,7 @@ const tidyStatusActive = "active" // 查询（listTidyPartitionsSQL / countTidyPartitionItemsSQL）是同一裁定的实例。 const listTidyPartitionItemsSQL = `SELECT id, persona_id, role_id, project_id, scope, kind, content, tags, confidence, source_conversation, created_at, updated_at, - decay_weight FROM memory_items + decay_weight, retention_tier FROM memory_items WHERE persona_id = ? AND scope = ? AND project_id = ? AND status = ? ORDER BY created_at DESC, rowid DESC LIMIT ?` diff --git a/internal/runtime/memory/timegate.go b/internal/runtime/memory/timegate.go index ef075ab..bd3d445 100644 --- a/internal/runtime/memory/timegate.go +++ b/internal/runtime/memory/timegate.go @@ -18,7 +18,7 @@ import "time" // 🔴 同时登记**两处刻意不过闸**（design/03 §11.2 第 3 条；与 design/23 §10.1 ① // 已立的先例同构）： // -// 1. 按 id 的 Store.Get（store.go）—— 撤销软合并 / 查看已软删行必须先读得到， +// 1. 按 id 的 Store.Get（store_list.go）—— 撤销软合并 / 查看已软删行必须先读得到， // 加闸等于把撤销与审计的抓手抹掉。过期行同理：清算与人工核查要读得到。 // 2. 整理与审查的**内部**召回 SQL —— listTidyPartitionItemsSQL（tidy_store.go） // 与 listReviewCandidatesSQL（preinsert_store.go）刻意**只过 status 闸**： @@ -64,7 +64,7 @@ func timeGateNow() string { // // 它不做任何判据（过去值不在这里被拒、也不被改写）：写入侧不决定可见性， // 一条「出生即过期」的行照常落库，四条列表读路径立刻看不见它，直到后台清扫 -// 把它删掉；Get 仍读得到（见 store.go 的 Get 注释）。 +// 把它删掉；Get 仍读得到（见 store_list.go 的 Get 注释）。 func expiryText(t time.Time) string { if t.IsZero() { return ""