package memory import ( "context" "fmt" "sort" "time" ) // 本文件是 D3「入库前审查」的**同步入库闸**(主人 2026-10-07 裁定:「把记忆整理 // 功能接到昔涟和记忆库中间做审查,由整理功能进行入库」)。 // // 与 preinsert.go 的**异步**审查(落库之后由后台队列补审)是**两层**,不是替代: // // - **闸(本文件)**:`Store.Add` 的写路径上当场判定。确定性判据命中 ⇒ 那一行 // 直接落 `status='merged'`(**照常插入、一个字不丢、可 revert**),因此 // `active` 集合**当场就是干净的**,面板/召回不再「先脏后清」; // - **异步队列(preinsert.go)**:闸没拦下的(真新 / 拿不准)照旧投入既有队列, // 由后台用**语义召回**再判一次(闸只用零依赖词项余弦,见下)。 // // 三条硬规格逐条落在这里: // // 1. **有界**:总墙钟预算 `defaultReviewGateBudget`(代码默认值)+ 候选扫描上限 // `defaultReviewGateScanLimit`(比异步侧更紧)。**超预算 / 候选源不可用 ⇒ // fail-open:照常插入 + 留痕**(绝不静默丢记忆)。🔴 **绝不把远端 embedding // 摆在同步路径上做决定**:远端 embedder 一次 2.3~3.0 s,且对同一文本返回 // 不可复现的向量(TIDY-2 实测)⇒ 闸的候选召回只用零依赖的词项余弦。 // 2. **判定以规则层确定性为主**:复用 `judgeIngest`(→ `tidy_rules.go` / // `tidy_containment.go`),**不另写第二套规则**;相似度只用于召回候选与 // 排序选存活行,绝不单凭一个分数改变落库形态。 // 3. **三态落库**:确定性判据命中 ⇒ 落 `merged`(不是「不插入」);真新 ⇒ 正常 // insert;拿不准 ⇒ 正常 insert + 投既有队列。落 `merged` 保住 `revert` // (`design/23 §8`:行永不物理删除 + 可回滚)。 const ( // defaultReviewGateBudget 是同步入库闸的**总墙钟预算**(任务书规格 1 的 // 「量级 100~300ms」)。取值依据见本文件末尾的实测登记:闸的全部工作量是 // 「一条有界 SELECT + O(n) 词项余弦 + O(n) 包含判定」,实测 p95 在**个位数 // 毫秒**量级;150 ms 给它留了约两个数量级的余量,同时远小于「一次远端 // embedding 的 2.3~3.0 s」—— 因此「超预算」只可能来自候选源本身异常, // 那时 fail-open 才是正确解。 defaultReviewGateBudget = 150 * time.Millisecond // defaultReviewGateScanLimit 是同步入库闸的候选扫描上限。它比异步队列的 // `ScanLimit`(默认 200)**更紧**:闸在写入路径上,宁可少比几条也不让 // 一次写入多花几十毫秒;且最终会与 cfg.ScanLimit 取小(见 gateScanLimit)。 defaultReviewGateScanLimit = 100 // gateAuditVerb 是同步入库闸的审计动作名(任务书「沿用既有 `memory.review.*` // 族或新增 `memory.ingest.gate`」⇒ 取后者):单独一个动作名,使「闸当场拦下 // 的」与「异步审查事后合并的」在审计链上可分。只带计数与原因码,不带正文。 gateAuditVerb = "memory.ingest.gate" // gateTraceAbsorbed 是闸真的做了决定(落 merged)的留痕原因码。 gateTraceAbsorbed = "gate:absorbed" // gateTraceDeferred 是**第五态**在闸侧做决定的留痕原因码(`expires_at` 已写)。 // 与 gateTraceAbsorbed 分开:前者写的是 status/merged_into,后者只写 // expires_at 一列 —— 「闸当场折回了多少重复」与「闸当场放进了多少临时记忆」 // 是两条不同的观测线,加成一个数会让面板读不出分型。 gateTraceDeferred = "gate:deferred" // gateTraceBudgetExceeded 是闸被墙钟预算截断的 fail-open 原因码。 gateTraceBudgetExceeded = "gate:budget_exceeded" // gateTraceUnavailable 是候选源不可用(读取失败)的 fail-open 原因码。 gateTraceUnavailable = "gate:candidates_unavailable" ) // ingestPlanAction 是入库闸对一条待落库记忆给出的落法(任务书规格 3)。 type ingestPlanAction string const ( // ingestPlanInsert 是真新 / 拿不准:正常 insert(active),并投既有异步队列。 ingestPlanInsert ingestPlanAction = "insert" // ingestPlanAbsorbNew 是确定性判据命中且**新行是子集**(skip / merge): // 照常落库,但直接落 `status='merged' + merged_into=<存活行>`,因此它**从不 // 进 active 集合**(active 计数不变)。 ingestPlanAbsorbNew ingestPlanAction = "absorb_new" // ingestPlanAbsorbOld 是确定性判据命中且**既有行是子集**(modify):新行正常 // insert(它承载更全的内容),既有行置为被合并(design §15 D3-6 的「改写」)。 ingestPlanAbsorbOld ingestPlanAction = "absorb_old" // ingestPlanDefer 是**第五态**(design/23 §18.1):真新、但不值得长期留存 ⇒ // 行**照常 active 落库**(一个字不丢),只是紧接着把 `expires_at` 写成 // 「现在 + 临时 TTL」,使它落进临时层 —— 到期由读侧时效闸隐藏 // (design/03 §9.6)、再由后台清扫物理删除(design/23 §19)。 // // 它**不投异步审查队列**(Async=false):defer 已经是这条记忆的**终局判断** // (「不值得长期留」),再送语义审查只会把它并进一条长期记忆、把临时性抹掉。 ingestPlanDefer ingestPlanAction = "defer" ) // ingestGatePlan 是同步入库闸的落库计划。零值(Action="")不存在:planIngest // 总会给出一个显式动作。 type ingestGatePlan struct { Action ingestPlanAction Decision IngestDecision // Survivor 是存活方(absorb_new / absorb_old 时非零)。 Survivor MemoryItem // Absorbed 是被吸收的那一行(仅 absorb_old 用;absorb_new 时被吸收的是 // **刚插进去的那一行**,此处为零值)。 Absorbed MemoryItem Reason string Veto TidyVetoReason Score float64 Mode string Degraded bool // Async 报告这条落库后是否仍投既有异步审查队列。只有「真新 / 拿不准 / // 闸关闭」为 true;闸挡下的那条**绝不再投**(否则同一件事两条路径)。 Async bool // Trace 是闸自己留下的原因码(空 = 闸没做决定,不留痕)。 Trace string // Elapsed 是闸在写路径上花掉的墙钟。它只进日志与留痕,**不参与任何判定** // (否则同一批数据会有两个结论,幂等性就没了)。 Elapsed time.Duration // ExpiresAt 是第五态(defer)要写进新行 `expires_at` 的时刻:**零值 = 不设 // 过期**(长期 / 日常记忆,落库为空串,四条列表读路径恒可见)。 // // 它是**时刻**而不是「TTL 秒数」:判定的那一刻就把 TTL 换算掉,落库侧因此 // 不需要认识任何配置,也不会出现「写的时候又算一次 TTL」的双口径。 ExpiresAt time.Time } // expiresAt 返回新行落库时的 `expires_at` 文本,与 `insertStatus()` / // `absorbedID()` 是同族:把「计划 → 要写进 INSERT 的值」这一步收在计划自己身上。 // // 🔴 **编码必须走 expiryText → formatTime**(定宽 30 字节 UTC),不许另写一种 // 布局:读侧时效闸是**纯文本比较**(timegate.go 的 timeGateClause,不碰 SQLite // 时间函数),布局一旦分叉就会**静默判错**(本仓已有历史事故,timegate_test.go // 钉着那条反证)。零值 ⇒ 空串 ⇒ 该行恒可见。 func (p ingestGatePlan) expiresAt() string { return expiryText(p.ExpiresAt) } // absorbedID 返回应该写进新行 `merged_into` 的存活行 id。 func (p ingestGatePlan) absorbedID() string { if p.Action == ingestPlanAbsorbNew { return p.Survivor.ID } return "" } // insertStatus 返回新行落库时的 status。 func (p ingestGatePlan) insertStatus() string { if p.Action == ingestPlanAbsorbNew { return ingestStatusMerged } return tidyStatusActive } // planIngest 在写入路径上跑同步入库闸。未接线审查(store 上没有 reviewer)、 // 总开关关闭、或闸被显式关掉时,它返回「只插不审」的 passthrough 计划 —— // 写入路径因此逐字回到本次改动之前。 func (s *Store) planIngest(ctx context.Context, item MemoryItem) ingestGatePlan { r := s.ingestReviewer() if !r.gateEnabled() { return ingestGatePlan{Action: ingestPlanInsert, Decision: IngestCreate, Async: true} } return r.gate(ctx, item) } // gate 是闸本体:在**有界**预算内召回同分区候选、用确定性规则判定,并给出 // 三态落库计划。任何异常(超预算 / 候选源不可用)都走 fail-open。 func (r *PreInsertReviewer) gate(ctx context.Context, item MemoryItem) ingestGatePlan { start := time.Now() if nonOwnerSource(item.SourceConversation) { // 群聊 / 陌生人来源:只读不写(design §6 / §15 D3-7 + AGENTS §9)。 // 既不合并、也不留痕、也不投队列(异步侧同样会拒它)—— 照常 insert。 return ingestGatePlan{Action: ingestPlanInsert, Decision: IngestCreate, Reason: "restricted_source", Elapsed: time.Since(start)} } gctx, cancel := context.WithTimeout(ctx, r.cfg.GateBudget) defer cancel() cands, err := r.gateCandidates(gctx, item) if err != nil { reason := gateTraceUnavailable if gctx.Err() != nil { reason = gateTraceBudgetExceeded } r.log.Warn("memory ingest gate failed open, item inserted unreviewed", "item_id", item.ID, "scope", string(item.Scope), "reason", reason, "error", err) return r.failOpenPlan(reason, start) } plan := r.planFromVerdict(item, judgeIngest(item, cands)) plan.Elapsed = time.Since(start) if plan.Elapsed > r.cfg.GateBudget { // 预算超了但结果已经算出来(判定在读完候选之后是纯 CPU):如实记一条 // WARN,但**不丢结论** —— 少算的只是「没来得及比的那些候选」,而那条 // 风险由既有异步队列兜住(真新照旧投队列)。 r.log.Warn("memory ingest gate exceeded its wall-clock budget", "item_id", item.ID, "elapsed_ms", plan.Elapsed.Milliseconds(), "budget_ms", r.cfg.GateBudget.Milliseconds()) } return plan } // planFromVerdict 把确定性判定映射成三态落库计划(任务书规格 3 的表格)。 func (r *PreInsertReviewer) planFromVerdict(item MemoryItem, v ingestVerdict) ingestGatePlan { plan := ingestGatePlan{ Decision: v.Decision, Reason: v.Reason, Veto: v.Veto, Score: v.Score, Mode: v.Mode, } switch v.Decision { case IngestSkip, IngestMerge: // 被吸收的是**新行**,存活方是既有行(design §15 D3-2 T0-1/T0-2)。 plan.Action = ingestPlanAbsorbNew plan.Survivor = v.Existing plan.Absorbed = item plan.Trace = gateTraceAbsorbed case IngestModify: // 既有行是新条目的子集 ⇒ 新行承载更全的内容、旧行置为被合并 // (design §15 D3-6「改写必须是新行 + 旧行置为被合并」)。 plan.Action = ingestPlanAbsorbOld plan.Survivor = item plan.Absorbed = v.Existing plan.Trace = gateTraceAbsorbed case IngestDefer: // 第五态(design/23 §18.1):行照常 active 落库(一个字不丢、可 revert), // 只把 expires_at 写成「现在 + 临时 TTL」;**不投异步队列**(defer 已是 // 终局判断,再送语义审查只会把临时性抹掉)。 plan.Action = ingestPlanDefer plan.ExpiresAt = r.deferDeadline(r.now().UTC()) plan.Trace = gateTraceDeferred default: // 真新 / 拿不准:正常 insert,并投既有异步队列(用语义召回再判一次)。 plan.Action = ingestPlanInsert plan.Async = true } return plan } // deferDeadline 把「现在」换算成第五态要写进 `expires_at` 的时刻(现在 + 临时 // TTL,design/03 §9.6)。 // // TTL 由 `cfg.TempTTL` 决定,**它已经过 normalize**(`NewPreInsertReviewer` 里 // normalizeReviewConfig 先跑),因此这里拿到的恒为正 —— 「≤0 的 TTL 让每条记忆 // 落库即过期」这条坏配置形态在构造期就被回落掉了(normalizeReviewConfig 兜底 + // ApplyReviewTuning 的 WARN)。 // // 闸(同步)与异步审查层**共用这一个函数**:同一份配置、同一种时刻语义,因此 // 同一条记忆无论被哪一层判成 defer,拿到的有效期都来自同一个算式。 func (r *PreInsertReviewer) deferDeadline(now time.Time) time.Time { return now.UTC().Add(r.cfg.TempTTL) } // failOpenPlan 组装 fail-open 计划:照常插入(active)+ 投既有队列 + 留痕。 // 「绝不静默丢记忆」不是一句承诺,而是这条路径的结构性结果。 func (r *PreInsertReviewer) failOpenPlan(reason string, start time.Time) ingestGatePlan { r.bump(func(s *ReviewStats) { s.GateFailed++ }) return ingestGatePlan{ Action: ingestPlanInsert, Decision: IngestCreate, Reason: reason, Async: true, Trace: reason, Elapsed: time.Since(start), } } // gateCandidates 是同步闸的候选召回。它与异步侧(recallCandidates)**共用** // 阈值、相似度实现(lexicalSimilarity)与包含判定,只有两点不同,且都是规格 // 要求的: // // - **绝不调用 embedder**:远端 embedding 在同步写路径上不可接受(规格 1), // 因此闸恒用零依赖的词项余弦;它不是「降级」,而是同步路径的既定形态; // - **候选收集是 O(n) 而不是 O(n²)**:异步侧复用 collectTidyPairs 会对**全部** // 两两对算相似度(含既有-既有),在写路径上白烧;闸只需要「新条目 vs 既有」 // 这 n 对(见 gatePairs,输出与 reviewPairs 逐字等价)。 func (r *PreInsertReviewer) gateCandidates(ctx context.Context, item MemoryItem) ([]ingestCandidate, error) { partition := tidyPartition{personaID: item.PersonaID, scope: item.Scope, projectID: item.ProjectID} lister := r.deps.GateCandidates if lister == nil { lister = r.deps.Store.listReviewCandidates } existing, err := lister(ctx, partition, item.ID, r.gateScanLimit()) if err != nil { return nil, fmt.Errorf("memory/review: gate candidates: %w", err) } if err := ctx.Err(); err != nil { return nil, err } r.bump(func(s *ReviewStats) { s.GateChecked++ }) return gatePairs(item, existing, r.cfg.Threshold), nil } // gatePairs 收集「新条目 vs 既有条目」的候选对:相似度 ≥ 阈值的留下,按 // 分数降序(再按候选 id)确定性排序 —— 与 collectTidyPairs 对**中心对**的 // 输出逐字等价(preinsert_gate_measure_test.go 有一条等价性断言钉住它),只是不做 // 那些与判定无关的既有-既有比较。 func gatePairs(item MemoryItem, existing []MemoryItem, threshold float64) []ingestCandidate { out := make([]ingestCandidate, 0, len(existing)) for _, ex := range existing { score := lexicalSimilarity(item.Content, ex.Content) if score < threshold { continue } out = append(out, ingestCandidate{item: ex, score: score, mode: tidyRecallLexical}) } sort.Slice(out, func(i, j int) bool { if out[i].score != out[j].score { return out[i].score > out[j].score } return out[i].item.ID < out[j].item.ID }) return out } // recordGate 为闸做过的一次决定落一条结构化留痕(design/23 §15 D3-8), // 并把它写进审计链(动作名 gateAuditVerb,**只带决定与 id,不带正文**)。 // Trace 为空表示闸没做决定(真新 / 拿不准 / 闸关闭):那时留痕由既有异步 // 队列负责,这里绝不重复落一条(否则同一件事两条路径)。 func (r *PreInsertReviewer) recordGate(ctx context.Context, item MemoryItem, plan ingestGatePlan, now time.Time) { if plan.Trace == "" { return } rev := IngestReview{ ID: mustNewTidyID(), ItemID: item.ID, PersonaID: item.PersonaID, Scope: item.Scope, ProjectID: item.ProjectID, Decision: plan.Decision, Reason: plan.Reason, Veto: plan.Veto, TargetID: plan.Survivor.ID, MergedIDs: []string{}, Score: plan.Score, RecallMode: plan.Mode, Degraded: plan.Degraded, RuleVer: IngestRuleVersion, SessionID: TidySessionID, CreatedAt: now, } if plan.Absorbed.ID != "" { rev.MergedIDs = []string{plan.Absorbed.ID} } if err := r.deps.Store.insertIngestReview(ctx, rev); err != nil { // 留痕失败**不改判决定**:改写已经生效,这里只如实记 WARN,绝不回滚 // 一次已经生效的合并。但失败原因必须带 trace,便于运维定位。 r.log.Warn("memory ingest gate record insert failed", "item_id", item.ID, "trace", plan.Trace, "error", err) } r.auditWith(gateAuditVerb, rev) } // finishIngest 落定闸给出的三态里「需要写第二笔」的部分:新行已经在 Add 里 // 插好了(absorb_new 时它已带 merged/merged_into),这里只补存活行的合并登记 // (absorb_new)或把既有行置为被合并(absorb_old)。失败**不让 Add 失败** —— // 记忆的正文早已在表里,回滚会把「已插入」变成「调用方以为没插入」而诱发重复 // 写入;这里只如实记 WARN 并把该条交给既有异步队列兜底。 func (s *Store) finishIngest(ctx context.Context, item MemoryItem, plan ingestGatePlan) { if plan.Action == ingestPlanInsert { return } if plan.Action == ingestPlanDefer { // 第五态(design/23 §18.1):新行已经在 Add 里照常插好了(active、正文 // 一字不差),这里只把 `expires_at` 补成计划给出的时刻 —— 与 // absorb_new / absorb_old 是**同一范式**:INSERT 落行、finishIngest 落 // 「那一行的派生列」。 // // 为什么不是「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 } if r := s.ingestReviewer(); r != nil { r.bump(func(st *ReviewStats) { st.GateDeferred++ }) } return } now := time.Now().UTC() var err error switch plan.Action { case ingestPlanAbsorbNew: err = s.bumpIngestSurvivor(ctx, plan.Survivor, item, now) default: err = s.absorbIngestItem(ctx, plan.Absorbed, plan.Survivor, now) } if err != nil { s.log.Warn("memory ingest gate apply failed, item left as inserted", "id", item.ID, "action", string(plan.Action), "error", err) s.enqueueReview(item) return } if r := s.ingestReviewer(); r != nil && plan.Trace == gateTraceAbsorbed { r.bump(func(st *ReviewStats) { st.GateAbsorbed++ }) } } // bumpIngestSurvivor 只刷新存活行的合并登记:merged_from 追加被吸收的 id、 // merge_count +1、confidence 取较大者、updated_at 刷新(design §8 的软动作)。 // 它是 absorbIngestItem 的「被吸收行已经落好 merged」那一半 —— 闸在插入时就 // 把新行写成了 merged,因此这里不能再 UPDATE 它一次。 func (s *Store) bumpIngestSurvivor(ctx context.Context, survivor, absorbed MemoryItem, now time.Time) error { if survivor.ID == "" || absorbed.ID == "" || survivor.ID == absorbed.ID { return fmt.Errorf("memory: bump ingest survivor: invalid pair %q -> %q", absorbed.ID, survivor.ID) } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("memory: bump ingest survivor: begin: %w", err) } defer func() { _ = tx.Rollback() }() if err := bumpSurvivor(ctx, tx, survivor, absorbed, now); err != nil { return err } if err := tx.Commit(); err != nil { return fmt.Errorf("memory: bump ingest survivor: commit: %w", err) } return nil }