Files
Cyrene_v2/internal/runtime/memory/preinsert.go
T
AskaEth 91101dedfb feat(memory): M45 提取器时间性元数据位 temporal 驱动留存判定
背景(M37 复审的原话):「真正卡住上游的是提取器的改写:一次性事件被写成
30~80 字的第三人称长期陈述……这不是阈值问题,是判定链缺信息位。」本片补的
就是那一位,不动 N1~N7 的任何阈值。

产出侧
- temporal.go:TemporalKind 闭集(one_off / recurring / durable / unknown,
  空串 = 未分型),normalizeTemporal 只认四个字面值,其余一律归 unknown(绝不猜
  同义词);提取提示词追加该枚举(模型只负责标注,判定在 Go 闭集 switch 里)。
- llmextractor:候选解析后归一并随 MemoryItem 带出;非法/缺失值不报错、不拖累
  同批其他候选。

落库
- 迁移 0027_memory_temporal(up: ADD COLUMN temporal TEXT NOT NULL DEFAULT '';
  down: DROP COLUMN,只撤结构、不改写任何行数据)+ NewStore 的自举前滚
  (duplicate column 是预期路径);Store.Add 的同一笔 INSERT 写该列,判定与
  留痕不分叉。

判定(仍是 resolveRetention 一处收口,judgeIngest 一字未改)
- one_off ⇒ temp(+ expires_at),但先过 temporalSanityVeto 六条语义安全阀:
  显式「记住」/ 在陈述约定 / 习惯词 / 硬性长期事实词 / 永久化 / 状态改变 ——
  命中即不给结论、落回 N1~N7。依据是离线重放的实测:去掉「记住+约定」两条时
  21 条长期约定会被判 temp(4.80%),全开后在最近 500 条窗口内翻转条数为 0。
- durable / recurring ⇒ long 且不再跑规则;unknown / 未分型 ⇒ 落回 N1~N7
  (不许因为「正文看着像今天的事」就判 temp)。
- IngestRuleVersion d3.4 → d3.5(判定结论会变,留痕必须可分)。

验证
- go build ./... + go vet ./... 全绿;go test -count=1 ./internal/runtime/memory/...
  ./internal/db/... 全绿;memory -race 全绿。
- 新增 temporal_test.go(验收点 ①~⑤ + 两条交叉护栏 + 闭集/归一化)与
  temporal_migration_test.go(0027 up/down/同形/真实启动顺序)。变异实测:
  摘掉 one_off 支路 ⇒ ① 全红;关掉 temporalSanityVeto ⇒ 三条护栏用例全红。
- 迁移测试的「回滚 N 步」改为从 schema_migrations 现算(migrateDownToBefore),
  新加一条迁移不再打红 0025/0026 的用例。
2026-10-26 18:57:38 +08:00

