diff --git a/cmd/cyrene/wire_memory_review.go b/cmd/cyrene/wire_memory_review.go index bb5b708..8cd83f2 100644 --- a/cmd/cyrene/wire_memory_review.go +++ b/cmd/cyrene/wire_memory_review.go @@ -24,6 +24,10 @@ import ( // // 关闭（MEMORY_REVIEW_ENABLED=false）时**不建、不接**：store 的写入路径因此 // 逐字回到本次改动之前（纯增量，无破坏性变更）。 +// +// 🔴 同步入库闸的开关（主人 2026-10-07 裁定）不在 env 里 —— `.env` 已冻结 +// （design/13 §2.1）：它读 `$DATA_DIR/config/memory-tidy.json` 的 +// `{"ingestGate": false}`，与整理自动档同一个文件、**零新 env 键**。 func (a *app) wireMemoryReview() error { cfg, err := config.LoadMemoryReviewConfig(envPath) if err != nil { @@ -34,6 +38,7 @@ func (a *app) wireMemoryReview() error { "memory ingest review disabled by configuration", "env", "MEMORY_REVIEW_ENABLED") return nil } + gateOff := a.memoryIngestGateOff() reviewer, err := memory.NewPreInsertReviewer(memory.ReviewConfig{ Enabled: cfg.Enabled, QueueSize: cfg.QueueSize, @@ -42,6 +47,10 @@ func (a *app) wireMemoryReview() error { // 召回门复用记忆整理的既有阈值（design/23 §13 开放点 2 = 0.80）： // 入库前审查与夜间整理必须用同一把尺子，否则同一对条目会有两个结论。 Threshold: a.cfg.MemoryTidy.RecallThreshold, + // 同步入库闸：`$DATA_DIR/config/memory-tidy.json` 的 `ingestGate: false` + // 一键回到「只插不审」（= 本次改动之前的异步行为）。预算与扫描上限走 + // 代码默认值（闸的总墙钟预算必须有界，但它是**实现参数**不是运维旋钮）。 + DisableIngestGate: gateOff, }, memory.ReviewDeps{ Store: a.memStore, Embedder: a.embedder, @@ -56,6 +65,20 @@ func (a *app) wireMemoryReview() error { return nil } +// memoryIngestGateOff 报告同步入库闸是否被配置显式关掉。读不到/解不开 +// `memory-tidy.json` 时返回 false（= 保持代码默认的「开」）并记 WARN —— +// 坏配置绝不能让一处防护静默失效，也绝不能让主人以为它关了却其实开着。 +func (a *app) memoryIngestGateOff() bool { + tuning, path, err := a.readMemoryTidyTuning() + if err != nil { + logger.WithSource(a.log, "memory/review").Warn( + "memory ingest gate tuning unreadable, falling back to code default (enabled)", + "path", path, "error", err) + return false + } + return tuning.IngestGateOff() +} + // auditMemoryReview 把一条审查结论写进网关审计链。它**只带决定、依据与 id**， // 绝不带任何记忆正文（design §15 D3-8「记忆正文不进日志」）。 func (a *app) auditMemoryReview(who, what, result string) { diff --git a/cmd/cyrene/wire_memory_tidy.go b/cmd/cyrene/wire_memory_tidy.go index 8d88e62..970bd0b 100644 --- a/cmd/cyrene/wire_memory_tidy.go +++ b/cmd/cyrene/wire_memory_tidy.go @@ -50,8 +50,7 @@ func (a *app) wireMemoryTidy() error { // // 读不到/解不开时只记 WARN 并保持 env 侧的值：那正是「回落代码默认值」。 func (a *app) applyMemoryTidyTuning(cfg *memory.TidyConfig) { - path := runtimeconfig.DefaultMemoryTidyPath(resolveDataDir(a.cfg)) - tuning, err := runtimeconfig.LoadMemoryTidyTuning(path) + tuning, path, err := a.readMemoryTidyTuning() if err != nil { logger.WithSource(a.log, "memory/tidy").Warn( "memory tidy tuning unreadable, falling back to code defaults", @@ -68,6 +67,19 @@ func (a *app) applyMemoryTidyTuning(cfg *memory.TidyConfig) { "auto_apply_per_round", cfg.AutoApplyPerRound) } +// readMemoryTidyTuning 读一次 `$DATA_DIR/config/memory-tidy.json`（**只读**）。 +// 它把「落点 + 读取 + 路径回报」收在一处，供整理自动档与**同步入库闸**两个 +// 消费者共用 —— 后者（wireMemoryReview）不属于整理服务，但它的开关就在同一个 +// 文件里（零新 env 键，design/13 §2.1）。错误由调用方记 WARN 并回落代码默认值。 +func (a *app) readMemoryTidyTuning() (runtimeconfig.MemoryTidyTuning, string, error) { + path := runtimeconfig.DefaultMemoryTidyPath(resolveDataDir(a.cfg)) + tuning, err := runtimeconfig.LoadMemoryTidyTuning(path) + if err != nil { + return runtimeconfig.MemoryTidyTuning{}, path, err + } + return tuning, path, nil +} + // auditMemoryTidy 把一轮整理的计数写进网关审计链。它**只带计数**，绝不带 // 任何记忆正文（design §8 的 who/what/when/result 口径）。 func (a *app) auditMemoryTidy(who, what, result string) { diff --git a/internal/config/runtimeconfig/memorytidy.go b/internal/config/runtimeconfig/memorytidy.go index 249f5a7..ff63fb3 100644 --- a/internal/config/runtimeconfig/memorytidy.go +++ b/internal/config/runtimeconfig/memorytidy.go @@ -8,6 +8,11 @@ // （`design/13 §2.1`：调优/门槛/上限/开关类参数走 `$DATA_DIR/config/**` + 代码 // 默认值，不进 `.env`）。于是这两项落在这里，**零新 env 键**。 // +// 2026-10-07 追加 `ingestGate`（同步入库闸，主人「把记忆整理功能接到昔涟和记忆库 +// 中间做审查，由整理功能进行入库」）：同一个理由 —— 闸是一处**改记忆落库形态**的 +// 开关，必须能一键回到「只插不审」，而 `.env` 已冻结 ⇒ 它落在同一个文件里， +// 不新开文件家族、不加 env 键。 +// // 边界与 backthink.go 同源（那是同一族的手续）： // // 1. 本文件只管**文件形状 + 读取**：它不认识 `TidyConfig`，也不认识「零值回落 @@ -47,11 +52,14 @@ const MemoryTidyConfigRelPath = memoryTidyConfigSubdir + "/" + memoryTidyFileNam // ErrInvalidMemoryTidy 是 memory-tidy.json 结构不合法时的分类错误（errors.Is 可判类）。 var ErrInvalidMemoryTidy = errors.New("runtimeconfig: invalid memory tidy config") -// MemoryTidyTuning 是 `memory-tidy.json` 的形状：记忆整理运行期的两个旋钮。 +// MemoryTidyTuning 是 `memory-tidy.json` 的形状：记忆模块运行期的旋钮。 +// 文件以「整理」命名，但它是**记忆模块运行期调优的同一处落点**（整理自动档 + +// 入库闸），不再为一个开关新开一个文件家族。 // -// AutoApply 用**指针**是刻意的：代码默认值是「开」，而 JSON 的 `bool` 零值也是 -// false —— 用 bool 会让「这个键没写」与「显式写 false」不可区分，于是任何只写 -// 上限的文件都会**静默把自动档关掉**。指针让「没写」= nil = 不覆盖默认。 +// AutoApply / IngestGate 用**指针**是刻意的：它们的代码默认值都是「开」，而 +// JSON 的 `bool` 零值也是 false —— 用 bool 会让「这个键没写」与「显式写 +// false」不可区分，于是任何只写上限的文件都会**静默把开关关掉**。指针让 +// 「没写」= nil = 不覆盖默认。 // // 字段名是 camelCase 的 JSON 键（文件是给人看、给人改的）。 type MemoryTidyTuning struct { @@ -61,6 +69,10 @@ type MemoryTidyTuning struct { // AutoApplyPerRound 是每轮自动落地的条数上限。零/负 = 没配 ⇒ 用代码默认值 // （**没有「无上限」语义**，最终还会被硬顶钳位）。 AutoApplyPerRound int `json:"autoApplyPerRound"` + // IngestGate 控制**同步入库闸**（D3 审查接在写入路径上当场拍板，主人 + // 2026-10-07 裁定）。nil = 没配（用代码默认值：开）；false = 一键回到 + // 「只插不审」（= 改动之前的异步行为）；true = 显式开（与默认同义）。 + IngestGate *bool `json:"ingestGate"` } // AutoApplyOff 报告自动档是否被这份配置**显式关掉**。nil（没配）返回 false —— @@ -69,6 +81,12 @@ func (t MemoryTidyTuning) AutoApplyOff() bool { return t.AutoApply != nil && !*t.AutoApply } +// IngestGateOff 报告**同步入库闸**是否被这份配置显式关掉（口径同 AutoApplyOff： +// nil 是「没配」⇒ 用代码默认的「开」，不是「关」）。 +func (t MemoryTidyTuning) IngestGateOff() bool { + return t.IngestGate != nil && !*t.IngestGate +} + // DefaultMemoryTidyPath 返回记忆整理调优的落点：`$DATA_DIR/config/memory-tidy.json`。 func DefaultMemoryTidyPath(dataDir string) string { return filepath.Join(dataDir, memoryTidyConfigSubdir, memoryTidyFileName) diff --git a/internal/config/runtimeconfig/memorytidy_test.go b/internal/config/runtimeconfig/memorytidy_test.go index db54510..1477776 100644 --- a/internal/config/runtimeconfig/memorytidy_test.go +++ b/internal/config/runtimeconfig/memorytidy_test.go @@ -23,7 +23,7 @@ func TestLoadMemoryTidyTuning_MissingIsNotAnError(t *testing.T) { if err != nil { t.Fatalf("缺失文件的 err = %v, want nil", err) } - if got.AutoApplyOff() || got.AutoApplyPerRound != 0 { + if got.AutoApplyOff() || got.IngestGateOff() || got.AutoApplyPerRound != 0 { t.Fatalf("缺失文件应当给零值：%+v", got) } } @@ -87,6 +87,53 @@ func TestLoadMemoryTidyTuning_BrokenIsClassifiedError(t *testing.T) { } } +// TestLoadMemoryTidyTuning_IngestGate 钉住**同步入库闸**的开关口径 +// （2026-10-07 主人裁定）：它与自动档同一个文件、同一个「指针 = 没配」语义， +// 且**零新 env 键**（`.env` 已冻结，design/13 §2.1）。 +func TestLoadMemoryTidyTuning_IngestGate(t *testing.T) { + t.Parallel() + + // ① 没写 ingestGate ⇒ 不覆盖（代码默认是「开」）。 + absent := writeMemoryTidyFile(t, `{"autoApplyPerRound": 3}`) + got, err := LoadMemoryTidyTuning(absent) + if err != nil { + t.Fatalf("LoadMemoryTidyTuning: %v", err) + } + if got.IngestGate != nil || got.IngestGateOff() { + t.Fatalf("没写 ingestGate 不得把闸关掉：%+v", got) + } + + // ② 显式 false ⇒ 一键回到「只插不审」。 + off := writeMemoryTidyFile(t, `{"ingestGate": false}`) + got, err = LoadMemoryTidyTuning(off) + if err != nil { + t.Fatalf("LoadMemoryTidyTuning(off): %v", err) + } + if !got.IngestGateOff() { + t.Fatal("显式 false 必须被读成「关掉入库闸」") + } + // 闸的开关与自动档的开关互不影响：一个键不能顺手改另一件事。 + if got.AutoApplyOff() { + t.Fatal("写 ingestGate 不得顺手把自动档也关掉") + } + + // ③ 显式 true ⇒ 与默认同义（写它只是为了自解释）。 + on := writeMemoryTidyFile(t, `{"ingestGate": true, "autoApply": false}`) + got, err = LoadMemoryTidyTuning(on) + if err != nil { + t.Fatalf("LoadMemoryTidyTuning(on): %v", err) + } + if got.IngestGateOff() || !got.AutoApplyOff() { + t.Fatalf("两个开关必须各自独立：%+v", got) + } + + // ④ 类型错 ⇒ 分类错误（调用方据此记 WARN 并回落代码默认值，绝不静默）。 + broken := writeMemoryTidyFile(t, `{"ingestGate": "nope"}`) + if _, err := LoadMemoryTidyTuning(broken); !errors.Is(err, ErrInvalidMemoryTidy) { + t.Fatalf("坏 ingestGate 的 err = %v, want ErrInvalidMemoryTidy", err) + } +} + func TestDefaultMemoryTidyPathIsDataDirConfig(t *testing.T) { t.Parallel() got := DefaultMemoryTidyPath(filepath.Join("root", "data")) diff --git a/internal/runtime/memory/preinsert.go b/internal/runtime/memory/preinsert.go index 8cff25d..1ea5dfa 100644 --- a/internal/runtime/memory/preinsert.go +++ b/internal/runtime/memory/preinsert.go @@ -102,6 +102,24 @@ type ReviewConfig struct { // 入库前审查与夜间整理必须用同一把尺子，否则同一对条目会有两个结论。 // 非正用设计稿默认值。 Threshold float64 + // DisableIngestGate 关闭**同步入库闸**（任务书规格 3 的开关）。 + // + // 它是**反向**字段，且这是刻意的：bool 的零值必须落在「默认行为」上 —— + // 同步闸本机制的默认取向（主人 2026-10-07「把记忆整理功能接到昔涟和记忆库 + // 中间做审查，由整理功能进行入库」），若写成正向的 EnableIngestGate，任何没 + // 显式配过的地方（含测试与组合根构造的一半配置）都会被读成「关掉」，与默认 + // 相反。要一键回到「只插不审」（= 本次改动之前的异步行为），写 + // `$DATA_DIR/config/memory-tidy.json` 的 `{"ingestGate": false}`（**零新 + // env 键**：`.env` 已冻结，design/13 §2.1）。 + DisableIngestGate bool + // GateBudget 是同步入库闸的**总墙钟预算**（任务书规格 1：写入路径不许变慢）。 + // 非正用代码默认值 defaultReviewGateBudget。超预算一律 fail-open（照常插入 + + // 留痕），绝不静默丢。 + GateBudget time.Duration + // GateScanLimit 是同步入库闸的候选扫描上限。它比异步队列更紧（后者 + // ScanLimit 默认 200），因为它在**写入路径**上：非正用代码默认值 + // defaultReviewGateScanLimit，且最终还会被 cfg.ScanLimit 收紧取小。 + GateScanLimit int } // ReviewDeps 是注入给入库前审查的依赖。Store 必需；其余可空并优雅降级。 @@ -118,6 +136,10 @@ type ReviewDeps struct { 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) } // IngestReview 是一条入库前审查留痕（design/23 §15 D3-8）。它**不含记忆正文**： @@ -143,17 +165,6 @@ type IngestReview struct { CreatedAt time.Time } -// ReviewStats 是审查的观测计数（进程内累计，供日志/测试断言）。 -type ReviewStats struct { - Reviewed int - Skipped int - Merged int - Modified int - Created int - Dropped int - Failed int -} - // PreInsertReviewer 是入库前审查服务。用 NewPreInsertReviewer 构建，用 Start // 拉起工作协程、Stop 停它；Review 是同步入口（测试与手动核查用）。 type PreInsertReviewer struct { @@ -206,9 +217,34 @@ func normalizeReviewConfig(cfg ReviewConfig) ReviewConfig { if cfg.Threshold <= 0 { cfg.Threshold = defaultTidyRecallThreshold } + if cfg.GateBudget <= 0 { + cfg.GateBudget = defaultReviewGateBudget + } + if cfg.GateScanLimit <= 0 { + cfg.GateScanLimit = defaultReviewGateScanLimit + } return cfg } +// gateEnabled 报告**同步入库闸**是否生效（任务书规格 3 的开关）。 +// 总开关关闭或闸被显式关掉时它都为 false，此时写入路径逐字回到「只插不审」。 +func (r *PreInsertReviewer) gateEnabled() bool { + return r != nil && r.cfg.Enabled && !r.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 } @@ -253,6 +289,9 @@ func (r *PreInsertReviewer) Start(ctx context.Context) { "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(), "sweep_interval", DefaultIngestReviewGCInterval.String(), "retention", DefaultIngestReviewRetention.String(), "max_reviews", DefaultIngestReviewMaxRows) @@ -380,7 +419,7 @@ func (r *PreInsertReviewer) Review(ctx context.Context, item MemoryItem) (Ingest // 一次已经生效的合并（回滚会把审查看成比结论更权威，那更危险）。 r.log.Warn("memory ingest review record insert failed", "item_id", item.ID, "error", err) } - r.audit(rev) + r.auditWith(reviewAuditVerb, rev) r.recordDecision(rev) return rev, nil } @@ -432,44 +471,3 @@ func (r *PreInsertReviewer) buildRecord(item MemoryItem, v ingestVerdict, merged } return rev } - -// audit 写一条审计。它**只带决定与 id，不带任何记忆正文**。 -func (r *PreInsertReviewer) audit(rev IngestReview) { - result := fmt.Sprintf("decision=%s reason=%s item=%s target=%s degraded=%t score=%.3f", - rev.Decision, rev.Reason, rev.ItemID, rev.TargetID, rev.Degraded, rev.Score) - if r.deps.Audit == nil { - r.log.Debug("memory ingest review audit (no sink configured)", "what", reviewAuditVerb, "result", result) - return - } - r.deps.Audit(reviewAuditWho, reviewAuditVerb, result) -} - -// recordDecision 累计观测计数（含降级计数，便于运维判读）。 -func (r *PreInsertReviewer) recordDecision(rev IngestReview) { - r.bump(func(s *ReviewStats) { - s.Reviewed++ - switch rev.Decision { - case IngestSkip: - s.Skipped++ - case IngestMerge: - s.Merged++ - case IngestModify: - s.Modified++ - default: - s.Created++ - } - }) -} - -// bump 在互斥下修改计数（工作协程与手动 Review 都可能调用）。 -func (r *PreInsertReviewer) bump(apply func(*ReviewStats)) { - r.statsMu.Lock() - defer r.statsMu.Unlock() - apply(&r.stats) -} - -// reviewStatsLabel 把计数渲染成一条稳定日志串。 -func reviewStatsLabel(s ReviewStats) string { - return fmt.Sprintf("reviewed=%d skip=%d merge=%d modify=%d create=%d dropped=%d failed=%d", - s.Reviewed, s.Skipped, s.Merged, s.Modified, s.Created, s.Dropped, s.Failed) -} diff --git a/internal/runtime/memory/preinsert_async_test.go b/internal/runtime/memory/preinsert_async_test.go index 2cbf39b..122f4f6 100644 --- a/internal/runtime/memory/preinsert_async_test.go +++ b/internal/runtime/memory/preinsert_async_test.go @@ -97,7 +97,11 @@ func TestReviewNeverCrossesPersonaOrScope(t *testing.T) { } // TestReviewIsNonBlockingAndAsyncPathWorks 是 design/23 §15 D3-3/D3-4： -// 写入方**绝不等待**审查；审查在后台（工作协程）按队列跑完。 +// 写入方**绝不等待**异步审查；审查在后台（工作协程）按队列跑完。 +// +// ⚠️ 2026-10-07 起 D3 是**两层**：写路径上有同步入库闸（preinsert_gate.go， +// 重复当场落 merged），闸没拦下的才进这条异步队列。因此本用例**显式关闸** +// （DisableIngestGate）来单独验异步那一层 —— 否则它测的会是闸而不是队列。 func TestReviewIsNonBlockingAndAsyncPathWorks(t *testing.T) { t.Parallel() ctx := context.Background() @@ -105,9 +109,21 @@ func TestReviewIsNonBlockingAndAsyncPathWorks(t *testing.T) { base := tidyTestTime(t) existing := seedTidyItem(t, store, reviewAt(fixtureHabitShort, base, 0.9)) - r := newTestReviewer(t, store, stubEmbedder{vecs: map[string]embedding.Vector{ - fixtureHabitShort: {1, 0}, - }}, 0.80) + asyncOnly, err := NewPreInsertReviewer(ReviewConfig{ + Enabled: true, QueueSize: 8, ScanLimit: 50, Timeout: 5 * time.Second, + Threshold: 0.80, DisableIngestGate: true, + }, ReviewDeps{ + Store: store, + Embedder: stubEmbedder{vecs: map[string]embedding.Vector{ + fixtureHabitShort: {1, 0}, + }}, + Now: func() time.Time { return tidyTestTime(t) }, + Log: testLogger(), + }) + if err != nil { + t.Fatalf("NewPreInsertReviewer: %v", err) + } + r := asyncOnly store.AttachIngestReviewer(r) // 未 Start：写入方投递后**立即返回**，审查尚未发生（异步性证据）。 start := time.Now() @@ -116,7 +132,7 @@ func TestReviewIsNonBlockingAndAsyncPathWorks(t *testing.T) { t.Fatalf("写入等了 %s，审查绝不能阻塞写入方", elapsed) } if got := readReviewRow(t, store, dup.ID); got.status != tidyStatusActive { - t.Fatalf("审查在写入方返回前就生效了，说明它是同步的：%+v", got) + t.Fatalf("关闸后审查仍在写入方返回前生效了：%+v", got) } r.Start(ctx) diff --git a/internal/runtime/memory/preinsert_gate.go b/internal/runtime/memory/preinsert_gate.go new file mode 100644 index 0000000..a6b81bb --- /dev/null +++ b/internal/runtime/memory/preinsert_gate.go @@ -0,0 +1,340 @@ +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" + // 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" +) + +// 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 +} + +// 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 + default: + // 真新 / 拿不准：正常 insert，并投既有异步队列（用语义召回再判一次）。 + plan.Action = ingestPlanInsert + plan.Async = true + } + return plan +} + +// 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 + } + 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 +} diff --git a/internal/runtime/memory/preinsert_gate_extra_test.go b/internal/runtime/memory/preinsert_gate_extra_test.go new file mode 100644 index 0000000..ba95f47 --- /dev/null +++ b/internal/runtime/memory/preinsert_gate_extra_test.go @@ -0,0 +1,130 @@ +package memory + +import ( + "context" + "strings" + "testing" + "time" +) + +// 本文件是同步入库闸（preinsert_gate.go）的两项**旁证**，与主用例集 +// （preinsert_gate_test.go 的九条）分开： +// +// - **真包含**（新 ⊆ 既有 ⇒ merge；既有 ⊆ 新 ⇒ modify）：任务书规格 3 的表格 +// 只写了三态，但确定性判据实际有 skip/merge/modify 三种命中形态，而 +// merge 与 modify 落的是**不同的两笔写**（前者新行落 merged + 存活行加登记， +// 后者新行存活 + 旧行置为被合并）。这两条路径必须各自被钉住，且都要能 revert。 +// - **审计口径**：闸的审计**只带决定与 id，绝不带记忆正文**（design §15 D3-8）。 + +// TestIngestGate_真包含_merge与modify各自落对 覆盖确定性判据命中的另两种形态。 +// 候选由测试直接注入，因此这条测的是**判定与落库**，不是召回。 +func TestIngestGate_真包含_merge与modify各自落对(t *testing.T) { + for _, tc := range []struct { + name string + existing string + incoming string + want IngestDecision + newActive bool + }{ + {"新条是子集 ⇒ merge（既有行存活）", fixtureHabitLong, fixtureHabitShort, IngestMerge, false}, + {"既有是子集 ⇒ modify（新行存活、旧行被合并）", fixtureHabitShort, fixtureHabitLong, IngestModify, true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + var existing MemoryItem + r := newGateReviewer(t, store, ReviewConfig{}, + func(context.Context, tidyPartition, string, int) ([]MemoryItem, error) { + if existing.ID == "" { + return nil, nil + } + return []MemoryItem{existing}, nil + }) + store.AttachIngestReviewer(r) + + existing = seedTidyItem(t, store, gateFixtureItem(tc.existing, base)) + incoming := seedTidyItem(t, store, gateFixtureItem(tc.incoming, base.Add(time.Minute))) + + if rev := gateRecordFor(t, store, incoming.ID); rev.Decision != tc.want { + t.Fatalf("闸判定 = %q, want %q", rev.Decision, tc.want) + } + // active 计数恒为 1：合并与改写都不该让 active 集合多出或少掉一条。 + if n := countActiveRows(t, store, "persona.a"); n != 1 { + t.Fatalf("active 计数 = %d, want 1", n) + } + survivorID, absorbedID := existing.ID, incoming.ID + if tc.newActive { + survivorID, absorbedID = incoming.ID, existing.ID + } + if got := readReviewRow(t, store, survivorID); got.status != tidyStatusActive || got.mergeCount != 2 { + t.Fatalf("存活行 = %+v, want active 且 merge_count=2", got) + } + absorbed := readReviewRow(t, store, absorbedID) + if absorbed.status != ingestStatusMerged || absorbed.mergedInto != survivorID { + t.Fatalf("被吸收行 = %+v, want merged 进 %s", absorbed, survivorID) + } + // 正文一个字不改（design §8 的可回滚口径）。 + wantContent := tc.incoming + if tc.newActive { + wantContent = tc.existing + } + if absorbed.content != wantContent { + t.Fatalf("被吸收行正文 = %q, want %q", absorbed.content, wantContent) + } + // modify 走的是一笔**独立**的写（把旧行置为被合并），因此这里额外验 + // 它同样可 revert。 + if tc.newActive { + if err := store.RevertMerged(ctx, absorbedID); err != nil { + t.Fatalf("RevertMerged: %v", err) + } + restored := readReviewRow(t, store, absorbedID) + if restored.status != tidyStatusActive || restored.mergedInto != "" || restored.content != wantContent { + t.Fatalf("revert 后 = %+v, want active 且原文无损", restored) + } + if n := countActiveRows(t, store, "persona.a"); n != 2 { + t.Fatalf("revert 后 active 计数 = %d, want 2", n) + } + } + }) + } +} + +// TestIngestGate_审计只带决定不带正文 钉住闸的审计口径（design §15 D3-8）： +// 动作名是 memory.ingest.gate（与异步审查的 memory.review.ingest 可分）， +// 内容里**绝不出现记忆正文**。 +func TestIngestGate_审计只带决定不带正文(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + var calls []string + r, err := NewPreInsertReviewer(ReviewConfig{ + Enabled: true, QueueSize: 8, ScanLimit: 50, Timeout: 5 * time.Second, Threshold: 0.80, + }, ReviewDeps{ + Store: store, + Log: testLogger(), + Audit: func(who, what, result string) { calls = append(calls, who+"|"+what+"|"+result) }, + }) + if err != nil { + t.Fatalf("NewPreInsertReviewer: %v", err) + } + store.AttachIngestReviewer(r) + + seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base)) + dup := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base.Add(time.Minute))) + + if len(calls) != 1 { + t.Fatalf("审计条数 = %d (%v), want 恰为一条（闸做过决定的那一条）", len(calls), calls) + } + got := calls[0] + if !strings.HasPrefix(got, reviewAuditWho+"|"+gateAuditVerb+"|") { + t.Fatalf("审计口径 = %q, want %s|%s|…", got, reviewAuditWho, gateAuditVerb) + } + if !strings.Contains(got, "decision=skip") || !strings.Contains(got, "item="+dup.ID) { + t.Fatalf("审计缺决定/命中 id：%q", got) + } + if strings.Contains(got, fixtureHabitShort) { + t.Fatalf("审计带了记忆正文：%q", got) + } +} diff --git a/internal/runtime/memory/preinsert_gate_measure_test.go b/internal/runtime/memory/preinsert_gate_measure_test.go new file mode 100644 index 0000000..3ac1943 --- /dev/null +++ b/internal/runtime/memory/preinsert_gate_measure_test.go @@ -0,0 +1,174 @@ +package memory + +import ( + "context" + "fmt" + "math" + "sort" + "testing" + "time" +) + +// 本文件是同步入库闸（preinsert_gate.go）的**等价性证明与墙钟实测**，从 +// preinsert_gate_test.go 拆出来只为守 AGENTS §10 的文件行数上限（≤500 行）。 +// +// 两件事各自的意义： +// - 等价性：闸的候选收集是 O(n) 的（写路径上不许白烧 O(n²)），因此必须**可证** +// 它与整理/异步侧共用的 collectTidyPairs 内核给出同一份中心对； +// - 墙钟：主人要的那个数（闸开 vs 闸关的额外墙钟 p50/p95）来自这里。 + +// TestIngestGate_gatePairs与reviewPairs等价 钉住「闸的候选收集不是第二套实现」： +// 闸用的 O(n) 收集器对**中心对**的输出必须与整理/异步侧共用的 collectTidyPairs +// 内核逐字等价（同样的阈值、同样的相似度函数、同样的排序）。 +func TestIngestGate_gatePairs与reviewPairs等价(t *testing.T) { + t.Parallel() + base := tidyTestTime(t) + contents := []string{ + fixtureHabitShort, fixtureHabitLong, fixtureHabitPlain, fixtureHabitVacation, + fixtureEnergyRedBull, fixtureEnergyIced, fixtureWorkShort, fixtureWorkLong, + fixtureOneOffEvent, + } + item := gateFixtureItem(fixtureHabitShort, base) + existing := make([]MemoryItem, 0, len(contents)-1) + for i, c := range contents[1:] { + existing = append(existing, gateFixtureItem(c, base.Add(time.Duration(i+1)*time.Minute))) + } + existing[0].ID = "id-b" + existing[1].ID = "id-a" + for i := 2; i < len(existing); i++ { + existing[i].ID = fmt.Sprintf("id-%02d", i) + } + item.ID = "id-center" + + const threshold = 0.80 + got := gatePairs(item, existing, threshold) + + items := append([]MemoryItem{item}, existing...) + sink := TidyStats{} + want := make([]ingestCandidate, 0) + for _, p := range reviewPairsAll(items, threshold, &sink) { + switch { + case p.a.ID == item.ID: + want = append(want, ingestCandidate{item: p.b, score: p.score, mode: p.mode}) + case p.b.ID == item.ID: + want = append(want, ingestCandidate{item: p.a, score: p.score, mode: p.mode}) + } + } + if len(got) != len(want) { + t.Fatalf("候选条数 = %d, want %d（got=%v want=%v）", len(got), len(want), candidateIDs(got), candidateIDs(want)) + } + for i := range got { + if got[i].item.ID != want[i].item.ID || got[i].score != want[i].score || got[i].mode != want[i].mode { + t.Fatalf("第 %d 条候选 = %+v, want %+v", i, got[i], want[i]) + } + } +} + +// reviewPairsAll 是异步侧生产口径的候选收集（reviewPairs 同源），供等价性断言 +// 取「参照实现」。它用与闸**同一个**阈值、相似度函数与内核。 +func reviewPairsAll(items []MemoryItem, threshold float64, sink *TidyStats) []tidyPair { + pairs, _ := collectTidyPairs(items, threshold, lexicalSimFunc(items), tidyRecallLexical, sink, tidyBudget{}) + return pairs +} + +// candidateIDs 取候选 id，仅用于失败信息。 +func candidateIDs(cands []ingestCandidate) []string { + out := make([]string, 0, len(cands)) + for _, c := range cands { + out = append(out, c.item.ID) + } + sort.Strings(out) + return out +} + +// TestIngestGate_Add墙钟实测_闸开vs关 给出**报告里要那个数字**：闸开与闸关时 +// `Add` 的墙钟 p50/p95（多轮取样，最坏可复现场景：每轮都扫满候选且都不合并）。 +// +// 为什么同时报「批均值」：本机计时分辨率比一次纯写入还粗（实测闸关时 +// `time.Since` 常常直接给出 0），单次 p50 因此不可分辨 —— 批均值把 200 次写入 +// 的总墙钟摊到每次上，是那条基线上唯一可信的数。 +// +// 它刻意**不并行**：并行的邻居测试会抢 CPU，把 p95 变成噪声。 +func TestIngestGate_Add墙钟实测_闸开vs关(t *testing.T) { + const ( + seeded = 100 + rounds = 200 + repeats = 3 + ) + offAll := make([]time.Duration, 0, rounds*repeats) + onAll := make([]time.Duration, 0, rounds*repeats) + var offSum, onSum time.Duration + for k := 0; k < repeats; k++ { + off, offTotal := measureAddLatency(t, false, seeded, rounds) + on, onTotal := measureAddLatency(t, true, seeded, rounds) + offAll, onAll = append(offAll, off...), append(onAll, on...) + offSum, onSum = offSum+offTotal, onSum+onTotal + t.Logf("第 %d 轮: 闸关 p50=%v p95=%v | 闸开 p50=%v p95=%v", + k+1, gatePercentile(off, 0.50), gatePercentile(off, 0.95), + gatePercentile(on, 0.50), gatePercentile(on, 0.95)) + } + offMean, onMean := offSum/time.Duration(rounds*repeats), onSum/time.Duration(rounds*repeats) + t.Logf("Add 墙钟（seeded=%d 条既有候选 / rounds=%d / 重复 %d 轮 / 最坏场景=扫满候选且不合并）", + seeded, rounds, repeats) + t.Logf(" 闸关: p50=%v p95=%v 批均值=%v", gatePercentile(offAll, 0.50), gatePercentile(offAll, 0.95), offMean) + t.Logf(" 闸开: p50=%v p95=%v 批均值=%v", gatePercentile(onAll, 0.50), gatePercentile(onAll, 0.95), onMean) + t.Logf(" 额外: p50=%v p95=%v 批均值=%v", + gatePercentile(onAll, 0.50)-gatePercentile(offAll, 0.50), + gatePercentile(onAll, 0.95)-gatePercentile(offAll, 0.95), onMean-offMean) + + // 松断言（防跑飞，不做性能回归哨兵）：闸的额外开销必须远小于预算本身。 + if extra := onMean - offMean; extra > defaultReviewGateBudget { + t.Fatalf("闸的额外批均值墙钟 = %v，超过总预算 %v", extra, defaultReviewGateBudget) + } +} + +// measureAddLatency 用同一批语料测 Add 的墙钟：gateOn=false 时不接线审查 +// （纯写入基线），true 时接线（闸开着）。第二个返回值是本轮总墙钟，供批均值用。 +func measureAddLatency(t *testing.T, gateOn bool, seeded, rounds int) ([]time.Duration, time.Duration) { + t.Helper() + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + var reviewer *PreInsertReviewer + if gateOn { + reviewer = newGateReviewer(t, store, ReviewConfig{}, nil) + store.AttachIngestReviewer(reviewer) + } + for i := 0; i < seeded; i++ { + seedTidyItem(t, store, gateTestCorpus(i, base)) + } + lat := make([]time.Duration, 0, rounds) + batchStart := time.Now() + for j := 0; j < rounds; j++ { + item := gateTestCorpus(seeded+j, base.Add(time.Duration(j+1)*time.Minute)) + start := time.Now() + if _, err := store.Add(ctx, item); err != nil { + t.Fatalf("Add: %v", err) + } + lat = append(lat, time.Since(start)) + } + total := time.Since(batchStart) + if gateOn { + if s := reviewer.Stats(); s.GateAbsorbed != 0 || s.GateChecked != seeded+rounds { + t.Fatalf("实测场景不成立（应当每轮都真扫真判且不合并）：%+v", s) + } + } + return lat, total +} + +// gatePercentile 返回 d 的 q 分位（最近秩法，确定性）。 +func gatePercentile(d []time.Duration, q float64) time.Duration { + if len(d) == 0 { + return 0 + } + sorted := append([]time.Duration(nil), d...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] }) + idx := int(math.Ceil(q*float64(len(sorted)))) - 1 + if idx < 0 { + idx = 0 + } + if idx >= len(sorted) { + idx = len(sorted) - 1 + } + return sorted[idx] +} diff --git a/internal/runtime/memory/preinsert_gate_test.go b/internal/runtime/memory/preinsert_gate_test.go new file mode 100644 index 0000000..5b729f0 --- /dev/null +++ b/internal/runtime/memory/preinsert_gate_test.go @@ -0,0 +1,442 @@ +package memory + +import ( + "context" + "errors" + "fmt" + "testing" + "time" +) + +// 本文件是**同步入库闸**（主人 2026-10-07 裁定「把记忆整理功能接到昔涟和记忆库 +// 中间做审查，由整理功能进行入库」）的验收用例集：三态落库、确定性判据为主、 +// 红线不误判、fail-open 不丢记忆、开关一键回退、revert 无损、墙钟有界。 +// +// 它与 preinsert_test.go / preinsert_async_test.go 的分工：那两份测的是**异步** +// 那一层（落库之后由后台队列补审），本文件测的是**写路径上**的那一层。 +// 闸开着时，重复行在 `Store.Add` 返回之前就已经是 `merged` —— 因此「异步队列 +// 才把它折回」的旧断言必须改成「闸当场折回」（见 preinsert_async_test.go 的 +// TestReviewIsNonBlockingAndAsyncPathWorks，它现在显式关闸来单独验异步层）。 + +// gateFixtureItem 构造同分区（persona.a / persona / 无项目）、owner 来源的待落库条目。 +func gateFixtureItem(content string, at time.Time) MemoryItem { + return habitItem(content, at, 0.9) +} + +// gateTestCorpus 生成「同主题、数值各异」的语料：它们互相的相似度很高（因此 +// 候选扫描打满），但数值不同 ⇒ 数值否决 ⇒ 恒为 create。墙钟实测用它构造 +// **最坏可复现场景**：每一轮都真扫满候选、真判完，且一次都不合并。 +func gateTestCorpus(i int, at time.Time) MemoryItem { + return gateFixtureItem(fmt.Sprintf("主人常在第 %d 个时段与助手聊天，有熬夜习惯", i), at) +} + +// newGateReviewer 构建「闸开着」的审查服务。candidates 非 nil 时注入候选源 +// （测试用它精确控制候选集、注入慢源与坏源）。零值配置项按代码默认值补齐 —— +// **尤其是 GateBudget**：它留零值才能让「默认预算」这条被测试覆盖到。 +func newGateReviewer(t *testing.T, store *Store, cfg ReviewConfig, + candidates func(context.Context, tidyPartition, string, int) ([]MemoryItem, error)) *PreInsertReviewer { + t.Helper() + cfg.Enabled = true + if cfg.QueueSize == 0 { + cfg.QueueSize = 8 + } + if cfg.ScanLimit == 0 { + cfg.ScanLimit = 50 + } + if cfg.Timeout == 0 { + cfg.Timeout = 5 * time.Second + } + if cfg.Threshold == 0 { + cfg.Threshold = 0.80 + } + r, err := NewPreInsertReviewer(cfg, ReviewDeps{ + Store: store, + Now: func() time.Time { return tidyTestTime(t) }, + GateCandidates: candidates, + Log: testLogger(), + }) + if err != nil { + t.Fatalf("NewPreInsertReviewer: %v", err) + } + return r +} + +// gateRecordFor 读回某一条记忆的闸留痕（按 item_id 过滤）。 +func gateRecordFor(t *testing.T, store *Store, itemID string) IngestReview { + t.Helper() + reviews, err := store.ListIngestReviews(context.Background(), 50) + if err != nil { + t.Fatalf("ListIngestReviews: %v", err) + } + for _, rev := range reviews { + if rev.ItemID == itemID { + return rev + } + } + return IngestReview{} +} + +// TestIngestGate_字面重复_新行落merged且active不变 是任务书必备用例①： +// 字面重复必须**当场**落 `merged`，active 计数不变、存活行 merge_count +1， +// 且正文一个字不丢（保住 revert）。 +func TestIngestGate_字面重复_新行落merged且active不变(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{}, nil) + store.AttachIngestReviewer(r) + + survivor := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base)) + dup := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base.Add(time.Minute))) + + got := readReviewRow(t, store, dup.ID) + if got.status != ingestStatusMerged || got.mergedInto != survivor.ID { + t.Fatalf("重复行未被闸当场折回：status=%q merged_into=%q，want merged/%s", got.status, got.mergedInto, survivor.ID) + } + if got.content != fixtureHabitShort { + t.Fatalf("闸下的行正文被改了：%q", got.content) + } + if n := countActiveRows(t, store, "persona.a"); n != 1 { + t.Fatalf("active 计数 = %d, want 1（闸的意义就是 active 当场干净）", n) + } + if s := readReviewRow(t, store, survivor.ID); s.mergeCount != 2 { + t.Fatalf("存活行 merge_count = %d, want 2", s.mergeCount) + } + + rev := gateRecordFor(t, store, dup.ID) + if rev.Decision != IngestSkip || rev.Reason != "exact_duplicate" || rev.TargetID != survivor.ID { + t.Fatalf("闸留痕 = %+v, want skip/exact_duplicate 指向存活行", rev) + } + if rev.SessionID != TidySessionID || rev.RuleVer != IngestRuleVersion { + t.Fatalf("闸留痕的会话/规则版本没登记：%+v", rev) + } + if rev.RecallMode != tidyRecallLexical { + t.Fatalf("闸留痕的召回来源 = %q, want %q（同步路径绝不调远端 embedding）", rev.RecallMode, tidyRecallLexical) + } + if s := r.Stats(); s.GateAbsorbed != 1 || s.GateChecked != 2 { + // 2 = 第一条（无候选 ⇒ 正常插入）+ 重复条（命中 ⇒ 吸收）。 + t.Fatalf("闸计数 = %+v, want gate_checked=2 gate_absorbed=1", s) + } + // 闸挡下的那条**绝不再投异步队列**（否则同一件事两条路径）：队列里只该有 + // 第一条（它没有候选 ⇒ 正常插入，照旧走既有队列）。 + if n := len(r.queue); n != 1 { + t.Fatalf("队列 = %d 条，want 1（重复的那条不该被再投一次）", n) + } + if err := ctx.Err(); err != nil { + t.Fatalf("ctx: %v", err) + } +} + +// TestIngestGate_归一化等价_同样落merged 是任务书必备用例②：大小写/空白归一化 +// 后等价的文本（dedupeKey 相等）必须和字面重复同一处理。 +func TestIngestGate_归一化等价_同样落merged(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{}, nil) + store.AttachIngestReviewer(r) + + survivor := seedTidyItem(t, store, gateFixtureItem("主人爱喝 ICED 美式", base)) + dup := seedTidyItem(t, store, gateFixtureItem(" 主人爱喝 iced 美式 ", base.Add(time.Minute))) + + got := readReviewRow(t, store, dup.ID) + if got.status != ingestStatusMerged || got.mergedInto != survivor.ID { + t.Fatalf("归一化等价未被闸折回：%+v（存活行 %s）", got, survivor.ID) + } + if n := countActiveRows(t, store, "persona.a"); n != 1 { + t.Fatalf("active 计数 = %d, want 1", n) + } + if s := readReviewRow(t, store, survivor.ID); s.mergeCount != 2 { + t.Fatalf("存活行 merge_count = %d, want 2", s.mergeCount) + } +} + +// TestIngestGate_真新_正常insert并走既有队列 是任务书必备用例③：确实新的条目 +// 正常 insert（active），并且**照旧投既有异步队列**（两层而不是替代）。 +func TestIngestGate_真新_正常insert并走既有队列(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{}, nil) + store.AttachIngestReviewer(r) + + first := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base)) + second := seedTidyItem(t, store, gateFixtureItem(fixtureEnergyRedBull, base.Add(time.Minute))) + + for _, it := range []MemoryItem{first, second} { + got := readReviewRow(t, store, it.ID) + if got.status != tidyStatusActive || got.mergedInto != "" { + t.Fatalf("真新条目被改了落库形态：%+v", got) + } + } + if n := countActiveRows(t, store, "persona.a"); n != 2 { + t.Fatalf("active 计数 = %d, want 2", n) + } + if s := r.Stats(); s.GateAbsorbed != 0 || s.GateChecked != 2 { + t.Fatalf("闸计数 = %+v, want 只检查、不吸收", s) + } + if n := len(r.queue); n != 2 { + t.Fatalf("真新条目投队列 = %d, want 2（闸与异步队列是两层）", n) + } + if revs, err := store.ListIngestReviews(context.Background(), 10); err != nil || len(revs) != 0 { + t.Fatalf("闸对「真新」不该留决定留痕（那段由异步队列负责）：%+v err=%v", revs, err) + } +} + +// TestIngestGate_红线_习惯与一次性事件绝不误合 是任务书必备用例④与 +// design/23 §5.2 的红线：时效/一次性事件的条目即便高度相似也**绝不合并** +// （候选直接喂进判定层，因此这条测的是规则层本身，不是召回）。 +func TestIngestGate_红线_习惯与一次性事件绝不误合(t *testing.T) { + for _, tc := range []struct { + name string + candidate string + wantVeto TidyVetoReason + }{ + {"一次性事件", fixtureOneOffEvent, VetoTemporal}, + {"有期限状态", fixtureHabitVacation, VetoTemporal}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + cand := gateFixtureItem(tc.candidate, base) + r := newGateReviewer(t, store, ReviewConfig{}, + func(context.Context, tidyPartition, string, int) ([]MemoryItem, error) { + return []MemoryItem{cand}, nil + }) + store.AttachIngestReviewer(r) + + got := seedTidyItem(t, store, gateFixtureItem(fixtureHabitPlain, base.Add(time.Minute))) + row := readReviewRow(t, store, got.ID) + if row.status != tidyStatusActive || row.mergedInto != "" { + t.Fatalf("%s 被误合了：%+v", tc.name, row) + } + if s := r.Stats(); s.GateAbsorbed != 0 { + t.Fatalf("%s 被闸吸收：%+v", tc.name, s) + } + }) + } +} + +// TestIngestGate_跨分区绝不合并 是任务书必备用例⑤：分区键（persona_id + +// scope + project_id）是硬门 —— 运行期 SQL 就把别的分区隔断，判定层再兜一次。 +func TestIngestGate_跨分区绝不合并(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{}, nil) + store.AttachIngestReviewer(r) + + otherPersona := gateFixtureItem(fixtureHabitShort, base) + otherPersona.PersonaID = "persona.b" + seedTidyItem(t, store, otherPersona) + otherScope := gateFixtureItem(fixtureHabitShort, base) + otherScope.Scope = ScopeUser + otherScopeItem := seedTidyItem(t, store, otherScope) + otherProject := gateFixtureItem(fixtureHabitShort, base) + otherProject.Scope = ScopeProject + otherProject.ProjectID = "proj.other" + otherProjectItem := seedTidyItem(t, store, otherProject) + + mine := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base.Add(time.Minute))) + row := readReviewRow(t, store, mine.ID) + if row.status != tidyStatusActive || row.mergedInto != "" { + t.Fatalf("跨分区被合并了：%+v", row) + } + for _, id := range []string{otherScopeItem.ID, otherProjectItem.ID} { + if got := readReviewRow(t, store, id); got.status != tidyStatusActive { + t.Fatalf("别的分区的记忆被改写了：%+v", got) + } + } + + // 纵深防御：把跨角色候选直接喂进判定层，硬门必须否决。 + verdict := judgeIngest(mine, []ingestCandidate{{item: otherPersona, score: 0.99, mode: tidyRecallLexical}}) + if verdict.Decision != IngestCreate || verdict.Veto != VetoPersona { + t.Fatalf("跨角色判定 = %+v, want create + persona 否决", verdict) + } +} + +// TestIngestGate_failOpen_候选源不可用绝不丢记忆 是任务书必备用例⑥（其一）： +// 候选源报错 ⇒ **照常插入 active** + 留痕（原因码），绝不静默丢。 +func TestIngestGate_failOpen_候选源不可用绝不丢记忆(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{}, + func(context.Context, tidyPartition, string, int) ([]MemoryItem, error) { + return nil, errors.New("candidate source down") + }) + store.AttachIngestReviewer(r) + + item := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base)) + row := readReviewRow(t, store, item.ID) + if row.status != tidyStatusActive || row.content != fixtureHabitShort { + t.Fatalf("fail-open 丢了记忆：%+v", row) + } + rev := gateRecordFor(t, store, item.ID) + if rev.Reason != gateTraceUnavailable || rev.Decision != IngestCreate { + t.Fatalf("fail-open 留痕 = %+v, want reason=%s", rev, gateTraceUnavailable) + } + if s := r.Stats(); s.GateFailed != 1 { + t.Fatalf("fail-open 计数 = %+v, want gate_failed=1", s) + } + if n := len(r.queue); n != 1 { + t.Fatalf("fail-open 的条目没投既有队列（%d 条）：兜底那一层断了", n) + } +} + +// TestIngestGate_failOpen_超预算绝不丢记忆 是任务书必备用例⑥（其二）： +// 候选源慢到越过闸的总预算 ⇒ 同样是照常插入 + 留痕，且**绝不超预算返回**。 +func TestIngestGate_failOpen_超预算绝不丢记忆(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{}, gateSlowCandidates(2*time.Second)) + store.AttachIngestReviewer(r) + + start := time.Now() + item := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base)) + elapsed := time.Since(start) + + row := readReviewRow(t, store, item.ID) + if row.status != tidyStatusActive || row.content != fixtureHabitShort { + t.Fatalf("超预算时丢了记忆：%+v", row) + } + rev := gateRecordFor(t, store, item.ID) + if rev.Reason != gateTraceBudgetExceeded { + t.Fatalf("超预算留痕 = %+v, want reason=%s", rev, gateTraceBudgetExceeded) + } + if elapsed > time.Second { + t.Fatalf("超预算的写入等了 %s，闸的预算没起作用", elapsed) + } +} + +// TestIngestGate_开关关闭_回到只插不审 是任务书必备用例⑦：闸被显式关掉（或总 +// 开关关闭）时，写入路径逐字回到改动之前 —— 正常 insert + 投既有异步队列。 +func TestIngestGate_开关关闭_回到只插不审(t *testing.T) { + for _, tc := range []struct { + name string + enabled bool + disableGate bool + enqueue bool + }{ + {"闸关（memory-tidy.json 的 ingestGate=false）", true, true, true}, + {"总开关关（MEMORY_REVIEW_ENABLED=false）", false, false, false}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + // newGateReviewer 构建的是「闸开着」的服务，因此总开关关闭那一支 + // 直接用构造函数建关闭态（它有独立的语义：不建、不接、不跑）。 + var r *PreInsertReviewer + if tc.enabled { + r = newGateReviewer(t, store, ReviewConfig{ + Enabled: tc.enabled, DisableIngestGate: tc.disableGate, + }, nil) + } else { + var err error + r, err = NewPreInsertReviewer(ReviewConfig{Enabled: false}, ReviewDeps{Store: store, Log: testLogger()}) + if err != nil { + t.Fatalf("NewPreInsertReviewer: %v", err) + } + } + store.AttachIngestReviewer(r) + + seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base)) + dup := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base.Add(time.Minute))) + + row := readReviewRow(t, store, dup.ID) + if row.status != tidyStatusActive || row.mergedInto != "" { + t.Fatalf("闸关闭时仍改了落库形态：%+v", row) + } + if n := countActiveRows(t, store, "persona.a"); n != 2 { + t.Fatalf("闸关闭时 active 计数 = %d, want 2", n) + } + if s := r.Stats(); s.GateChecked != 0 || s.GateAbsorbed != 0 { + t.Fatalf("闸关闭时闸仍跑了：%+v", s) + } + if revs, err := store.ListIngestReviews(context.Background(), 10); err != nil || len(revs) != 0 { + t.Fatalf("闸关闭时不该有留痕：%+v err=%v", revs, err) + } + wantQueued := 0 + if tc.enqueue { + wantQueued = 2 + } + if n := len(r.queue); n != wantQueued { + t.Fatalf("投队列 = %d, want %d（关闸即回到改动之前的异步行为）", n, wantQueued) + } + }) + } +} + +// TestIngestGate_revert_能把闸下落的merged行复原 是任务书必备用例⑧： +// 闸落的是 `merged`（不是「不插入」）⇒ revert 无损复原。 +func TestIngestGate_revert_能把闸下落的merged行复原(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{}, nil) + store.AttachIngestReviewer(r) + + survivor := seedTidyItem(t, store, gateFixtureItem(fixtureHabitLong, base)) + dup := seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base.Add(time.Minute))) + if got := readReviewRow(t, store, dup.ID); got.status != ingestStatusMerged { + t.Fatalf("前置条件不成立（闸没折回）：%+v", got) + } + + if err := store.RevertMerged(ctx, dup.ID); err != nil { + t.Fatalf("RevertMerged: %v", err) + } + restored := readReviewRow(t, store, dup.ID) + if restored.status != tidyStatusActive || restored.mergedInto != "" || restored.content != fixtureHabitShort { + t.Fatalf("revert 后 = %+v, want active 且原文无损", restored) + } + if s := readReviewRow(t, store, survivor.ID); s.mergeCount != 1 { + t.Fatalf("revert 后存活行 merge_count = %d, want 1（登记也一并摘掉）", s.mergeCount) + } + if n := countActiveRows(t, store, "persona.a"); n != 2 { + t.Fatalf("revert 后 active 计数 = %d, want 2", n) + } +} + +// TestIngestGate_Add墙钟有界 是任务书必备用例⑨：候选源慢过预算时，`Add` 的 +// 墙钟仍由闸的预算封顶（注入慢审查器 ⇒ 断言不超预算）。 +// +// 它**刻意不设 GateBudget**，走代码默认值 —— 这样「去掉预算」这条变异会把 +// 它变红（见报告里的变异自证）。 +func TestIngestGate_Add墙钟有界(t *testing.T) { + t.Parallel() + store := newTestStore(t) + base := tidyTestTime(t) + r := newGateReviewer(t, store, ReviewConfig{}, gateSlowCandidates(2*time.Second)) + store.AttachIngestReviewer(r) + + start := time.Now() + seedTidyItem(t, store, gateFixtureItem(fixtureHabitShort, base)) + elapsed := time.Since(start) + + budget := defaultReviewGateBudget + if elapsed > budget+500*time.Millisecond { + t.Fatalf("Add 等了 %s，越过闸预算 %s 太多：闸没有界", elapsed, budget) + } + if s := r.Stats(); s.GateFailed != 1 || s.GateChecked != 0 { + t.Fatalf("闸计数 = %+v, want 一条 fail-open", s) + } +} + +// gateSlowCandidates 返回一个「慢候选源」：它一直等到 ctx 取消或 d 到点。 +// 它**尊重 ctx**（生产里的候选源是 SQL 查询，同样尊重 ctx），因此预算真的 +// 能把它截断；这是「预算有界」可确定性验证的前提。 +func gateSlowCandidates(d time.Duration) func(context.Context, tidyPartition, string, int) ([]MemoryItem, error) { + return func(ctx context.Context, _ tidyPartition, _ string, _ int) ([]MemoryItem, error) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(d): + return nil, nil + } + } +} diff --git a/internal/runtime/memory/preinsert_stats.go b/internal/runtime/memory/preinsert_stats.go new file mode 100644 index 0000000..899787a --- /dev/null +++ b/internal/runtime/memory/preinsert_stats.go @@ -0,0 +1,74 @@ +package memory + +import "fmt" + +// 本文件是入库前审查（design/23 §15.1 D3）的**观测面**：进程内计数与审计出口。 +// 它从 preinsert.go 拆出来只为守 AGENTS §10 的文件行数上限（≤500 行）与 +// 「模块按职责拆源文件」那条规范 —— 口径逐字未变。 +// +// 为什么要**分列**闸与异步队列的计数：同步入库闸（写路径上当场拍板）与异步 +// 审查队列（后台补审）是两件事，把它们加成一个数会让「闸拦下了多少重复」与 +// 「事后补审了多少」不可分辨。分列之后，面板/日志能分别读到两条路径的量。 + +// ReviewStats 是审查的观测计数（进程内累计，供日志/测试断言）。 +type ReviewStats struct { + Reviewed int + Skipped int + Merged int + Modified int + Created int + Dropped int + Failed int + // GateChecked 是**同步入库闸**实际做了一次判定的条数（跳过闸的条目不计）。 + GateChecked int + // GateAbsorbed 是闸当场拦下并落成 `merged` 的条数（真重复，含 modify 的 + // 「旧行被吸收」）。 + GateAbsorbed int + // GateFailed 是闸因预算/候选源不可用而 fail-open 的次数（照常插入 + 留痕）。 + GateFailed int +} + +// bump 在互斥下修改计数（工作协程、手动 Review 与同步闸都可能调用）。 +func (r *PreInsertReviewer) bump(apply func(*ReviewStats)) { + r.statsMu.Lock() + defer r.statsMu.Unlock() + apply(&r.stats) +} + +// recordDecision 累计观测计数（含降级计数，便于运维判读）。 +func (r *PreInsertReviewer) recordDecision(rev IngestReview) { + r.bump(func(s *ReviewStats) { + s.Reviewed++ + switch rev.Decision { + case IngestSkip: + s.Skipped++ + case IngestMerge: + s.Merged++ + case IngestModify: + s.Modified++ + default: + s.Created++ + } + }) +} + +// auditWith 写一条审计。verb 让「同步闸当场拦下的」与「异步审查事后合并的」 +// 在审计链上可分（前者 memory.ingest.gate，后者 memory.review.ingest）。 +// 它**只带决定与 id，绝不带任何记忆正文**。 +func (r *PreInsertReviewer) auditWith(verb string, rev IngestReview) { + result := fmt.Sprintf("decision=%s reason=%s item=%s target=%s degraded=%t score=%.3f", + rev.Decision, rev.Reason, rev.ItemID, rev.TargetID, rev.Degraded, rev.Score) + if r.deps.Audit == nil { + r.log.Debug("memory ingest review audit (no sink configured)", "what", verb, "result", result) + return + } + r.deps.Audit(reviewAuditWho, verb, result) +} + +// reviewStatsLabel 把计数渲染成一条稳定日志串。 +func reviewStatsLabel(s ReviewStats) string { + return fmt.Sprintf("reviewed=%d skip=%d merge=%d modify=%d create=%d dropped=%d failed=%d "+ + "gate_checked=%d gate_absorbed=%d gate_failed=%d", + s.Reviewed, s.Skipped, s.Merged, s.Modified, s.Created, s.Dropped, s.Failed, + s.GateChecked, s.GateAbsorbed, s.GateFailed) +} diff --git a/internal/runtime/memory/store.go b/internal/runtime/memory/store.go index 78f821f..b58ef68 100644 --- a/internal/runtime/memory/store.go +++ b/internal/runtime/memory/store.go @@ -163,19 +163,14 @@ type Store struct { proj *projectindex.Store // reviewMu 保护 review，使入库前审查（design/23 §15.1 D3）能在装配后 - // 并发接入：接入后每条经 Add 落库的条目都会被投入审查队列，由独立的 - // 记忆管理会话判定 merge/modify/create/skip。review 为 nil 表示未启用 - // 或未接线，此时写入路径行为逐字与接入前一致。 + // 并发接入：接入后每条经 Add 落库的条目都会**先过同步入库闸**（确定性判据 + // 命中即落 merged，见 preinsert_gate.go），闸没拦下的再投入异步审查队列， + // 由独立的记忆管理会话判定 merge/modify/create/skip。review 为 nil 表示 + // 未启用或未接线，此时写入路径行为逐字与接入前一致。 reviewMu sync.RWMutex review *PreInsertReviewer } -// rowScanner 同时被 *sql.Row 与 *sql.Rows 满足，因此单个 scanItem -// 辅助函数即可服务 Get 与 ListByPersona。 -type rowScanner interface { - Scan(dest ...any) error -} - // NewStore 在 db 上构建 Store，确保 memory_items schema 存在。 // // db 必须是存活且已迁移的连接（迁移框架本身由 @@ -220,12 +215,22 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) // Scope/Kind/Confidence/DecayWeight 规范化到默认值。它返回已落库的 // 条目，id 与时间戳均已填充。 // +// 🔴 **同步入库闸**（主人 2026-10-07 裁定）：本函数是全仓**唯一**写路径，因此 +// 闸就摆在这里 —— 落库**之前**先在同分区内跑一次确定性判定（复用 +// judgeIngest → tidy_rules.go / tidy_containment.go），判出「真重复」时这一行 +// 直接落 status='merged' + merged_into=<存活行>（**照常插入、一个字不丢、可 +// revert、从不进 active 集合**），于是 active 集合当场就是干净的。闸的总墙钟 +// 预算显式且有界（见 defaultReviewGateBudget），超预算/候选源不可用一律 +// fail-open（照常插入 + 留痕，**绝不静默丢记忆**）。闸未接线或被开关关掉时， +// 这段逻辑是零成本的 passthrough，写入路径逐字回到改动之前。 +// // 接入持久语义索引（见 attachVectorIndex）时，落库后同一条目也会 // 被向量化并写入索引，使索引与 memory_items 同步 —— 语义检索路径因此 // 真的有向量可查，而不是对一个恒空索引搜索（design/03 §3「真正启用 // 向量检索」、design/04 §7）。索引写入失败只记日志：条目已经在 // 关系表里，语义增强的失败绝不使记忆写入失败，检索侧也会回退到 -// 内存余弦（见 Retriever.semanticScores）。 +// 内存余弦（见 Retriever.semanticScores）。被闸判为重复的行**不建索引**： +// 它不进召回、也不该占掉语义检索 top-K 的一个位置。 func (s *Store) Add(ctx context.Context, item MemoryItem) (MemoryItem, error) { if err := item.normalize(); err != nil { return MemoryItem{}, err @@ -239,22 +244,33 @@ func (s *Store) Add(ctx context.Context, item MemoryItem) (MemoryItem, error) { return MemoryItem{}, fmt.Errorf("memory: encode tags: %w", err) } + plan := s.planIngest(ctx, item) _, 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) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + source_conversation, created_at, updated_at, decay_weight, status, merged_into) + 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) + formatTime(item.CreatedAt), formatTime(item.UpdatedAt), item.DecayWeight, + plan.insertStatus(), plan.absorbedID()) if err != nil { return MemoryItem{}, fmt.Errorf("memory: insert item: %w", err) } - s.indexItem(ctx, item) - s.ensureProjectIndex(ctx, item) - // 入库前审查（design/23 §15.1 D3）：唯一写入路径的这一处把刚落库的条目 - // 投入审查队列。投递是非阻塞的 —— 审查跑在独立的记忆管理会话里，绝不 - // 让写入方（回合收尾）等它。未接线审查时是空操作。 - s.enqueueReview(item) + s.finishIngest(ctx, item, plan) + if plan.Action != ingestPlanAbsorbNew { + s.indexItem(ctx, item) + s.ensureProjectIndex(ctx, item) + } + if r := s.ingestReviewer(); r != nil { + r.recordGate(ctx, item, plan, time.Now().UTC()) + } + // 入库前审查（design/23 §15.1 D3）的**异步**那一层：闸没拦下的（真新 / + // 拿不准 / 闸关闭）照旧投入审查队列，由后台用语义召回再判一次；闸挡下的 + // 那条**绝不再投**（同一件事不该有两条路径）。投递是非阻塞的。未接线审查 + // 时是空操作。 + if plan.Async { + s.enqueueReview(item) + } return item, nil } @@ -324,26 +340,6 @@ func (s *Store) List(ctx context.Context) ([]MemoryItem, error) { return scanItems(rows, "item") } -// scanItems 把每一行读入 MemoryItem 切片，关闭 rows 并用 -// 操作名包裹 scan/遍历失败。成功时总是返回 -// 非 nil、可能为空的切片。 -func scanItems(rows *sql.Rows, op string) ([]MemoryItem, error) { - defer func() { _ = rows.Close() }() - - items := make([]MemoryItem, 0) - for rows.Next() { - item, err := scanItem(rows) - if err != nil { - return nil, fmt.Errorf("memory: scan %s: %w", op, err) - } - items = append(items, item) - } - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("memory: iterate %s: %w", op, err) - } - return items, nil -} - // Delete 删除给定 id 的条目，id 为空或 // 没有删除任何行时返回 ErrNotFound。调用方只应传入经 // 角色作用域读取（ListByPersona/Get）得到的 id，使删除始终留在 @@ -440,40 +436,6 @@ func applyTimestamps(item MemoryItem) (MemoryItem, error) { return item, nil } -// scanItem 读取一行 memory_items。tags 以 JSON 数组字符串存储； -// 时间戳以定宽 UTC 字符串（timeLayout）存储，因此文本 -// 比较即可按时间排序。 -func scanItem(sc rowScanner) (MemoryItem, error) { - var ( - it MemoryItem - scope, kind 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 { - return MemoryItem{}, err - } - it.Scope = Scope(scope) - it.Kind = Kind(kind) - if err := json.Unmarshal([]byte(tagsJSON), &it.Tags); err != nil { - return MemoryItem{}, fmt.Errorf("memory: decode tags: %w", err) - } - if it.Tags == nil { - it.Tags = []string{} - } - - var err error - if it.CreatedAt, err = parseTime(createdAt); err != nil { - return MemoryItem{}, err - } - if it.UpdatedAt, err = parseTime(updatedAt); err != nil { - return MemoryItem{}, err - } - return it, nil -} - // newID 返回一个 128 位密码学随机 id，十六进制编码。它避开 // UUID 依赖，使本包保持自包含。rand.Read 只在 // 平台熵源损坏时失败，因此错误向上传播。 @@ -492,5 +454,3 @@ func discardLogger(log *slog.Logger) *slog.Logger { } return slog.New(slog.NewTextHandler(io.Discard, nil)) } - - diff --git a/internal/runtime/memory/store_rows.go b/internal/runtime/memory/store_rows.go new file mode 100644 index 0000000..b507f09 --- /dev/null +++ b/internal/runtime/memory/store_rows.go @@ -0,0 +1,74 @@ +package memory + +import ( + "database/sql" + "encoding/json" + "fmt" +) + +// 本文件是 memory_items 的**行映射**（scanItem / scanItems 与它们的公共接口 +// rowScanner），从 store.go 拆出来只为守 AGENTS §10 的文件行数上限（≤500 行） +// 与「模块按职责拆源文件」那条规范 —— 口径逐字未变。 +// +// 放在独立的源文件里还有一层好处：读路径的每一处「哪些列被读出来」都集中在此， +// 与写路径（store.go 的 Add 与 preinsert_gate.go 的闸）对照时不必在 500 行里翻。 + +// rowScanner 同时被 *sql.Row 与 *sql.Rows 满足，因此单个 scanItem +// 辅助函数即可服务 Get 与 ListByPersona。 +type rowScanner interface { + Scan(dest ...any) error +} + +// scanItems 把每一行读入 MemoryItem 切片，关闭 rows 并用 +// 操作名包裹 scan/遍历失败。成功时总是返回 +// 非 nil、可能为空的切片。 +func scanItems(rows *sql.Rows, op string) ([]MemoryItem, error) { + defer func() { _ = rows.Close() }() + + items := make([]MemoryItem, 0) + for rows.Next() { + item, err := scanItem(rows) + if err != nil { + return nil, fmt.Errorf("memory: scan %s: %w", op, err) + } + items = append(items, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("memory: iterate %s: %w", op, err) + } + return items, nil +} + +// scanItem 读取一行 memory_items。tags 以 JSON 数组字符串存储； +// 时间戳以定宽 UTC 字符串（timeLayout）存储，因此文本 +// 比较即可按时间排序。 +func scanItem(sc rowScanner) (MemoryItem, error) { + var ( + it MemoryItem + scope, kind 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 { + return MemoryItem{}, err + } + it.Scope = Scope(scope) + it.Kind = Kind(kind) + if err := json.Unmarshal([]byte(tagsJSON), &it.Tags); err != nil { + return MemoryItem{}, fmt.Errorf("memory: decode tags: %w", err) + } + if it.Tags == nil { + it.Tags = []string{} + } + + var err error + if it.CreatedAt, err = parseTime(createdAt); err != nil { + return MemoryItem{}, err + } + if it.UpdatedAt, err = parseTime(updatedAt); err != nil { + return MemoryItem{}, err + } + return it, nil +}