package main import ( "bytes" "context" "errors" "log/slog" "os" "path/filepath" "strings" "testing" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" "git.yeij.top/AskaEth/Cyrene_v2/internal/db" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) // 本文件锁定 D3「入库前审查」(design/23 §15.1 D3)在**组合根**上的三件事: // 1. 跑在与整理同一个专属隔离会话(sess-memorytidy),不是主人的对话车道; // 2. 接线存在(服务被装配、被接到 store 的写入路径、被启动、被停止), // 少一处就会**静默不工作**(不报错、不审查、日志里一句都没有); // 3. 「独立会话、不碰主会话上下文」是**可验证**的:审查跑完后主会话的消息条数、 // 最后一条消息与已读游标逐字不变,且 chat_messages 里没有审查会话的行。 // reviewEmbedder 是组合根测试用的确定性 embedder:所有文本同向(余弦 1), // 因此「同内容的两条记忆」一定被召回,审查的判定路径真的会跑到。 type reviewEmbedder struct{} func (reviewEmbedder) Embed(_ context.Context, texts []string) ([]embedding.Vector, error) { out := make([]embedding.Vector, 0, len(texts)) for range texts { out = append(out, embedding.Vector{1, 0}) } return out, nil } func (reviewEmbedder) Dim() int { return 2 } // TestMemoryReviewIsWired 是接线存在性防回归:组合根少任何一处,入库前审查 // 都会静默不工作,而这不会让别的用例变红。 // // 刻意**不**在这里断言 `.env.example`:该文件由并行切片(MT-A)持有, // 新增的 MEMORY_REVIEW_* 键已登记在 MTB-report.md 里,由主会话统一补模板。 func TestMemoryReviewIsWired(t *testing.T) { t.Parallel() seams := map[string][]string{ "wire_runtime.go": { // 记忆栈建好后就装配审查(它只需要 store + embedder)。 "a.wireMemoryReview()", }, "wire_memory_review.go": { "config.LoadMemoryReviewConfig(envPath)", "memory.NewPreInsertReviewer(memory.ReviewConfig{", // 依赖面里**没有**任何会话/历史读数 —— 这是「独立会话、零上下文」的 // 结构性保证,而不是靠自觉。 "Audit: a.auditMemoryReview", // 接到 store 的唯一写入路径上。 "a.memStore.AttachIngestReviewer(reviewer)", // 同步入库闸(2026-10-07)的开关来自 `$DATA_DIR/config/memory-tidy.json` // 的 `{"ingestGate": false}`(**零新 env 键**,`.env` 已冻结)。 "a.memoryIngestGateOff()", "DisableIngestGate: gateOff", // T1 灰区模型判定(design/23 §15.1 D3-2):开关同样来自那个文件 // (`{"t1Judge": false}`),模型出口**在调用时才解析**(本服务早于 // 模型栈装配,与 a.gw 的晚组装同一手法)。 "a.memoryT1JudgeOff()", "DisableT1: t1Off", "Judge: a.memoryT1Judge()", // T1 的**全局限速**额度(design/23 §15.4)同样来自那个文件, // 缺位/坏文件一律回落代码默认(120 次/60 分钟)。 "a.memoryT1RateLimit()", "T1PerWindow: t1PerWindow", "T1Window: time.Duration(t1WindowSec) * time.Second", // 临时记忆 TTL(design/03 §9.6):第五态 `defer` 写进 `expires_at` 的 // 存活秒数,同样来自那个文件(`tempTtlSec`),缺位/坏值回落代码默认 // 7 天。**它没有 `.env` 档**(`.env` 已冻结,本键在那里从来没有出处), // 因此这一行被摘掉 = 面板配了也不生效,正是本缝断言要拦的回归。 "a.memoryTempTTL()", "TempTTL: tempTTL", }, "main.go": { "a.memoryReview.Start(ctx)", }, "wire_services.go": { // 关库之前先停审查(它要写 memory_items)。 "a.memoryReview.Stop(ctx)", }, } for file, wants := range seams { src, err := os.ReadFile(file) if err != nil { t.Fatalf("read %s: %v", file, err) } for _, want := range wants { if !strings.Contains(string(src), want) { t.Fatalf("%s 里找不到 %q —— 入库前审查没有被接上组合根", file, want) } } } // store 侧的唯一写入路径必须投递审查(否则新记忆永远不进队列), // 且必须**先过同步入库闸**(2026-10-07:闸就摆在写路径上,重复当场落 merged)。 storeSrc, err := os.ReadFile("../../internal/runtime/memory/store.go") if err != nil { t.Fatalf("read memory/store.go: %v", err) } for _, want := range []string{"s.enqueueReview(item)", "s.planIngest(ctx, item)", "s.finishIngest(ctx, item, plan)"} { if !strings.Contains(string(storeSrc), want) { t.Fatalf("memory/store.go 的写入路径上找不到 %q(唯一写路径上的闸/投递断了)", want) } } } // TestMemoryIngestGate_ReadsDataDirConfig 钉住**同步入库闸**的开关确实来自 // `$DATA_DIR/config/memory-tidy.json`(它没有 env 键:`.env` 已冻结)—— // 若组合根没读那个文件,「一键回到只插不审」就是一句空话。 // // 口径与自动档同一条:没配 = 开;显式 false = 关;坏文件 = 回落代码默认(开) // 并记 WARN,**绝不静默把一处防护关掉**。 func TestMemoryIngestGate_ReadsDataDirConfig(t *testing.T) { t.Parallel() empty := &app{cfg: &config.Config{DataDir: t.TempDir()}, log: tidyTestLog()} if empty.memoryIngestGateOff() { t.Fatal("没配过时入库闸应当保持代码默认(开)") } dataDir := t.TempDir() dir := filepath.Join(dataDir, "config") if err := os.MkdirAll(dir, 0o700); err != nil { t.Fatalf("mkdir: %v", err) } path := filepath.Join(dir, "memory-tidy.json") if err := os.WriteFile(path, []byte(`{"ingestGate": false}`), 0o600); err != nil { t.Fatalf("write: %v", err) } off := &app{cfg: &config.Config{DataDir: dataDir}, log: tidyTestLog()} if !off.memoryIngestGateOff() { t.Fatal("ingestGate: false 没被读到(闸没有一键关掉的路径)") } if err := os.WriteFile(path, []byte(`{"ingestGate": "nope"}`), 0o600); err != nil { t.Fatalf("write broken: %v", err) } broken := &app{cfg: &config.Config{DataDir: dataDir}, log: tidyTestLog()} if broken.memoryIngestGateOff() { t.Fatal("坏配置把入库闸静默关掉了:必须回落代码默认(开)") } } // TestMemoryT1Judge_ReadsDataDirConfig 钉住 **T1 灰区模型判定**的开关确实来自 // `$DATA_DIR/config/memory-tidy.json` 的 `t1Judge`(它没有 env 键:`.env` 已冻结) // —— T1 是一处**会花掉模型额度**的动作,「一行配置关掉」必须真的成立。 // // 口径与入库闸同一条:没配 = 开(主人 2026-10-07 要的就是「真的走记忆管理会话」); // 显式 false = 关(回落纯规则、零模型调用);坏文件 = 回落代码默认(开)并记 // WARN,**绝不静默把一处防护关掉**。 func TestMemoryT1Judge_ReadsDataDirConfig(t *testing.T) { t.Parallel() empty := &app{cfg: &config.Config{DataDir: t.TempDir()}, log: tidyTestLog()} if empty.memoryT1JudgeOff() { t.Fatal("没配过时 T1 应当保持代码默认(开)") } dataDir := t.TempDir() dir := filepath.Join(dataDir, "config") if err := os.MkdirAll(dir, 0o700); err != nil { t.Fatalf("mkdir: %v", err) } path := filepath.Join(dir, "memory-tidy.json") // 只写一个上限键:它不许把 T1 静默关掉(指针字段的存在理由)。 if err := os.WriteFile(path, []byte(`{"autoApplyPerRound": 5}`), 0o600); err != nil { t.Fatalf("write: %v", err) } partial := &app{cfg: &config.Config{DataDir: dataDir}, log: tidyTestLog()} if partial.memoryT1JudgeOff() { t.Fatal("只写了别的键时 T1 被静默关掉了(必须保持代码默认的「开」)") } if err := os.WriteFile(path, []byte(`{"t1Judge": false}`), 0o600); err != nil { t.Fatalf("write: %v", err) } off := &app{cfg: &config.Config{DataDir: dataDir}, log: tidyTestLog()} if !off.memoryT1JudgeOff() { t.Fatal("t1Judge: false 没被读到(T1 没有一键关掉的路径)") } if err := os.WriteFile(path, []byte(`{"t1Judge": "nope"}`), 0o600); err != nil { t.Fatalf("write broken: %v", err) } broken := &app{cfg: &config.Config{DataDir: dataDir}, log: tidyTestLog()} if broken.memoryT1JudgeOff() { t.Fatal("坏配置把 T1 静默关掉了:必须回落代码默认(开)") } } // TestMemoryT1RateLimit_ReadsDataDirConfig 钉住 **T1 全局限速**的额度确实来自 // `$DATA_DIR/config/memory-tidy.json` 的 `t1PerWindow` / `t1WindowSec` // (design/23 §15.4):T1 默认开,每次有灰区候选的审查就发一次调用 ⇒ 额度必须能 // 按现网节奏调,而 `.env` 已冻结、不许为它开 env 键。 // // 口径与入库闸/T1 开关同一条:没配 = 返回零值(交给 normalizeReviewConfig 回落 // 代码默认 120 次/60 分钟);坏文件 = 同样回落默认额 **并记 WARN** —— // 坏配置绝不能让限速静默变成「无上限」。 func TestMemoryT1RateLimit_ReadsDataDirConfig(t *testing.T) { t.Parallel() empty := &app{cfg: &config.Config{DataDir: t.TempDir()}, log: tidyTestLog()} if perWindow, windowSec := empty.memoryT1RateLimit(); perWindow != 0 || windowSec != 0 { t.Fatalf("没配过时应当返回零值(= 回落代码默认):%d/%d", perWindow, windowSec) } dataDir := t.TempDir() dir := filepath.Join(dataDir, "config") if err := os.MkdirAll(dir, 0o700); err != nil { t.Fatalf("mkdir: %v", err) } path := filepath.Join(dir, "memory-tidy.json") if err := os.WriteFile(path, []byte(`{"t1PerWindow": 240, "t1WindowSec": 1800}`), 0o600); err != nil { t.Fatalf("write: %v", err) } set := &app{cfg: &config.Config{DataDir: dataDir}, log: tidyTestLog()} if perWindow, windowSec := set.memoryT1RateLimit(); perWindow != 240 || windowSec != 1800 { t.Fatalf("额度没被读到:%d/%d, want 240/1800", perWindow, windowSec) } // 坏配置:回落默认额(零值)+ 一条 WARN,绝不静默不限流。 if err := os.WriteFile(path, []byte(`{"t1PerWindow": "many"}`), 0o600); err != nil { t.Fatalf("write broken: %v", err) } var buf bytes.Buffer broken := &app{cfg: &config.Config{DataDir: dataDir}, log: slog.New(slog.NewTextHandler(&buf, nil))} if perWindow, windowSec := broken.memoryT1RateLimit(); perWindow != 0 || windowSec != 0 { t.Fatalf("坏配置下应当回落零值:%d/%d", perWindow, windowSec) } if logged := buf.String(); !strings.Contains(logged, "t1 rate limit tuning unreadable") || !strings.Contains(logged, "level=WARN") { t.Fatalf("坏配置没有留下 WARN:%q", logged) } } // TestMemoryT1Judge_UsesTheWiredModelStack 钉住 T1 的模型出口**复用既有模型栈** // (不新造 LLM 客户端),且返回一个闭包而不是当场取值 —— 本服务在 wireMemory // 阶段装配,而 a.modelAdapter 要等更晚的 wireRuntime 才建好。 func TestMemoryT1Judge_UsesTheWiredModelStack(t *testing.T) { t.Parallel() a := &app{log: tidyTestLog()} judge := a.memoryT1Judge() if judge == nil { t.Fatal("memoryT1Judge 返回了 nil") } // 模型栈还没建好:必须 fail-safe(返回错误),绝不 panic、绝不假装成功。 if _, err := judge(context.Background(), model.Request{}, model.Options{}); !errors.Is(err, errT1ModelNotReady) { t.Fatalf("模型栈未就绪时 err = %v, want %v", err, errT1ModelNotReady) } // 接上模型栈后,同一个闭包就能发请求(晚解析的证据)。 stub := &reviewAdapterStub{reply: "{}"} a.modelAdapter = stub if _, err := judge(context.Background(), model.Request{Purpose: "memory_t1_judge"}, model.Options{}); err != nil { t.Fatalf("接上模型栈后调用失败:%v", err) } if stub.calls != 1 { t.Fatalf("模型栈被调用 %d 次, want 1", stub.calls) } } // reviewAdapterStub 是最小的 model.Adapter 桩(只为验「闭包解析到既有的那个 // 模型栈」这件事,不参与判定正确性)。 type reviewAdapterStub struct { calls int reply string } func (s *reviewAdapterStub) Name() string { return "stub" } func (s *reviewAdapterStub) Chat(context.Context, model.Request, model.Options) (model.Response, error) { s.calls++ return model.Response{Content: s.reply}, nil } func (s *reviewAdapterStub) ChatStream( ctx context.Context, req model.Request, onEvent func(model.StreamEvent), ) (model.Response, error) { return s.Chat(ctx, req, model.Options{}) } // TestMemoryReviewSessionIsNotConversationLane 与整理同一条口径:审查会话不落在 // 主人的对话车道("web" / "web:")上,因此不改她的心情、不进主会话泳道。 func TestMemoryReviewSessionIsNotConversationLane(t *testing.T) { t.Parallel() if isConversationLane(memory.TidySessionID) { t.Fatalf("审查会话 %q 被判成主人对话回合", memory.TidySessionID) } if !isConversationLane(history.MainSessionID) { t.Fatal("主人车道判定被改坏了") } } // TestMemoryReviewLeavesChatHistoryUntouched 是 design/23 §15 D3-7 的可验证版本: // 审查真的干活(把重复行折回既有行)之后,主会话的消息条数、最后一条消息与 // read_at 游标逐字不变,且 chat_messages 里没有任何属于审查会话的行。 func TestMemoryReviewLeavesChatHistoryUntouched(t *testing.T) { ctx := context.Background() conn, err := db.Open(ctx, ":memory:", tidyTestLog()) if err != nil { t.Fatalf("open db: %v", err) } defer func() { _ = conn.Close() }() if err := db.MigrateUp(ctx, conn, tidyTestLog()); err != nil { t.Fatalf("migrate: %v", err) } hist, err := history.NewStore(ctx, conn, tidyTestLog()) if err != nil { t.Fatalf("history store: %v", err) } if _, err := hist.AppendFull(ctx, history.Message{ SessionID: history.MainSessionID, Role: history.RoleUser, Content: "今晚又要加班到很晚", CreatedAt: 1_700_000_000_000, }); err != nil { t.Fatalf("append user message: %v", err) } last, err := hist.AppendFull(ctx, history.Message{ SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "别熬太晚,我陪你", CreatedAt: 1_700_000_001_000, }) if err != nil { t.Fatalf("append assistant message: %v", err) } if _, err := hist.MarkSessionRead(ctx, history.MainSessionID, time.UnixMilli(1_700_000_000_500)); err != nil { t.Fatalf("mark read: %v", err) } before := readConversationFingerprint(t, ctx, conn, history.MainSessionID) store, err := memory.NewStore(ctx, conn, tidyTestLog()) if err != nil { t.Fatalf("memory store: %v", err) } content := "主人常在深夜与助手聊天,有熬夜习惯" seed := func(at time.Time) memory.MemoryItem { got, err := store.Add(ctx, memory.MemoryItem{ PersonaID: "persona.a", Scope: memory.ScopePersona, Kind: memory.KindRelationship, Content: content, Confidence: 0.9, CreatedAt: at, SourceConversation: memory.FormatSourceConversation(history.MainSessionID, "turn-1"), }) if err != nil { t.Fatalf("seed memory: %v", err) } return got } seed(time.UnixMilli(1_700_000_000_000)) reviewer, err := memory.NewPreInsertReviewer(memory.ReviewConfig{ Enabled: true, QueueSize: 8, ScanLimit: 20, Timeout: 5 * time.Second, Threshold: 0.8, }, memory.ReviewDeps{Store: store, Embedder: reviewEmbedder{}, Log: tidyTestLog()}) if err != nil { t.Fatalf("NewPreInsertReviewer: %v", err) } store.AttachIngestReviewer(reviewer) reviewer.Start(ctx) defer func() { stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _ = reviewer.Stop(stopCtx) }() dup := seed(time.UnixMilli(1_700_000_060_000)) deadline := time.Now().Add(3 * time.Second) for time.Now().Before(deadline) { var status string if err := conn.QueryRowContext(ctx, "SELECT status FROM memory_items WHERE id = ?", dup.ID).Scan(&status); err != nil { t.Fatalf("read dup status: %v", err) } if status == "merged" { break } time.Sleep(5 * time.Millisecond) } after := readConversationFingerprint(t, ctx, conn, history.MainSessionID) if before != after { t.Fatalf("审查改动了主会话上下文:\n before=%+v\n after=%+v", before, after) } if after.lastContent != last.Content || after.lastID != last.ID { t.Fatalf("主会话最后一条消息变了:%+v", after) } var reviewMsgs int if err := conn.QueryRowContext(ctx, "SELECT count(*) FROM chat_messages WHERE session_id = ?", memory.TidySessionID).Scan(&reviewMsgs); err != nil { t.Fatalf("count review messages: %v", err) } if reviewMsgs != 0 { t.Fatalf("审查向 chat_messages 写入了 %d 行(§15 D3-7 要求零写入)", reviewMsgs) } reviews, err := store.ListIngestReviews(ctx, 10) if err != nil { t.Fatalf("ListIngestReviews: %v", err) } if len(reviews) != 1 || reviews[0].Decision != memory.IngestSkip { t.Fatalf("审查留痕 = %+v, want 恰为一条 skip", reviews) } }