AskaEth
a684328b76
主人 2026-10-07 裁定：按最优解加全局限流。 - 新增 t1RateGate（滑动窗口、可注入时钟，~70 行）；默认 120 次 / 60 分钟（现网 700-1000 条新记忆/天 ≈ 25-42/小时， 取 120 留约 3-5 倍余量） - 过闸摆在「t1Enabled + 有灰区候选」两个守卫之后、发请求之前 ⇒ 不调模型就不占额度；放行才记账 - 超限语义与既有失败路径同源：跳过 T1、决定仍是 create、零改写， 只多一个 reason=t1_throttled 进留痕 + ReviewStats.T1Throttled 计数 - 配置键 t1PerWindow / t1WindowSec 放 $DATA_DIR/config/memory-tidy.json （零新 env 键，启动期读一次）；坏配置回落默认 + WARN，不启动失败 - 每窗口只记一条 WARN（超限是预期行为，一天几百条会淹掉真故障） - IngestRuleVersion 仍为 d3.2：判定规则集与解析契约没变，变的只是 额度状态，reason 列已可分辨（升版会有 §17.6 记过的幂等键重复行副作用） 自研而非复用 session.NewRateLimiter 的理由（design/12 §N C.5 自己登记过 这条取舍）：既有件 Allow() 内部写死 time.Now()，没有时钟注入缝， 而本切片必须能注入可控时钟；且复用会把 channel 层概念拖进 memory 包。
403 lines
16 KiB
Go
403 lines
16 KiB
Go
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",
|
||
},
|
||
"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:<uuid>"）上，因此不改她的心情、不进主会话泳道。
|
||
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)
|
||
}
|
||
}
|