按 r1 复审裁决 D:这一位先只做观测,`resolveRetention` 不许读它。理由(实测):
无条件 one_off ⇒ temp 会当天删 21 条长期约定;只放开 N6 位置判据能翻的 13 条
逐条是约定、而复审点名的三条目标行 0/3 救得了。在「模型真实标签分布」这份证据
缺位时不该做 7 天后物理删的不可逆决定。
撤接线(结论集逐字回到 M45 之前):
- retention.go:resolveRetention 不再读 item.Temporal;与之对比 dev base
(af61b8c)本文件**只有注释差异**,规则与判定入口逐字未变。
- extract_retention.go:judgeExtractionRetention 签名回退单参;删除
judgeRetentionTimepointWide 与 n6_timepoint_wide 规则码。与 dev base 对比
**同样只有注释差异** ⇒ 「结论集 = 不含 temporal 的那个集合」是结构性的,
不是靠调用点记得别传。
- 连带取消初版的 durable / recurring ⇒ long(零影响必须是双向的)。
保留观测面(一律未动):迁移 0027 的列、自举前滚、提取器的 temporal 产出、
prompt 枚举、闭集与 normalizeTemporal、store_temporal.go 的校验/编码。
另外把 prompt 里「标成 one_off 会让它进临时层、到期消失」这句**撤接线后已是假话**
改成中性标注说明 —— 否则会诱导模型避开 one_off,观测数据直接被污染。
测试:
- TestTemporalBitDoesNotAffectRetention —— 4 类语料 × 5 个取值,(档位,规则码)
与未分型逐字相同(核心正断言);
- TestTemporalOneOffDoesNotMakeAnythingTemporary / ...DoesNotRescueRuleHitRows
—— 零影响的双向点名;
- TestTemporalIsStillRecorded —— one_off 的约定行仍落 temporal='one_off' 而档位
long、expires 空(观测在、判定不动);
- TestTemporalDoesNotTouchExplicitTierOrRestrictedSource —— 人工入口与受限来源不变。
变异实测:把 `if item.Temporal == TemporalOneOff { return temp }` 插回
resolveRetention ⇒ 上述四条用例全红(FAIL 输出已贴给塔),恢复后全绿。
IngestRuleVersion 保持 d3.5,注释改准为「记录 temporal 信号,但不参与判定」
(留痕面变了:库里多了「模型当时标了什么」这一维,回退会让新旧留痕混在一起)。
P2 三条保留:store.go 500 行(拆 store_temporal.go)、temporal.go 头注('' 的口径
+ 两次被否的接线方案 + 接线前必须看的那两条统计 SQL)、报告语境段。
回归:go build ./... + go vet ./... 全绿;memory/db 包测试全绿;memory -race
38.2s 全绿。
539 lines
25 KiB
Go
539 lines
25 KiB
Go
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` 信号,但不参与判定**:提取器在产出侧多
|
||
// 标了一维时间性位(`one_off` / `recurring` / `durable` / `unknown`,迁移 0027),
|
||
// 随同一笔 INSERT 落库。**判定的结论集与 d3.4 逐字相同**(`resolveRetention`
|
||
// 不读这一位,见 retention.go 头注:复审裁决 D —— 两次候选接线都被生产重放否掉)。
|
||
// 那为什么仍升版?因为**留痕面变了**:同一对条目在 d3.4 与 d3.5 下虽然结论相同,
|
||
// 但 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
|
||
}
|