feat(memory): D3 从「异步后置」改成「同步入库闸」—— 由整理层决定入库形态

主人 2026-10-07 裁定：「把记忆整理功能接到昔涟和记忆库中间做审查，由整理功能
进行入库」。现状是**异步 + 入库后软合并** ⇒ 行已经落库才审，面板/召回先脏后清，
而写入速率实测 ~25 条/小时（≈600/天）⇒ 清理永远追不上。

落法（三条硬规格，全部落地）：
① **同步闸但有界**：闸在 `store.go:247` 的 `planIngest`（INSERT **之前**），
   本体 `preinsert_gate.go`。墙钟预算 **150ms**（实测最坏 p95 仅 4~7ms ⇒ 20~50 倍
   余量）、候选扫描上限 **100**（比异步侧 200 更紧，因为它在写路径上）。
   🔴 **零远端 embedding**：闸恒用词项余弦（`deps.Embedder` 根本不参与）——
   远端 embedding 一次 2.3~3.0s 且对同文本返回不同向量（不可复现），不许摆在同步
   路径上拍板。
② **fail-open，绝不丢记忆**：候选源报错 ⇒ `gate:candidates_unavailable`；超预算 ⇒
   `gate:budget_exceeded` ⇒ **照常 insert(active) + 投既有队列 + 留痕 + WARN**。
③ **三态落库（落 merged，而不是"不插入"）**：确定性命中（字面重复 / 归一化等价 /
   真包含）⇒ INSERT **直接带 `status='merged'` + `merged_into=<存活行>`**，存活行
   `merged_from` 追加、`merge_count`+1 ⇒ **新行从不进 active、从不建向量索引**，
   而**行与原文都在表里 ⇒ revert 无损**。若"干脆不插"，那条文本从未存在过行，
   revert 无从下手，违反 design/23 §8 的红线。
- **不与异步 D3 重复审**：闸做决定的那条 `Async=false`（留痕由闸自己落，
  审计动作名 `memory.ingest.gate`）；真新/拿不准的走原有队列（留痕仍归异步层）。
- **开关**：`$DATA_DIR/config/memory-tidy.json` 的 `{"ingestGate": false}` ⇒ 一键回到
  「只插不审」（= 改动前的异步行为）。**零新 env 键**（`.env` 未碰）。
- 顺手把 `store.go` 的行映射拆到 `store_rows.go`（守 §10 的 ≤500 行）。

验收（主会话复跑 + 亲自变异）：`go build ./...` + `go vet` 绿；memory 包测试绿；
我用 -overlay 摘掉闸的预算 ⇒ `TestIngestGate_Add墙钟有界` 与
`…failOpen_超预算绝不丢记忆` 变红（"Add 等了 2.0s，越过闸预算 150ms"）。
子代理另做 2 个变异（去掉预算 / 把 fail-open 改成 fail-closed）全红，并实测
**闸开 vs 关的 `Add` 额外墙钟：p50 ≈2.5~3.1ms、p95 ≈4~7ms**（基线批均值 35.8µs）。

⚠️ 未部署（要一次 build + restart）；⚠️ 已知边界（如实）：闸只用词项余弦 ⇒
「同义但字面差得远」拦不住（有意为之，由异步层/夜间整理兜）；闸的候选是**最新
100 条**（更老的重复闸看不到）；**并发同内容双写**仍可能两条都落 active（既有竞态，
SELECT 与 INSERT 不在同一事务，非本次引入，第二层会清）。

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) {

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) {

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)

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"))

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)
-}

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)

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
+}

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)
+	}
+}

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]
+}

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
+		}
+	}
+}

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)
+}

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))
 }
-
-

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
+}