Files
Cyrene_v2/internal/runtime/memory/preinsert.go
T
AskaEth a87fbeb494 fix(memory): 面板调优引入的两处运行期写入面收口(-race 修前红 → 修后绿)
一、审查档(preinsert.go / preinsert_gate.go):面板调优会在**运行期**改 cfg 的
两个字段(DisableIngestGate / TempTTL),而读这两个字段的两处不持锁
(gateEnabled 在同步入库闸上、deferDeadline 在第五态 defer 上)。
- 读侧改走新的 tuningSnapshot()(r.mu 由 Mutex 换成 RWMutex,读锁下取整份快照);
- **只**收这两个字段的读点:其余读点读的是构造期冻结的字段,不与写方交手,
  故刻意不给入库热路径加无谓的锁;注释写明「将来面板再接管别的字段,那些读点
  必须与它一起收」;
- 锁序:写方持锁期间只做字段赋值、不取任何别的锁(normalize/回落都在取锁之前
  跑完)⇒ 不引入新锁序。

二、整理档(tidy_service_tuning.go):修掉我自己的那条真竞争 —— 原顺序是
「Stop → Start → 写 cfg」,而新起的循环**立刻**读 s.cfg 重算下一拍,与写入并排
(那一拍还可能按旧配置算出来)。改成「Stop → 写 → Start」。
顺带去掉 Stop 的超时分支:超时意味着 wg 的 Wait 仍在飞,紧接着 Start 的 Add 会
踩上 M3 在 tidy_service.go 明文写下的 WaitGroup 复用纪律(-race 下 panic);`Stop`
期间也不持 runMu,避免与「心跳正要开跑一轮」互相等死。

证据(全部逐条跑过再还原,还原用 diff 逐字校验为 0 差异):
- 审查档:撤回读侧修复 ⇒ 2 处 DATA RACE(写 preinsert_tuning.go 的两个字段 × 读
  gateEnabled / deferDeadline);恢复 ⇒ 绿。
- 整理档:把顺序改回「先起后写」⇒ DATA RACE(写 applyTidyConfig × 读 nextDelay);
  恢复 ⇒ 绿。
- 新增两条只在 -race 下有意义的回归哨兵(tidy_service_tuning_test.go):审查档真
  并发读写;整理档用**真心跳**当读者(不伪造生产不存在的调用方,那个入口在包里
  只有心跳 goroutine 会调)。

热路径成本:新增的同步是「一次无竞争 RLock + 整份小结构体快照」,隔离实测约
4ns/次(5e6 次共 22.9ms);入库闸那条测量用例的批均值在 2.62ms → 2.71~2.81ms
之间波动,落在那条用例自身的运行间噪声里(同一份代码连跑两次,纯写入基线就从
39µs 飘到 60µs)。
2026-10-13 22:11:41 +08:00

515 lines
23 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)。
const IngestRuleVersion = "d3.3"
// 调优旋钮(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):条目**早已落库**(异步审查跑在写入之后),
// 这里只把 `expires_at` 写成「现在 + 临时 TTL」把它放进临时层。
// **不删行、不改正文、不碰任何其它行** ⇒ 回滚 = 清空该列(§18.4)。
// 写不进去就如实返回错误(fail-safe:行留在长期层 = 宁可多留)。
if err := r.deps.Store.setIngestExpiry(ctx, item.ID, 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
}