538 lines
25 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package memory
import (
"context"
"errors"
"fmt"
"log/slog"
"sync"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding"
)
// 本文件是 D3「入库前审查」(design/23-记忆整理与去重 §15.1 D3)的服务形态:
// 新记忆落库**之后**,由**独立的记忆管理会话**(复用既有的 sess-memorytidy
// 装配,见 TidySessionID)异步判定它与同分区既有记忆的关系,四态结论见
// IngestDecision。
//
// 与 design §15 D3-1/3/5 的取舍(逐条,因为设计稿这一处同时出现「写入之前」
// 与「首选异步 + 结论回写」两种表述,实现必须把口径写死):
//
// 1. **触发点在写入之后**(异步):回合收尾的写入路径上多出一次 embedding
// 网络调用就会把主人的回合拖长(§15 D3-3「不许因为审查把对话回合拖长」),
// 而 §15 D3-3 要求的正是「审查结论**回写**到 memory_items」—— 只有行
// 已存在才谈得上回写。因此本实现是「先落库、再由审查把重复条目软合并
// 回去」,在观测上与「不新建」等价(活跃行数不变),且**不阻塞任何回合**;
// 2. **失败即安全**:条目先落库 ⇒ 审查失败/超时/embedder 不可用时「记忆绝不
// 丢」是结构性成立的,不靠回滚(§15 D3-5)。失败路径**只记 WARN、绝不
// 改写任何一行**;
// 3. **规则优先、模型兜底**:T0 是纯规则(design §15 D3-2 T0),**恒为零模型
// 调用**;T1(灰区调一次轻量模型,§15.1 D3-2)默认**开**,但只在「T0 判不出
// 结论且存在灰区候选」时才发出**至多一次**调用(preinsert_t1.go),并且
// 要有模型出口才成立 —— 两者任一缺席即是结构性零模型调用。每一次这样的
// 调用还要先过**全局限速**(默认 120 次/小时,preinsert_t1_limit.go):
// 超限即跳过 T1、回落 create + `t1_throttled`。
// **同步入库闸(preinsert_gate.go)不碰 T1**,写入路径因此仍是零模型调用。
// 4. **第五态 `defer` 的判定在两层各有一处来源**(design/23 §18.2):
// **T0 提取留存规则闸**(`extract_retention.go` → `judgeIngest`,纯规则、
// 零模型)拦「结构上显然不值得留」的(`reason=t0_low_value`);**T1 则把
// defer 收进同一个闭集**(`merge/modify/keep/defer`,**同一次调用**、
// 同一个 prompt、同一套候选编号),只在关系判为「真新」时才问「值不值得
// 长期留」(`reason=t1_defer`)。**绝不新增第二次模型调用**。
// 两条失败路径都回落 `create`(= 当作长期记住),fail-safe 方向恒为
// 「**宁可多留,不可误弃**」(§18.3)。
//
// 代价(如实登记):极短窗口内库里会多一行待裁定的重复条目,随后被软合并
// (status='merged' + merged_into,可 revert,绝不物理删除,design §8)。
// IngestDecision 是入库前审查的四态结论(design/23 §15.1 D3)。
type IngestDecision string
const (
// IngestSkip 表示已有**归一化后完全等价**的条目:新行折回既有行,
// 不产生第二条活跃记忆(§15 D3-2 T0-1)。
IngestSkip IngestDecision = "skip"
// IngestMerge 表示新条目是既有条目事实集合的**子集**:并入既有,
// 既有行存活(§15 D3-2 T0-2)。
IngestMerge IngestDecision = "merge"
// IngestModify 表示既有条目是新条目事实集合的子集:**新行承载更全的
// 内容**,旧行置为被合并(§15 D3-6「改写必须是新行 + 旧行置为被合并」)。
IngestModify IngestDecision = "modify"
// IngestCreate 表示确属新条目(互不包含 / 被规则否决 / 没有召回):
// 正常新建,不做任何改写(§15 D3-2 T0-3)。
IngestCreate IngestDecision = "create"
// IngestDefer 是**第五态**(design/23 §18.1):与既有记忆不冲突(真新),
// 但不值得长期留存 ⇒ **照常插入**,只把 `expires_at` 写成「现在 + 临时 TTL」
// (默认 7 天,见 tempTtlSec)落进**临时层**:到期由读侧时效闸隐藏
// (design/03 §9.6)、再由后台清扫物理删除(design/23 §19)。
//
// 它与其余四态的正交关系(§18.2 的不变式):`defer` ≡「关系判为**真新**
// ∧ 价值判为**短期**」。因此它**只在被判为真新时**才可能产生 —— 已被既有
// 记忆代表的那条(merge/modify/skip)留不留由既有行决定,与价值判定无关。
//
// 🔴 仍是**软动作**:只写 `expires_at` 一列,**不删行、不改正文**;回滚 =
// 清空该列(§18.4)。选 `defer` 而非 `drop` 是 fail-safe 取向的落点
// (§18.3「宁可多留,不可误弃」):最坏后果只是「多留 7 天」。
IngestDefer IngestDecision = "defer"
)
// IngestRuleVersion 是入库前审查规则集的版本号,进每条审查留痕(§15 D3-8)。
// **改判定规则必须同时升它**,否则新结论与旧留痕无法区分。
//
// 版本史:
// - `d3.1` 四态规则层(T0)+ 同步入库闸;
// - `d3.2` 加入 **T1 灰区模型判定**(preinsert_t1.go,design/23 §15.1 D3-2):
// 灰区(相似、互不包含、未被规则否决)在模型出口接上时会被改判成
// merge/modify,因此 `reason` 上出现了 `t1_*` 族。
// - `d3.3` 加入 **第五态 `defer`**(design/23 §18.1):T0 提取留存规则闸
// (`t0_low_value`)与 T1 的同一次调用(`t1_defer`)都能给出它,落点 =
// 照常插入 + 写短 `expires_at`。**升版是硬要求**:同一对条目在 d3.2 下是
// `create`、在 d3.3 下可能是 `defer`,不升版新旧留痕无法区分(§18.4)。
// - `d3.4`(切片 M37)把「判 `defer`」与「判 `retention_tier=temp`」**合并成一个
// 判定**(迁移 0026):档位由 `Store.Add` 的 `resolveRetention` 一次落定
// (调用方显式档位 > 显式有效期 > 受限来源不分型 > 留存规则 → temp > 默认 long),
// `defer` 从此只是「档位是 temp ∧ 关系判为真新」,闸与异步审查都不再各跑一遍
// 规则。**规则内容与阈值未改**(同一份 N1~N7),但**判据的作用范围变了**:
// 档位与 `expires_at` 现在对**每一行**在同一笔 INSERT 里落下,且不再依赖同步闸
// 是否开启 —— 同一对条目在 d3.3 与 d3.4 下的落库形态因此可能不同,按规矩升版。
// - `d3.5`(切片 M45)加入提取器的**时间性元数据位**(`temporal`,迁移 0027):
// `one_off` ⇒ temp(⇒ 第五态 defer)、`durable` / `recurring` ⇒ long、
// `unknown` / 未分型 ⇒ 照旧走 N1~N7。**N1~N7 的规则内容与阈值一字未改**,但
// 同一段正文在 d3.4 与 d3.5 下可能落到不同档位(d3.4 只能靠词判,d3.5 还能
// 读到「这是不是一次性事件」这一位)—— 结论会变,按规矩升版,否则新旧留痕
// 无法区分。
const IngestRuleVersion = "d3.5"
// 调优旋钮(ReviewConfig)与它们的默认值常量在 preinsert_config.go。
var (
// ErrReviewDisabled 在审查被配置显式关闭却又被直接调用时返回。
ErrReviewDisabled = errors.New("memory/review: disabled by configuration")
// ErrReviewRestrictedSource 在待审条目来自群聊/陌生人这类**只读**上下文时
// 返回。它不是失败:这类条目凭什么都不该被审查(既不合并也不改写,
// design §6 / §15 D3-7 + AGENTS §9 的作用域隔离红线)。
ErrReviewRestrictedSource = errors.New("memory/review: restricted source is read-only")
// ErrReviewItemNotActive 在该条目已经不在册(被前一次审查合并掉)却又被
// 要求审查时返回。它是幂等保护:再判一次会把存活行的 merge_count 重复加一。
ErrReviewItemNotActive = errors.New("memory/review: item is no longer active")
)
// ReviewDeps 是注入给入库前审查的依赖。Store 必需;其余可空并优雅降级。
type ReviewDeps struct {
// Store 是记忆存储。必需。
Store *Store
// Embedder 是候选召回的相似度来源。为 nil 时退回零依赖的字符二元组
// 余弦(与整理生产轮的回退同一口径,见 tidy_recall.go)。
Embedder embedding.Embedder
// Audit 是审计出口(可空;为 nil 时只记日志)。
Audit func(who, what, result string)
// Now 是墙钟来源,便于测试注入。
Now func() time.Time
// Log 是基础日志器,来源标记为 "memory/review"。
Log *slog.Logger
// GateCandidates 是**同步入库闸**的候选读取出口。为 nil 时用 store 的
// 生产口径(Store.listReviewCandidates,分区过滤写在 SQL 里)。测试用它
// 注入慢/坏的候选源,验证「预算有界」与「超时 fail-open」这两条。
GateCandidates func(ctx context.Context, p tidyPartition, excludeID string, limit int) ([]MemoryItem, error)
// Judge 是 **T1 灰区判定**的一次性模型调用口(design/23 §15.1 D3-2)。
//
// 为 nil 时 T1 **结构性关闭**(零模型调用)—— 这正是「T1 关闭 ⇒ 零模型调用」
// 从承诺变成事实的那条线。它只取模型适配器的 Chat 一面(T1 不流式、不重试、
// 不并发),注入的是**既有** model.Adapter 的调用口(组合根装配的就是模型栈
// 本身,含按角色路由与主→备→本地降级),不新造任何 LLM 客户端。
//
// ⚠️ 它**只被异步审查层使用**(PreInsertReviewer.Review);同步入库闸
// (preinsert_gate.go)的判定链里根本没有它,写入路径因此零模型调用。
Judge T1Judge
}
// IngestReview 是一条入库前审查留痕(design/23 §15 D3-8)。它**不含记忆正文**:
// 命中的既有 id 足以让 owner-only 的读侧回表取原文。
type IngestReview struct {
ID string
ItemID string
PersonaID string
Scope Scope
ProjectID string
Decision IngestDecision
Reason string
// Veto 是命中的否决规则(`merge`/`modify`/`skip` 时为空):它是结构化依据,
// 面板据此聚合「为什么这一条没被合并」,而不是去解析 reason 字符串。
Veto TidyVetoReason
TargetID string
MergedIDs []string
Score float64
RecallMode string
Degraded bool
RuleVer string
SessionID string
CreatedAt time.Time
}
// PreInsertReviewer 是入库前审查服务。用 NewPreInsertReviewer 构建,用 Start
// 拉起工作协程、Stop 停它;Review 是同步入口(测试与手动核查用)。
type PreInsertReviewer struct {
cfg ReviewConfig
deps ReviewDeps
log *slog.Logger
now func() time.Time
queue chan MemoryItem
// t1Gate 是 T1 的**全局限速**闸(design/23 §15.4)。挂在实例上而不是包级
// 变量:AGENTS §10 禁包级全局变量,而组合根只建一个审查服务 ⇒ 实例字段
// 事实上就是「全进程共享一个窗口额度」。它与 r.now 同源,因此测试可以整体
// 推进时钟(见 preinsert_t1_limit.go)。
t1Gate *t1RateGate
// mu 保护 running / cancel 两个生命周期字段,以及**运行期可变的那两个配置
// 字段**(`DisableIngestGate` / `TempTTL`,面板调优会改它们,见
// preinsert_tuning.go 的 ApplyPanelTuning)。
//
// 它用 RWMutex 而不是 Mutex,是因为后者在**写入路径**上被高频读:同步入库闸
// 每写一条记忆就要问一次 `gateEnabled()`,第五态 `defer` 还要问一次
// `deferDeadline()`;读锁之间不互斥,因此这两处不会因为面板保存而互相排队。
//
// 🔴 口径(切片 MEMPANEL / M4 立):**谁在运行期写 cfg,谁就必须让读它的地方
// 也持这把锁**。今天运行期只写上面那两个字段,因此只有读这两个字段的两处
// (`gateEnabled` / `deferDeadline`)走 `tuningSnapshot`;其余读点读的是构造期
// 冻结的字段,不与写方交手,故**没有**一并加锁(不给热路径加无谓的锁)。
// 将来面板再接管别的字段,那几处读点必须与它一起收。
mu sync.RWMutex
running bool
cancel context.CancelFunc
wg sync.WaitGroup
statsMu sync.Mutex
stats ReviewStats
}
// NewPreInsertReviewer 由 cfg 与 deps 构建审查服务。Store 为 nil 时返回
// ErrNilStore(与整理器同一口径:没有记忆存储就没有审查)。
func NewPreInsertReviewer(cfg ReviewConfig, deps ReviewDeps) (*PreInsertReviewer, error) {
if deps.Store == nil {
return nil, ErrNilStore
}
cfg = normalizeReviewConfig(cfg)
now := deps.Now
if now == nil {
now = time.Now
}
return &PreInsertReviewer{
cfg: cfg,
deps: deps,
log: logger.WithSource(baseTidyLog(deps.Log), "memory/review"),
now: now,
queue: make(chan MemoryItem, cfg.QueueSize),
// 限流闸在 normalize **之后**构建:闸把非正的额度读成「恒放行」,因此
// 归一前的 ≤0 一旦直达它,「回落默认」就会静默变成「无上限」
// (design/12 §N C.5 G3 登记过的失败形态)。这里的 cfg 来自上面那行
// normalizeReviewConfig。
t1Gate: newT1RateGate(cfg.T1PerWindow, cfg.T1Window, now),
}, nil
}
// tuningSnapshot 在读锁下取一份配置快照。
//
// 为什么需要它(切片 MEMPANEL / M4):面板调优(preinsert_tuning.go 的
// `ApplyPanelTuning`)会在**运行期**改 `cfg` 的两个字段;在这条特性之前 `cfg` 是
// 构造期冻结的,读侧无锁本来安全。凡是读这两个字段的地方因此都必须与写方同步,
// 否则就是「写方持锁、读方不持锁」的数据竞争(`go test -race` 可复现)。
//
// 取**整份**快照而不是逐个字段读,是为了让一次判定看到的是**同一个版本**的配置
// (两个字段分别读可能落到两次不同的写入之间)。
//
// 它只做一次字段拷贝、**不调用任何别的函数**,因此不会引入新的锁序:写方
// (ApplyPanelTuning)持写锁时也只做字段赋值,不会再去等别的锁。
func (r *PreInsertReviewer) tuningSnapshot() ReviewConfig {
r.mu.RLock()
defer r.mu.RUnlock()
return r.cfg
}
// gateEnabled 报告**同步入库闸**是否生效(任务书规格 3 的开关)。
// 总开关关闭或闸被显式关掉时它都为 false,此时写入路径逐字回到「只插不审」。
func (r *PreInsertReviewer) gateEnabled() bool {
if r == nil {
return false
}
cfg := r.tuningSnapshot()
return cfg.Enabled && !cfg.DisableIngestGate
}
// gateScanLimit 返回同步闸实际使用的候选上限:配置值与其默认值的较小者 ——
// 写入路径上的扫描必须比异步队列更紧(任务书规格 1)。
func (r *PreInsertReviewer) gateScanLimit() int {
limit := r.cfg.GateScanLimit
if limit <= 0 {
limit = defaultReviewGateScanLimit
}
if r.cfg.ScanLimit > 0 && r.cfg.ScanLimit < limit {
return r.cfg.ScanLimit
}
return limit
}
// SessionID 返回审查运行的会话标识:**复用既有的 sess-memorytidy 装配**
// (design/23 §4.1 B;§15 D3 要求「走记忆管理会话」,不许造第二形态会话)。
func (r *PreInsertReviewer) SessionID() string { return TidySessionID }
// Enabled 报告审查是否被配置开启。
func (r *PreInsertReviewer) Enabled() bool { return r.cfg.Enabled }
// 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")
return
}
if ctx == nil {
ctx = context.Background()
}
r.mu.Lock()
if r.running {
r.mu.Unlock()
return
}
r.running = true
cctx, cancel := context.WithCancel(ctx)
r.cancel = cancel
// 两个协程(审查工作协程 + 留痕清扫循环)**在释放锁之前**一次性登记,
// 避免 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,
"queue", r.cfg.QueueSize,
"scan_limit", r.cfg.ScanLimit,
"timeout", r.cfg.Timeout.String(),
"recall_threshold", r.cfg.Threshold,
"rule_version", IngestRuleVersion,
"sync_gate", r.gateEnabled(),
"gate_budget", r.cfg.GateBudget.String(),
"gate_scan_limit", r.gateScanLimit(),
// T1 是否有模型出口:`t1_judge=false` 有两种成因(配置关掉 / 没接模型),
// 分开写让「以为开着其实没接」在启动日志里就看得见。
"t1_judge", r.t1Enabled(),
"t1_off_by_config", r.cfg.DisableT1,
"t1_timeout", r.t1Timeout().String(),
"t1_max_gray", defaultT1MaxGray,
// T1 的全局限速额度(已 normalize 的**生效值**,不是配置原文):
// 「以为不限流其实限着」与反过来的误解都在启动行上即可排除。
"t1_per_window", r.cfg.T1PerWindow,
"t1_window", r.cfg.T1Window.String(),
"sweep_interval", DefaultIngestReviewGCInterval.String(),
"retention", DefaultIngestReviewRetention.String(),
"max_reviews", DefaultIngestReviewMaxRows)
}
// Stop 停止工作协程并等待正在审的那一条跑完。幂等;ctx 在排空前被取消时返回
// 包装后的 context 错误,而不是泄漏 goroutine。
func (r *PreInsertReviewer) Stop(ctx context.Context) error {
if ctx == nil {
ctx = context.Background()
}
r.mu.Lock()
if !r.running {
r.mu.Unlock()
return nil
}
r.running = false
cancel := r.cancel
r.mu.Unlock()
cancel()
done := make(chan struct{})
go func() {
r.wg.Wait()
close(done)
}()
select {
case <-done:
r.log.Info("memory ingest review stopped", "stats", reviewStatsLabel(r.Stats()))
return nil
case <-ctx.Done():
return fmt.Errorf("memory/review: stop: %w", ctx.Err())
}
}
// Enqueue 把一条刚落库的记忆投入审查队列。它是**非阻塞**的:队列满时返回
// false(该条不做审查,记忆本身早已落库,绝不因此丢记忆)并记一条 WARN。
func (r *PreInsertReviewer) Enqueue(item MemoryItem) bool {
if !r.cfg.Enabled {
return false
}
select {
case r.queue <- item:
return true
default:
r.bump(func(s *ReviewStats) { s.Dropped++ })
r.log.Warn("memory ingest review queue full, item left unreviewed",
"item_id", item.ID, "scope", string(item.Scope), "queue", r.cfg.QueueSize)
return false
}
}
// Stats 返回累计计数(快照)。
func (r *PreInsertReviewer) Stats() ReviewStats {
r.statsMu.Lock()
defer r.statsMu.Unlock()
return r.stats
}
// worker 是单读者循环:一条一条审(单写者的自然形态,design §9 的单写者口径)。
func (r *PreInsertReviewer) worker(ctx context.Context) {
defer r.wg.Done()
for {
select {
case <-ctx.Done():
return
case item := <-r.queue:
r.process(ctx, item)
}
}
}
// process 审一条:单条超时由 r.cfg.Timeout 把守。失败**绝不改写任何一行**,
// 只如实记 WARN(design §15 D3-5)。
func (r *PreInsertReviewer) process(ctx context.Context, item MemoryItem) {
rctx, cancel := context.WithTimeout(ctx, r.cfg.Timeout)
defer cancel()
if _, err := r.Review(rctx, item); err != nil {
if errors.Is(err, ErrReviewRestrictedSource) || errors.Is(err, ErrReviewItemNotActive) {
// 只读来源 / 已不在册:都不是失败,也都零写入。
r.log.Debug("memory ingest review skipped", "item_id", item.ID, "reason", err)
return
}
r.bump(func(s *ReviewStats) { s.Failed++ })
r.log.Warn("memory ingest review failed, memory left untouched",
"item_id", item.ID, "scope", string(item.Scope), "error", err)
}
}
// Review 同步审一条并返回留痕。它在以下情形**不做任何改写**并返回错误:
// 召回不可用(超时/取消/存储错误)、改写失败。调用方据此记 WARN,
// 而记忆本身早已落库,因此「不丢记忆」是结构性成立的。
func (r *PreInsertReviewer) Review(ctx context.Context, item MemoryItem) (IngestReview, error) {
if !r.cfg.Enabled {
return IngestReview{}, ErrReviewDisabled
}
if nonOwnerSource(item.SourceConversation) {
return IngestReview{}, ErrReviewRestrictedSource
}
// 幂等:已经不在册的行不再判(否则重复判会把存活行的 merge_count 数两次)。
status, err := r.deps.Store.reviewItemStatus(ctx, item.ID)
if err != nil {
return IngestReview{}, fmt.Errorf("memory/review: item status: %w", err)
}
if status != string(tidyStatusActive) {
return IngestReview{}, ErrReviewItemNotActive
}
now := r.now().UTC()
cands, degraded, err := r.recallCandidates(ctx, item)
if err != nil {
return IngestReview{}, fmt.Errorf("memory/review: recall: %w", err)
}
if err := ctx.Err(); err != nil {
// 超时/取消发生在召回之后:同样不改写(fail-safe,§15 D3-5)。
return IngestReview{}, fmt.Errorf("memory/review: %w", err)
}
verdict := judgeIngest(item, cands)
// T1(灰区模型判定,design/23 §15.1 D3-2)**只挂在异步审查层**:规则层判不出
// 结论且有灰区候选时,用至多一次模型调用改判。未启用 / 无出口 / 失败一律
// 原样回落 create —— 它没有「改写记忆」的代码路径,因此失败绝不可能丢记忆。
verdict = r.maybeJudgeT1(ctx, item, verdict)
merged, err := r.applyVerdict(ctx, verdict, item, now)
if err != nil {
return IngestReview{}, fmt.Errorf("memory/review: apply: %w", err)
}
rev := r.buildRecord(item, verdict, merged, degraded, now)
if err := r.deps.Store.insertIngestReview(ctx, rev); err != nil {
// 留痕失败**不改判决定**:改写已经生效,这里只如实记 WARN,绝不回滚
// 一次已经生效的合并(回滚会把审查看成比结论更权威,那更危险)。
r.log.Warn("memory ingest review record insert failed", "item_id", item.ID, "error", err)
}
r.auditWith(reviewAuditVerb, rev)
r.recordDecision(rev)
return rev, nil
}
// applyVerdict 落定判定:create 什么都不做;其余把「被吸收的那一行」软合并到
// 存活行(design §8:软删除 + merged_into + merged_from + merge_count,
// 绝不物理删除,绝不改任何一行的 content)。
func (r *PreInsertReviewer) applyVerdict(ctx context.Context, v ingestVerdict, item MemoryItem, now time.Time) ([]string, error) {
if v.Decision == IngestCreate {
return nil, nil
}
if v.Decision == IngestDefer {
// 第五态(design/23 §18.1):条目**早已落库**(异步审查跑在写入之后),
// 这里把「档位 + 有效期」改成临时层(**同一条语句**写,两者不可能分叉)。
// 候选是 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
}
if v.Absorbed.ID == "" || v.Existing.ID == "" {
return nil, fmt.Errorf("memory/review: %s without a matched existing item", v.Decision)
}
survivor := item
if v.Absorbed.ID == item.ID {
// merge / skip:被吸收的是新行,存活方是既有行。
survivor = v.Existing
}
if err := r.deps.Store.absorbIngestItem(ctx, v.Absorbed, survivor, now); err != nil {
return nil, err
}
return []string{v.Absorbed.ID}, nil
}
// buildRecord 组装留痕(design §15 D3-8:决定 / 依据 / 命中的既有 id /
// 是否降级 + 规则版本与会话标识)。
func (r *PreInsertReviewer) buildRecord(item MemoryItem, v ingestVerdict, merged []string, degraded bool, now time.Time) IngestReview {
rev := IngestReview{
ID: mustNewTidyID(),
ItemID: item.ID,
PersonaID: item.PersonaID,
Scope: item.Scope,
ProjectID: item.ProjectID,
Decision: v.Decision,
Reason: v.Reason,
Veto: v.Veto,
TargetID: v.Existing.ID,
MergedIDs: merged,
Score: v.Score,
RecallMode: v.Mode,
Degraded: degraded,
RuleVer: IngestRuleVersion,
SessionID: TidySessionID,
CreatedAt: now,
}
if rev.MergedIDs == nil {
rev.MergedIDs = []string{}
}
return rev
}