From a684328b765c8bd8dd843a9f005c6c4d979316fb Mon Sep 17 00:00:00 2001 From: AskaEth Date: Wed, 7 Oct 2026 16:25:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(memory):=20T1=20=E5=8A=A0=E4=BF=9D?= =?UTF-8?q?=E5=AE=88=E5=85=A8=E5=B1=80=E9=80=9F=E7=8E=87=E9=99=90=E6=B5=81?= =?UTF-8?q?=EF=BC=88=E9=98=B2=E6=AF=8F=E5=A4=A9=E6=95=B0=E7=99=BE=E5=88=B0?= =?UTF-8?q?=E4=B8=8A=E5=8D=83=E6=AC=A1=E8=B0=83=E7=94=A8=E5=A4=B1=E6=8E=A7?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 主人 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 包。 --- cmd/cyrene/wire_memory_review.go | 39 +- cmd/cyrene/wire_memory_review_test.go | 53 +++ internal/config/runtimeconfig/memorytidy.go | 21 + .../config/runtimeconfig/memorytidy_test.go | 48 ++ internal/runtime/memory/preinsert.go | 18 +- internal/runtime/memory/preinsert_config.go | 35 ++ internal/runtime/memory/preinsert_stats.go | 9 +- internal/runtime/memory/preinsert_t1.go | 55 ++- internal/runtime/memory/preinsert_t1_limit.go | 102 +++++ .../runtime/memory/preinsert_t1_limit_test.go | 412 ++++++++++++++++++ 10 files changed, 782 insertions(+), 10 deletions(-) create mode 100644 internal/runtime/memory/preinsert_t1_limit.go create mode 100644 internal/runtime/memory/preinsert_t1_limit_test.go diff --git a/cmd/cyrene/wire_memory_review.go b/cmd/cyrene/wire_memory_review.go index 146a120..84f19c8 100644 --- a/cmd/cyrene/wire_memory_review.go +++ b/cmd/cyrene/wire_memory_review.go @@ -39,7 +39,9 @@ var errT1ModelNotReady = errors.New("memory review: t1 judge: model adapter not // 🔴 两个开关都不在 env 里（`.env` 已冻结，design/13 §2.1），它们读 // `$DATA_DIR/config/memory-tidy.json`，与整理自动档同一个文件、**零新 env 键**： // - `{"ingestGate": false}` ⇒ 一键回到「只插不审」（同步入库闸）； -// - `{"t1Judge": false}` ⇒ 一键回到纯规则（T1 灰区判定，**零模型调用**）。 +// - `{"t1Judge": false}` ⇒ 一键回到纯规则（T1 灰区判定，**零模型调用**）； +// - `{"t1PerWindow": 120, "t1WindowSec": 3600}` ⇒ T1 的**全局限速**额度 +// （design/23 §15.4）；缺位 = 代码默认 120 次/60 分钟。 func (a *app) wireMemoryReview() error { cfg, err := config.LoadMemoryReviewConfig(envPath) if err != nil { @@ -55,6 +57,10 @@ func (a *app) wireMemoryReview() error { // 一键回到纯规则（零模型调用）。与入库闸同一个文件、同一套「没配 = 保持默认」 // 口径，**零新 env 键**。 t1Off := a.memoryT1JudgeOff() + // T1 的全局限速额度（design/23 §15.4）：同样读那个文件，缺位/坏文件一律 + // 交零值给 normalizeReviewConfig 回落**代码默认值**（120 次/60 分钟）—— + // 坏配置绝不能让一处限速静默变成「无上限」。 + t1PerWindow, t1WindowSec := a.memoryT1RateLimit() reviewer, err := memory.NewPreInsertReviewer(memory.ReviewConfig{ Enabled: cfg.Enabled, QueueSize: cfg.QueueSize, @@ -70,6 +76,12 @@ func (a *app) wireMemoryReview() error { // T1 的时长走代码默认值（8s，且被单条审查的 Timeout 收紧取小）： // 它是**实现参数**不是运维旋钮 —— 要让 T1 完全不花钱，就关掉它。 DisableT1: t1Off, + // T1 的**全局限速**额度（design/23 §15.4）：数值来自 + // `$DATA_DIR/config/memory-tidy.json` 的 `t1PerWindow` / `t1WindowSec`， + // 缺位或坏配置时是零值 ⇒ 由 normalizeReviewConfig 回落代码默认 + // （120 次/60 分钟）。**没有「无上限」这个状态**：要零花费请关 t1Judge。 + T1PerWindow: t1PerWindow, + T1Window: time.Duration(t1WindowSec) * time.Second, }, memory.ReviewDeps{ Store: a.memStore, Embedder: a.embedder, @@ -114,6 +126,31 @@ func (a *app) memoryT1JudgeOff() bool { return tuning.T1JudgeOff() } +// memoryT1RateLimit 读 **T1 全局限速**的两项旋钮：窗口额度（次）与窗口长度（秒）。 +// +// 口径与 memoryIngestGateOff / memoryT1JudgeOff 逐字相同：读不到/解不开 +// `memory-tidy.json` 时返回零值（= 不覆盖）并记 WARN —— 零值经 +// normalizeReviewConfig 回落**代码默认值**（120 次/60 分钟），因此坏配置既不会 +// 让限速静默变成「无上限」，也不会让 T1 静默失效。 +// +// 只有**两项都写了且为正**才报一条 Info（自解释的配置回执）；只写了一项时静默 +// 回落缺的那一项，不报「已配置」以免报出一个与生效值不符的印象。 +func (a *app) memoryT1RateLimit() (perWindow int, windowSec int) { + tuning, path, err := a.readMemoryTidyTuning() + if err != nil { + logger.WithSource(a.log, "memory/review").Warn( + "memory t1 rate limit tuning unreadable, falling back to code defaults", + "path", path, "error", err) + return 0, 0 + } + if tuning.T1PerWindow > 0 && tuning.T1WindowSec > 0 { + logger.WithSource(a.log, "memory/review").Info( + "memory t1 rate limit loaded", + "path", path, "t1_per_window", tuning.T1PerWindow, "t1_window_sec", tuning.T1WindowSec) + } + return tuning.T1PerWindow, tuning.T1WindowSec +} + // memoryT1Judge 把「调用时才解析的模型适配器」包成 T1 的一次性调用口。 // // 为什么是闭包而不是直接塞 `a.modelAdapter`：组合根的顺序是 **wireMemory（建 diff --git a/cmd/cyrene/wire_memory_review_test.go b/cmd/cyrene/wire_memory_review_test.go index 07b2720..9b4459a 100644 --- a/cmd/cyrene/wire_memory_review_test.go +++ b/cmd/cyrene/wire_memory_review_test.go @@ -1,8 +1,10 @@ package main import ( + "bytes" "context" "errors" + "log/slog" "os" "path/filepath" "strings" @@ -68,6 +70,11 @@ func TestMemoryReviewIsWired(t *testing.T) { "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)", @@ -186,6 +193,52 @@ func TestMemoryT1Judge_ReadsDataDirConfig(t *testing.T) { } } +// 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 才建好。 diff --git a/internal/config/runtimeconfig/memorytidy.go b/internal/config/runtimeconfig/memorytidy.go index 09bfedb..74c07e0 100644 --- a/internal/config/runtimeconfig/memorytidy.go +++ b/internal/config/runtimeconfig/memorytidy.go @@ -17,6 +17,12 @@ // 是同一条 —— 它是一处**会花掉模型额度**的开关，必须能一行配置关掉（关掉即回落 // 纯规则、零模型调用），所以同样落在这里，不新开文件家族、不加 env 键。 // +// 2026-10-07 同日追加 `t1PerWindow` / `t1WindowSec`（T1 的**全局限速**， +// design/23 §15.4 主人裁定「加一层保守的全局速率限流」）：T1 默认开，每次「有灰区 +// 候选的审查」就发一次调用，现网每天数百到上千条新记忆 ⇒ 那是一处**没有刹车**的 +// 额度出口。额度必须能按现网节奏调（且不许为它开 env 键），因此落在这里； +// **零值不表示「无限制」**，只表示「没配、用代码默认值」。 +// // 边界与 backthink.go 同源（那是同一族的手续）： // // 1. 本文件只管**文件形状 + 读取**：它不认识 `TidyConfig`，也不认识「零值回落 @@ -87,6 +93,21 @@ type MemoryTidyTuning struct { // 指针也是同一个理由：代码默认值是「开」，用 bool 会让「这个键没写」与 // 「显式写 false」不可区分。 T1Judge *bool `json:"t1Judge"` + // T1PerWindow 是 T1 的**全局**窗口额度（次数）—— 全进程共享一个窗口，超限 + // 即跳过 T1 并回落 `create`（`reason=t1_throttled`，design/23 §15.4）。 + // 零/负 = 没配 ⇒ 用代码默认值（memory 侧的 defaultT1PerWindow）。 + // + // **刻意没有「无上限」语义**（与 autoApplyPerRound 同一口径）：≤0 一律回落 + // 默认额 —— 「不限流」不该是敲一个 0 就能得到的状态（design/12 §N C.5 G3 + // 登记过那条静默失败形态）。要完全不花模型额度，请关 `t1Judge`。 + T1PerWindow int `json:"t1PerWindow"` + // T1WindowSec 是上面那个额度的时间窗（**秒**）。零/负 = 没配 ⇒ 用代码默认值 + // （memory 侧的 defaultT1Window）。 + // + // 用秒而不是 Go 的时长字面量（`"1h"`）是刻意的：这个文件是给人改的，写 + // `3600` 比要求主人记住 Go 的时长语法更不容易写错，也与既有 + // `$DATA_DIR/config/backthink.json` 的 `*_SEC` 口径一致。 + T1WindowSec int `json:"t1WindowSec"` } // AutoApplyOff 报告自动档是否被这份配置**显式关掉**。nil（没配）返回 false —— diff --git a/internal/config/runtimeconfig/memorytidy_test.go b/internal/config/runtimeconfig/memorytidy_test.go index 1477776..ee7a2ef 100644 --- a/internal/config/runtimeconfig/memorytidy_test.go +++ b/internal/config/runtimeconfig/memorytidy_test.go @@ -134,6 +134,54 @@ func TestLoadMemoryTidyTuning_IngestGate(t *testing.T) { } } +// TestLoadMemoryTidyTuning_T1RateLimit 钉住 **T1 全局限速**的两个数值键 +// （design/23 §15.4，主人 2026-10-07 裁定）：它们与 `t1Judge` 同一个文件、 +// **零新 env 键**，且**零值只能是「没配」**——绝不能是「无上限」。 +func TestLoadMemoryTidyTuning_T1RateLimit(t *testing.T) { + t.Parallel() + + // ① 没写 ⇒ 保持零值（调用方据此回落代码默认额，不是「不限流」）。 + absent := writeMemoryTidyFile(t, `{"t1Judge": true}`) + got, err := LoadMemoryTidyTuning(absent) + if err != nil { + t.Fatalf("LoadMemoryTidyTuning: %v", err) + } + if got.T1PerWindow != 0 || got.T1WindowSec != 0 { + t.Fatalf("没写限速键应当给零值：%+v", got) + } + + // ② 写了 ⇒ 如实回读，且不连带改动别的键（一个键不能顺手关掉另一件事）。 + set := writeMemoryTidyFile(t, `{"t1PerWindow": 240, "t1WindowSec": 1800}`) + got, err = LoadMemoryTidyTuning(set) + if err != nil { + t.Fatalf("LoadMemoryTidyTuning(set): %v", err) + } + if got.T1PerWindow != 240 || got.T1WindowSec != 1800 { + t.Fatalf("限速键回读 = %d/%d, want 240/1800", got.T1PerWindow, got.T1WindowSec) + } + if got.T1JudgeOff() || got.IngestGateOff() || got.AutoApplyOff() { + t.Fatalf("限速键顺手改了别的开关：%+v", got) + } + + // ③ 显式 0 ⇒ 仍是零值（「没法用 0 关掉限流」是刻意的：那会是一处静默的失效面）。 + zero := writeMemoryTidyFile(t, `{"t1PerWindow": 0, "t1WindowSec": 0}`) + got, err = LoadMemoryTidyTuning(zero) + if err != nil { + t.Fatalf("LoadMemoryTidyTuning(zero): %v", err) + } + if got.T1PerWindow != 0 || got.T1WindowSec != 0 { + t.Fatalf("显式 0 的回读 = %+v", got) + } + + // ④ 类型错 ⇒ 分类错误（调用方据此记 WARN 并回落代码默认额，绝不静默不限流）。 + for _, body := range []string{`{"t1PerWindow": "many"}`, `{"t1WindowSec": "1h"}`} { + broken := writeMemoryTidyFile(t, body) + if _, err := LoadMemoryTidyTuning(broken); !errors.Is(err, ErrInvalidMemoryTidy) { + t.Fatalf("坏配置 %s 的 err = %v, want ErrInvalidMemoryTidy", body, err) + } + } +} + func TestDefaultMemoryTidyPathIsDataDirConfig(t *testing.T) { t.Parallel() got := DefaultMemoryTidyPath(filepath.Join("root", "data")) diff --git a/internal/runtime/memory/preinsert.go b/internal/runtime/memory/preinsert.go index 3bd4e13..0d28f98 100644 --- a/internal/runtime/memory/preinsert.go +++ b/internal/runtime/memory/preinsert.go @@ -31,7 +31,9 @@ import ( // 3. **规则优先、模型兜底**：T0 是纯规则（design §15 D3-2 T0），**恒为零模型 // 调用**；T1（灰区调一次轻量模型，§15.1 D3-2）默认**开**，但只在「T0 判不出 // 结论且存在灰区候选」时才发出**至多一次**调用（preinsert_t1.go），并且 -// 要有模型出口才成立 —— 两者任一缺席即是结构性零模型调用。 +// 要有模型出口才成立 —— 两者任一缺席即是结构性零模型调用。每一次这样的 +// 调用还要先过**全局限速**（默认 120 次/小时，preinsert_t1_limit.go）： +// 超限即跳过 T1、回落 create + `t1_throttled`。 // **同步入库闸（preinsert_gate.go）不碰 T1**，写入路径因此仍是零模型调用。 // // 代价（如实登记）：极短窗口内库里会多一行待裁定的重复条目，随后被软合并 @@ -141,6 +143,11 @@ type PreInsertReviewer struct { log *slog.Logger now func() time.Time queue chan MemoryItem + // t1Gate 是 T1 的**全局限速**闸（design/23 §15.4）。挂在实例上而不是包级 + // 变量：AGENTS §10 禁包级全局变量，而组合根只建一个审查服务 ⇒ 实例字段 + // 事实上就是「全进程共享一个窗口额度」。它与 r.now 同源，因此测试可以整体 + // 推进时钟（见 preinsert_t1_limit.go）。 + t1Gate *t1RateGate mu sync.Mutex running bool @@ -168,6 +175,11 @@ func NewPreInsertReviewer(cfg ReviewConfig, deps ReviewDeps) (*PreInsertReviewer log: logger.WithSource(baseTidyLog(deps.Log), "memory/review"), now: now, queue: make(chan MemoryItem, cfg.QueueSize), + // 限流闸在 normalize **之后**构建：闸把非正的额度读成「恒放行」，因此 + // 归一前的 ≤0 一旦直达它，「回落默认」就会静默变成「无上限」 + // （design/12 §N C.5 G3 登记过的失败形态）。这里的 cfg 来自上面那行 + // normalizeReviewConfig。 + t1Gate: newT1RateGate(cfg.T1PerWindow, cfg.T1Window, now), }, nil } @@ -243,6 +255,10 @@ func (r *PreInsertReviewer) Start(ctx context.Context) { "t1_off_by_config", r.cfg.DisableT1, "t1_timeout", r.t1Timeout().String(), "t1_max_gray", defaultT1MaxGray, + // T1 的全局限速额度（已 normalize 的**生效值**，不是配置原文）： + // 「以为不限流其实限着」与反过来的误解都在启动行上即可排除。 + "t1_per_window", r.cfg.T1PerWindow, + "t1_window", r.cfg.T1Window.String(), "sweep_interval", DefaultIngestReviewGCInterval.String(), "retention", DefaultIngestReviewRetention.String(), "max_reviews", DefaultIngestReviewMaxRows) diff --git a/internal/runtime/memory/preinsert_config.go b/internal/runtime/memory/preinsert_config.go index e806d99..8627033 100644 --- a/internal/runtime/memory/preinsert_config.go +++ b/internal/runtime/memory/preinsert_config.go @@ -24,6 +24,18 @@ const ( // 不是任何角色（design §4.1 C 的同一理由）。 reviewAuditWho = "system/memoryreview" reviewAuditVerb = "memory.review.ingest" + + // defaultT1PerWindow / defaultT1Window 是 T1 **全局限速**的默认额度 + // （design/23 §15.4，主人 2026-10-07 裁定「加一层保守的全局速率限流」）。 + // + // 推算依据（2026-10-07）：现网节奏是每天 700–1000 条新记忆（≈25–42 条/小时）， + // 而 T1 只在「T0 判不出结论**且**存在灰区候选」时才发调用 —— 每条新记忆至多换 + // 一次调用。取 **120 次 / 60 分钟**，即在最忙的那一小时也留 ≈3–5 倍余量： + // 正常节奏下这个闸几乎碰不到；只有「召回/规则突然把大批条目判成灰区」这类 + // 异常形态才会触顶，而那正是要拦的东西。超限不是错误：跳过 T1、回落 create、 + // 留痕 `t1_throttled`，记忆一条不丢、一个字不改。 + defaultT1PerWindow = 120 + defaultT1Window = 60 * time.Minute ) // ReviewConfig 携带入库前审查的调优旋钮。零值经 normalize 选取安全默认值。 @@ -73,6 +85,20 @@ type ReviewConfig struct { // T1Timeout 是**单次** T1 模型调用的墙钟上限。非正用代码默认值 // defaultT1Timeout；最终还会被 cfg.Timeout 收紧取小（见 t1Timeout）。 T1Timeout time.Duration + // T1PerWindow 是 T1 的**全局**窗口额度（次）—— 不是每条各自的限制，而是 + // 全进程共享的一个窗口：T1Window 内至多这么多次模型调用，超限即跳过 T1 并 + // 回落 `create`（reason=`t1_throttled`，design/23 §15.4）。 + // + // 非正用代码默认值 defaultT1PerWindow。**刻意没有「无上限」语义**： + // design/12 §N C.5 G3 登记过那条静默失败形态（未归一的 ≤0 直达滑窗 ⇒ 被读成 + // 「无限制」，没有报错、没有 WARN）；要完全不花钱请用 + // `$DATA_DIR/config/memory-tidy.json` 的 `{"t1Judge": false}` 关掉 T1。 + // + // 额度**只被真正的模型调用消耗**（闸放行才记账，被拒的那次不占），因此 + // 「窗口过了就恢复」是结构性成立的。 + T1PerWindow int + // T1Window 是 T1PerWindow 的时间窗。非正用代码默认值 defaultT1Window。 + T1Window time.Duration } // normalizeReviewConfig 用设计稿默认值填充零值字段。 @@ -98,5 +124,14 @@ func normalizeReviewConfig(cfg ReviewConfig) ReviewConfig { if cfg.T1Timeout <= 0 { cfg.T1Timeout = defaultT1Timeout } + // 🔴 限流的两项必须在这里回落（design/12 §N C.5 G3 的教训）：闸是照着 + // 「非正 = 恒放行」实现的，未归一的 ≤0 直达它就会把「回落默认」**静默**变成 + // 「无上限」。因此 NewPreInsertReviewer 保证闸在 normalize **之后**构建。 + if cfg.T1PerWindow <= 0 { + cfg.T1PerWindow = defaultT1PerWindow + } + if cfg.T1Window <= 0 { + cfg.T1Window = defaultT1Window + } return cfg } diff --git a/internal/runtime/memory/preinsert_stats.go b/internal/runtime/memory/preinsert_stats.go index 3d57f4c..d4a8dac 100644 --- a/internal/runtime/memory/preinsert_stats.go +++ b/internal/runtime/memory/preinsert_stats.go @@ -33,6 +33,11 @@ type ReviewStats struct { // 的差值就是「调了但没用上」（非法回复 / 超时 / 报错回落 create）—— // 这个差值本身是可观测信号，因此两个数分开计。 T1Decided int + // T1Throttled 是**全局限速**命中、因而跳过 T1 的次数（design/23 §15.4， + // reason=`t1_throttled`）。它**不**计入 T1Called（一次模型调用都没发）， + // 也不计入 T1Decided（没定任何结论）——「想调但被闸拦下」是与「调了没用上」 + // 不同的一件事，分开计才看得出额度是不是成了瓶颈。 + T1Throttled int } // bump 在互斥下修改计数（工作协程、手动 Review 与同步闸都可能调用）。 @@ -75,7 +80,7 @@ func (r *PreInsertReviewer) auditWith(verb string, rev IngestReview) { // 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 t1_called=%d t1_decided=%d", + "gate_checked=%d gate_absorbed=%d gate_failed=%d t1_called=%d t1_decided=%d t1_throttled=%d", s.Reviewed, s.Skipped, s.Merged, s.Modified, s.Created, s.Dropped, s.Failed, - s.GateChecked, s.GateAbsorbed, s.GateFailed, s.T1Called, s.T1Decided) + s.GateChecked, s.GateAbsorbed, s.GateFailed, s.T1Called, s.T1Decided, s.T1Throttled) } diff --git a/internal/runtime/memory/preinsert_t1.go b/internal/runtime/memory/preinsert_t1.go index 1d29be0..cddee1d 100644 --- a/internal/runtime/memory/preinsert_t1.go +++ b/internal/runtime/memory/preinsert_t1.go @@ -34,6 +34,10 @@ import ( // 4. **有界 + 不阻塞写入**：每次审查**最多一次**模型调用（灰区候选打包进同一个 // prompt，最多 `defaultT1MaxGray` 条），且有自己的子超时（≤ 单条审查总预算）。 // 超时 / 报错 / 非法回复 ⇒ `create` + WARN + **零改写**，绝不阻塞、绝不丢记忆。 +// 此外还有一层**全局滑窗限速**（`defaultT1PerWindow` 次 / `defaultT1Window`， +// 全进程共享一个窗口）堵住最坏情形：现网每天数百到上千条新记忆，若大半带灰区 +// 候选就是每天数百到上千次调用；超限 ⇒ 跳过 T1 + 回落 `create` + +// `t1_throttled`（见 preinsert_t1_limit.go）。 // T1 只挂在**异步审查层**（`PreInsertReviewer.Review`）；同步入库闸 // （preinsert_gate.go）不碰它 ⇒ 写入路径仍是零模型调用。 // 5. **不新造 LLM 客户端**：注入的是**既有** `model.Adapter` 的调用口（组合根 @@ -41,10 +45,10 @@ import ( // 厂商协议，也不新增依赖。 // // 可观测（§15 D3-8）：T1 是否参与、结论是什么，在留痕 `reason` 列上如实可辨 -// （`t1_merge` / `t1_modify` / `t1_keep` / `t1_reply_invalid` / `t1_unavailable`）， -// 规则版本随之升到 `IngestRuleVersion`。模型的自由文本理由**既不落库也不落日志**： -// 留痕表按设计不存正文，而模型的理由可能改述正文 ⇒ 一律丢弃（它在 prompt 里的 -// 唯一作用，是逼模型先表态，从而把动作判得更稳）。 +// （`t1_merge` / `t1_modify` / `t1_keep` / `t1_reply_invalid` / `t1_unavailable` / +// `t1_throttled`），规则版本随之升到 `IngestRuleVersion`。模型的自由文本理由 +// **既不落库也不落日志**：留痕表按设计不存正文，而模型的理由可能改述正文 ⇒ 一律 +// 丢弃（它在 prompt 里的唯一作用，是逼模型先表态，从而把动作判得更稳）。 const ( // defaultT1MaxGray 是打包进**一次** prompt 的灰区候选上限。灰区候选按召回 @@ -71,6 +75,11 @@ const ( t1ReasonKeep = "t1_keep" t1ReasonReplyInvalid = "t1_reply_invalid" t1ReasonUnavailable = "t1_unavailable" + // t1ReasonThrottled 是**全局限速**命中时的原因码（preinsert_t1_limit.go， + // design/23 §15.4）。它与上面两个失败码同族、同样只改 `reason`：决定恒 + // `create`、零改写。与它们的唯一区别是语义 —— 失败是「模型没答上」， + // 限流是「这一轮先不判、留给以后」，因此不走失败那条 WARN。 + t1ReasonThrottled = "t1_throttled" ) // ErrT1ReplyInvalid 是模型回复解不出三态枚举（或候选编号越界）时的分类错误。 @@ -118,12 +127,19 @@ func (r *PreInsertReviewer) t1Timeout() time.Duration { // maybeJudgeT1 在「T0 判不出结论 + 有灰区候选」时，用**一次**模型调用改判。 // // 任何非「模型给出合法三态」的情形都**原样返回 v**：未启用、没有模型出口、 -// 没有灰区候选、调用超时/报错、回复非法。因此「失败 ⇒ create + 零改写」是 -// 结构性成立的 —— 这条路径上根本没有改写的代码。 +// 没有灰区候选、**全局额度用尽（限流）**、调用超时/报错、回复非法。因此 +// 「失败/限流 ⇒ create + 零改写」是结构性成立的 —— 这条路径上根本没有改写的 +// 代码。 func (r *PreInsertReviewer) maybeJudgeT1(ctx context.Context, item MemoryItem, v ingestVerdict) ingestVerdict { if !r.t1Enabled() || len(v.Gray) == 0 { return v } + // 全局限速（design/23 §15.4）摆在**发请求之前**，且排在两个前置守卫**之后**： + // 没有灰区候选的审查一次额度都不占（它本来就不调模型），只有真的要花钱时才 + // 过闸、才记账 —— 于是「超限」永远等于「本来会多花一次调用」。 + if allowed, firstRefusal := r.t1Gate.allow(); !allowed { + return r.t1Throttled(item, v, firstRefusal) + } gray := v.Gray if len(gray) > defaultT1MaxGray { gray = gray[:defaultT1MaxGray] @@ -149,6 +165,9 @@ func (r *PreInsertReviewer) maybeJudgeT1(ctx context.Context, item MemoryItem, v // callT1 发起**唯一**的那次 T1 调用（每次审查至多一次）。它自带子超时，并把 // 同一个预算也交给适配器（model.Options.Timeout）—— 进程内与 HTTP 侧各把一道， // 一次灰区判定绝不可能拖长审查。模型正文（prompt 与响应）**不进日志**。 +// +// 走到这里说明额度**已经**被闸放行（放行才记账）：因此「这一条发得出去」与 +// 「这条占掉了一格额度」是同一件事，不存在「调了但没记账」的缝。 func (r *PreInsertReviewer) callT1(ctx context.Context, item MemoryItem, gray []ingestCandidate) (t1Reply, error) { budget := r.t1Timeout() cctx, cancel := context.WithTimeout(ctx, budget) @@ -178,6 +197,30 @@ func (r *PreInsertReviewer) t1Fallback(ctx context.Context, item MemoryItem, v i return v } +// t1Throttled 组装**限流**回落结论：与 t1Fallback 逐字同结果 —— 决定恒 `create`、 +// 只把 `reason` 换成 `t1_throttled`、**零改写**，差分只在外面的口径与日志档位。 +// +// 为什么日志与失败路径不同：失败（超时 / 报错 / 回复非法）是**故障**，每条都该 +// 现形；超限是**预期内的省钱行为**（主人 2026-10-07：「超限不是错误，是这一轮先 +// 不判、留给以后」），一天可能几百条 —— 全记 WARN 会把真正的故障淹掉。因此： +// 每个窗口只记**一条** WARN（附窗口额度与窗口长度，够运维定位），其余记 Debug。 +// 留痕 `reason` 一律如实（那才是机器可查的证据面）。日志只带 id / 作用域 / 额度， +// **绝不带记忆正文**。 +func (r *PreInsertReviewer) t1Throttled(item MemoryItem, v ingestVerdict, firstRefusal bool) ingestVerdict { + v.Reason = t1ReasonThrottled + r.bump(func(s *ReviewStats) { s.T1Throttled++ }) + args := []any{ + "item_id", item.ID, "scope", string(item.Scope), "reason", t1ReasonThrottled, + "t1_per_window", r.cfg.T1PerWindow, "t1_window", r.cfg.T1Window.String(), + } + if firstRefusal { + r.log.Warn("memory ingest review t1 judge throttled, gray-zone item kept as create", args...) + return v + } + r.log.Debug("memory ingest review t1 judge throttled, gray-zone item kept as create", args...) + return v +} + // t1Verdict 把模型回复映射成判定结论。ok=false 表示回复不可用（非法枚举 / // 编号越界），调用方据此回落 create —— **绝不猜、绝不挑一个最近的候选**。 func t1Verdict(reply t1Reply, item MemoryItem, gray []ingestCandidate, base ingestVerdict) (ingestVerdict, bool) { diff --git a/internal/runtime/memory/preinsert_t1_limit.go b/internal/runtime/memory/preinsert_t1_limit.go new file mode 100644 index 0000000..600c052 --- /dev/null +++ b/internal/runtime/memory/preinsert_t1_limit.go @@ -0,0 +1,102 @@ +package memory + +import ( + "sync" + "time" +) + +// 本文件是 T1（灰区模型判定）的**全局限速闸**（design/23 §15.4 的「加一层保守的 +// 全局速率限流」，主人 2026-10-07 裁定）。 +// +// 为什么要有它：T1 默认**开**，而它只在「T0 判不出结论 + 存在灰区候选」时才发一次 +// 模型调用 —— 现网节奏是每天 700–1000 条新记忆，若其中多数都带灰区候选，就是 +// **每天数百到上千次**额外调用。在此之前 T1 上**没有任何上限**：那是一处没有刹车 +// 的额度出口。加的这层闸是**全进程共享的一个窗口额度**（不是每条各自的限制）： +// 超限 ⇒ 跳过 T1 并回落 `create`（reason=`t1_throttled`），与既有失败路径逐字同 +// 语义（**只改 reason、决定恒 create、零改写**）。超限**不是错误**，是「这一轮 +// 先不判、留给以后」—— 记忆一条不丢、一个字不改。 +// +// 为什么自研这一件、而不是复用既有滑窗件 —— 两条都是硬理由（不是图省事）： +// +// 1. **窗口过期必须可确定性测试**。既有共享件是 `session.NewRateLimiter` +// （`internal/channel/session/proactive.go`，design/12 §N C.5 指定的那个）， +// 它内部写死 `time.Now()`，**没有时钟注入缝**：它的「窗口过后额度恢复」只能 +// 靠「极小窗口 + 真实短等」覆盖 —— 那正是 design/12 §N C.5 自己登记的已知 +// 取舍。本切片要求用**注入时钟**验这条通路（与本包 `ReviewDeps.Now` 同一 +// 手法），而 `allow()` 的每一次判定都要看时间，没有缝就做不到。 +// 2. **依赖方向**：那一件住在 `internal/channel/session`，且会带进 +// `internal/channel/adapter`（渠道层的适配器/对话概念）。记忆运行时为 20 行的 +// 计数去依赖渠道层，是拿架构换省事。 +// +// design/12 §N C.5 对同一取舍已经写下了结论：「若主会话要求完全确定，改法是让闸 +// 自带 `now func() time.Time`（约 +15 行，代价是出现第二份滑窗实现）」。本文件 +// 就是那一改法的落地，并**刻意只做这一件事**：它不认识令牌桶、不按 key 分桶， +// 只有一个窗口一份额度 —— 不是「第二套通用限流机制」，是 T1 的一处刹车。 + +// t1RateGate 是 T1 的全局限速闸：一个滑动窗口额度，**全进程共享一个实例** +// （组合根只建一个 `PreInsertReviewer`，闸就挂在那一个实例上 ⇒ 事实上是全进程 +// 额度；AGENTS §10 禁包级全局变量，因此它只能是实例字段）。 +// +// 并发安全：审查可能来自后台工作协程，也可能来自测试/手动 Review，状态全部在 +// 互斥下读写。 +type t1RateGate struct { + limit int + window time.Duration + now func() time.Time + + mu sync.Mutex + hits []time.Time + // refused 记账「本窗口已经拒过一次」，只为把日志收敛成**每窗口一条 WARN**： + // 超限是预期内的省钱行为、不是故障，一天几百条 WARN 会把真正的故障淹掉。 + refused bool +} + +// newT1RateGate 构建闸。limit / window **必须是 normalize 之后的正值**：非正会 +// 得到一个恒放行的闸（见 allow），那正是 design/12 §N C.5 G3 登记过的静默失败 +// 形态「≤0 被读成无上限」—— 因此调用方（NewPreInsertReviewer）保证它在 +// `normalizeReviewConfig` **之后**构建，绝不让未归一的 ≤0 直达滑窗。 +func newT1RateGate(limit int, window time.Duration, now func() time.Time) *t1RateGate { + if now == nil { + now = time.Now + } + return &t1RateGate{limit: limit, window: window, now: now} +} + +// allow 报告这一次 T1 调用是否放行；第二个返回值是「本窗口的第一次拒绝」， +// 调用方据此把日志收成每窗口一条 WARN、其余记 Debug。 +// +// 额度口径与既有滑窗件一致：**放行才消耗** —— 被拒的那次不记账，因此超限不会把 +// 窗口往后推，额度到点（最后一次放行的窗口之后）自然恢复。 +// +// nil 或非正额度 ⇒ 恒放行：那是「没配限流器」的安全默认，而不是「静默关掉一处 +// 防护」（构造处已由 normalize 保证不会发生）。 +func (g *t1RateGate) allow() (allowed bool, firstRefusal bool) { + if g == nil || g.limit <= 0 || g.window <= 0 { + return true, false + } + now := g.now() + cutoff := now.Add(-g.window) + + g.mu.Lock() + defer g.mu.Unlock() + + keep := 0 + for _, t := range g.hits { + if t.After(cutoff) { + g.hits[keep] = t + keep++ + } + } + g.hits = g.hits[:keep] + if keep == 0 { + // 窗口已排空 ⇒ 新一轮开始，「本窗口已报过一次 WARN」随之复位。 + g.refused = false + } + if keep >= g.limit { + first := !g.refused + g.refused = true + return false, first + } + g.hits = append(g.hits, now) + return true, false +} diff --git a/internal/runtime/memory/preinsert_t1_limit_test.go b/internal/runtime/memory/preinsert_t1_limit_test.go new file mode 100644 index 0000000..81b0d2a --- /dev/null +++ b/internal/runtime/memory/preinsert_t1_limit_test.go @@ -0,0 +1,412 @@ +package memory + +import ( + "bytes" + "context" + "log/slog" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/embedding" +) + +// 本文件是 **T1 全局限速**（design/23 §15.4，preinsert_t1_limit.go）的测试： +// 超限即不调模型、额度只被真调用消耗、窗口过后恢复、限流路径零改写、日志收敛 +// （每窗口一条 WARN），以及「≤0 不许被读成无上限」那条归一化红线。 +// +// 三态落地 / 失败回落 / 开关口径在 preinsert_t1_test.go，调用次数有界与硬门隔离 +// 在 preinsert_t1_bound_test.go（本文件不动它们，只加限流这一层）。 +// +// 所有用例都**不**靠真实等待：窗口过期由可控时钟推进（下面 t1Clock），与 +// t1JudgeStub「被调用即 panic」是同一思路 —— 把「不该发生的调用」变成立刻现形。 + +// t1Clock 是可推进的墙钟：限流的「窗口过后额度恢复」必须由它驱动，真实等待会让 +// 测试变慢且不确定（窗口是墙钟量，不是逻辑量）。 +type t1Clock struct { + mu sync.Mutex + at time.Time +} + +func newT1Clock(at time.Time) *t1Clock { return &t1Clock{at: at} } + +func (c *t1Clock) now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.at +} + +func (c *t1Clock) advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.at = c.at.Add(d) +} + +// newT1LimiterReviewer 构建「异步审查层 + 已接线 T1 + 可控时钟」的服务： +// 同步入库闸显式关掉（这批用例只验异步那一层，否则测到的是闸），其余走默认值； +// 限流额度由 cfg.T1PerWindow / cfg.T1Window 给出（非正即回落代码默认）。 +func newT1LimiterReviewer( + t *testing.T, store *Store, emb embedding.Embedder, cfg ReviewConfig, + judge T1Judge, now func() time.Time, log *slog.Logger, +) *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 + } + cfg.DisableIngestGate = true + if log == nil { + log = testLogger() + } + r, err := NewPreInsertReviewer(cfg, ReviewDeps{ + Store: store, Embedder: emb, Judge: judge, Now: now, Log: log, + }) + if err != nil { + t.Fatalf("NewPreInsertReviewer: %v", err) + } + return r +} + +// TestT1ThrottledSkipsModelCallAndFallsBackToCreate 是任务书 §5①：额度用尽后 +// **不再发起模型调用**，且结论仍是 `create` + `reason=t1_throttled`（与既有失败 +// 路径同语义：只改 reason、决定恒 create、零改写）。 +// +// 桩回 `keep` 是刻意的：灰区条目因此每次都留在册，同一条可以被反复审 —— 于是 +// 「第 N 次」这件事只由额度决定，与判定结果无关。 +func TestT1ThrottledSkipsModelCallAndFallsBackToCreate(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + _, fresh := seedT1GrayPair(t, store) + + judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`} + r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), + ReviewConfig{T1PerWindow: 2, T1Window: time.Hour}, judge.judge, nil, nil) + + for i := 1; i <= 2; i++ { + rev, err := r.Review(ctx, fresh) + if err != nil { + t.Fatalf("第 %d 次 Review: %v", i, err) + } + if rev.Reason != t1ReasonKeep { + t.Fatalf("额度内第 %d 次的 reason = %q, want %q", i, rev.Reason, t1ReasonKeep) + } + } + if judge.count() != 2 { + t.Fatalf("额度内模型调用 = %d 次, want 2", judge.count()) + } + + rev, err := r.Review(ctx, fresh) + if err != nil { + t.Fatalf("超限那次 Review: %v", err) + } + if judge.count() != 2 { + t.Fatalf("超限后仍调了模型：共 %d 次, want 仍为 2", judge.count()) + } + if rev.Decision != IngestCreate { + t.Fatalf("超限后决定 = %q, want create（限流不是报错，是这一轮先不判）", rev.Decision) + } + if rev.Reason != t1ReasonThrottled { + t.Fatalf("超限后 reason = %q, want %q", rev.Reason, t1ReasonThrottled) + } + st := r.Stats() + if st.T1Called != 2 || st.T1Decided != 2 || st.T1Throttled != 1 { + t.Fatalf("计数 = %+v, want T1Called=2 T1Decided=2 T1Throttled=1", st) + } + if label := reviewStatsLabel(st); !strings.Contains(label, "t1_throttled=1") { + t.Fatalf("统计串里看不到限流：%q", label) + } +} + +// TestT1ThrottleRecoversAfterWindow 是任务书 §5②：额度在窗口过后**恢复** +// （限流不把窗口往后推）。时钟由注入的 t1Clock 推进，不靠真实等待。 +func TestT1ThrottleRecoversAfterWindow(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + _, fresh := seedT1GrayPair(t, store) + + clock := newT1Clock(tidyTestTime(t)) + judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`} + window := 30 * time.Minute + r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), + ReviewConfig{T1PerWindow: 1, T1Window: window}, judge.judge, clock.now, nil) + + if rev, err := r.Review(ctx, fresh); err != nil || rev.Reason != t1ReasonKeep { + t.Fatalf("窗口内首次 = %+v err=%v, want keep", rev, err) + } + if rev, err := r.Review(ctx, fresh); err != nil || rev.Reason != t1ReasonThrottled { + t.Fatalf("窗口内第二次 = %+v err=%v, want %q", rev, err, t1ReasonThrottled) + } + + // 差一秒钟到期：仍然超限（边界口径与既有滑窗件逐字相同：严格晚于 cutoff 才算在窗内）。 + clock.advance(window - time.Second) + if rev, err := r.Review(ctx, fresh); err != nil || rev.Reason != t1ReasonThrottled { + t.Fatalf("窗口未满却恢复了额度：%+v err=%v", rev, err) + } + if judge.count() != 1 { + t.Fatalf("窗口未满时模型调用 = %d 次, want 1", judge.count()) + } + + // 推进过窗口：额度恢复，模型再次被调用。 + clock.advance(2 * time.Second) + rev, err := r.Review(ctx, fresh) + if err != nil { + t.Fatalf("窗口过后 Review: %v", err) + } + if judge.count() != 2 { + t.Fatalf("窗口过后模型调用 = %d 次, want 2（额度必须恢复）", judge.count()) + } + if rev.Reason != t1ReasonKeep { + t.Fatalf("窗口过后的 reason = %q, want %q", rev.Reason, t1ReasonKeep) + } + if st := r.Stats(); st.T1Throttled != 2 { + t.Fatalf("超限计数 = %d, want 2", st.T1Throttled) + } +} + +// TestT1ThrottledNeverTouchesMemory 是任务书 §5③：限流路径**只改 reason**， +// 记忆行一个字不改（与 TestReviewFailureNeverTouchesMemory 同一手法）。 +// +// 守卫手法照 TestT1DisabledMakesZeroModelCalls：把桩换成「被调用即 panic」并回 +// `merge` —— 限流一旦失效，要么当场 panic，要么真把两行改写掉，两种都在断言里现形。 +func TestT1ThrottledNeverTouchesMemory(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + existing, fresh := seedT1GrayPair(t, store) + + judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`} + r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), + ReviewConfig{T1PerWindow: 1, T1Window: time.Hour}, judge.judge, nil, nil) + + // 第一次：额度内，桩回 keep（不改写任何行）。 + if _, err := r.Review(ctx, fresh); err != nil { + t.Fatalf("额度内 Review: %v", err) + } + if judge.count() != 1 { + t.Fatalf("额度内模型调用 = %d 次, want 1", judge.count()) + } + + // 第二次：额度已用尽。桩换成「被调用即 panic」并回 merge。 + judge.mu.Lock() + judge.panics = true + judge.reply = `{"action":"merge","target":1}` + judge.mu.Unlock() + + rev := t1ReviewGuarded(t, r, fresh) + if rev.Decision != IngestCreate || rev.Reason != t1ReasonThrottled { + t.Fatalf("超限那次 = %+v, want create + %q", rev, t1ReasonThrottled) + } + if got := readReviewRow(t, store, fresh.ID); got.status != tidyStatusActive || + got.mergedInto != "" || got.content != fixtureT1CodeViaChannel { + t.Fatalf("限流却改写了新行：%+v", got) + } + old := readReviewRow(t, store, existing.ID) + if old.status != tidyStatusActive || old.mergeCount != 1 || old.content != fixtureT1CodeAgreed { + t.Fatalf("限流却改了既有行：%+v", old) + } + if n := countActiveRows(t, store, "persona.a"); n != 2 { + t.Fatalf("活跃行数 = %d, want 2（限流是「不调」，不是「调失败」）", n) + } +} + +// TestT1ThrottleQuotaIsOnlySpentByRealCalls 钉住「额度只被真调用消耗」： +// 没有灰区候选的审查（本来就不调模型）一次额度都不占 —— 否则整库的普通新建 +// 会把 T1 的额度白白吃光，限流就成了随机停机。 +func TestT1ThrottleQuotaIsOnlySpentByRealCalls(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + base := tidyTestTime(t) + _, fresh := seedT1GrayPair(t, store) + // 一条与灰区对**毫不相似**的条目：它落不到任何候选上 ⇒ 零灰区、零调用。 + plain := seedTidyItem(t, store, habitItem(fixtureHabitVacation, base.Add(2*time.Hour), 0.9)) + + judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`} + r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), + ReviewConfig{T1PerWindow: 1, T1Window: time.Hour}, judge.judge, nil, nil) + + rev, err := r.Review(ctx, plain) + if err != nil { + t.Fatalf("零灰区那次 Review: %v", err) + } + if rev.Decision != IngestCreate || judge.count() != 0 { + t.Fatalf("零灰区那次 = %+v calls=%d，不该调模型", rev, judge.count()) + } + if st := r.Stats(); st.T1Throttled != 0 { + t.Fatalf("零灰区那次被记成限流：%+v", st) + } + + // 额度仍然完好在册：真需要判定的这一条照样调得到模型。 + if rev, err := r.Review(ctx, fresh); err != nil || rev.Reason != t1ReasonKeep { + t.Fatalf("灰区那条 = %+v err=%v, want keep（额度被普通新建吃掉了）", rev, err) + } + if judge.count() != 1 { + t.Fatalf("模型调用 = %d 次, want 1", judge.count()) + } +} + +// TestT1ThrottleLogsOneWarnPerWindow 钉住日志口径：超限是**预期内的省钱行为**， +// 不是故障 ⇒ 每窗口只记一条 WARN，其余记 Debug；且日志里绝不出现记忆正文。 +func TestT1ThrottleLogsOneWarnPerWindow(t *testing.T) { + t.Parallel() + ctx := context.Background() + store := newTestStore(t) + _, fresh := seedT1GrayPair(t, store) + + var buf bytes.Buffer + log := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})) + judge := &t1JudgeStub{reply: `{"action":"keep","target":0}`} + limit := ReviewConfig{T1PerWindow: 1, T1Window: time.Hour} + r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), limit, judge.judge, nil, log) + + // 三次审查、额度 1：第 1 次真调、第 2 次超限（本窗口第一条 WARN）、 + // 第 3 次仍超限（降级成 Debug）—— 日志条数 2、其中 WARN 1。 + for i := 0; i < 3; i++ { + if _, err := r.Review(ctx, fresh); err != nil { + t.Fatalf("第 %d 次 Review: %v", i, err) + } + } + logged := buf.String() + if n := strings.Count(logged, "t1 judge throttled"); n != 2 { + t.Fatalf("限流日志条数 = %d, want 2（两次超限各一条）：%q", n, logged) + } + if n := strings.Count(logged, "level=WARN"); n != 1 { + t.Fatalf("限流 WARN 条数 = %d, want 1（每窗口一条）：%q", n, logged) + } + for _, leaked := range []string{fixtureT1CodeAgreed, fixtureT1CodeViaChannel} { + if strings.Contains(logged, leaked) { + t.Fatalf("日志里泄漏了记忆正文 %q：%q", leaked, logged) + } + } + + // 窗口排空之后「本窗口已报过」必须复位：新窗口的第一次超限仍然要 WARN。 + clock := newT1Clock(tidyTestTime(t)) + r2 := newT1LimiterReviewer(t, store, t1GrayEmbedder(), limit, judge.judge, clock.now, log) + for i := 0; i < 2; i++ { + if _, err := r2.Review(ctx, fresh); err != nil { + t.Fatalf("新一轮第 %d 次 Review: %v", i, err) + } + } + if n := strings.Count(buf.String(), "level=WARN"); n != 2 { + t.Fatalf("新窗口的 WARN 条数 = %d, want 2：%q", n, buf.String()) + } + clock.advance(2 * time.Hour) + for i := 0; i < 2; i++ { + if _, err := r2.Review(ctx, fresh); err != nil { + t.Fatalf("过期轮第 %d 次 Review: %v", i, err) + } + } + if n := strings.Count(buf.String(), "level=WARN"); n != 3 { + t.Fatalf("跨窗口后的 WARN 条数 = %d, want 3（新窗口要重新报一次）：%q", n, buf.String()) + } +} + +// TestT1DisabledLimitsNothingAndCallsNothing 是任务书 §5④：`t1Judge: false` +// （DisableT1）时**零模型调用**，限流器连碰都不碰 —— 既有那条「被调用即 panic」 +// 的守卫测试（TestT1DisabledMakesZeroModelCalls）因此不受本切片影响。 +func TestT1DisabledLimitsNothingAndCallsNothing(t *testing.T) { + t.Parallel() + store := newTestStore(t) + _, fresh := seedT1GrayPair(t, store) + + judge := &t1JudgeStub{panics: true, reply: `{"action":"merge","target":1}`} + r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), + ReviewConfig{DisableT1: true, T1PerWindow: 1, T1Window: time.Hour}, judge.judge, nil, nil) + + rev := t1ReviewGuarded(t, r, fresh) + if judge.count() != 0 { + t.Fatalf("T1 关掉后仍调了 %d 次模型", judge.count()) + } + if rev.Decision != IngestCreate || rev.Reason != "no_contained_match" { + t.Fatalf("T1 关掉后 = %+v, want create + T0 原始原因码（限流不该改这里的语义）", rev) + } + if st := r.Stats(); st.T1Called != 0 || st.T1Decided != 0 || st.T1Throttled != 0 { + t.Fatalf("T1 关掉后计数 = %+v, want 全零", st) + } +} + +// TestT1RateGateIsConcurrencySafe 钉住闸的**并发安全**：审查可能来自后台工作 +// 协程，也可能来自手动 Review。N 个协程同时抢 K 个额度 ⇒ **恰好**放行 K 个 —— +// 多放行就是超发模型调用（额度失效），少放行就是并发把额度吃掉（额度虚耗）。 +func TestT1RateGateIsConcurrencySafe(t *testing.T) { + t.Parallel() + const ( + limit = 7 + workers = 32 + ) + gate := newT1RateGate(limit, time.Hour, newT1Clock(tidyTestTime(t)).now) + + var ( + wg sync.WaitGroup + mu sync.Mutex + allowed int + ) + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if ok, _ := gate.allow(); ok { + mu.Lock() + allowed++ + mu.Unlock() + } + }() + } + wg.Wait() + if allowed != limit { + t.Fatalf("并发放行 = %d, want 恰为 %d", allowed, limit) + } +} + +// TestNormalizeReviewConfigFillsT1ThrottleDefaults 是 design/12 §N C.5 G3 那条 +// 红线的探针：额度**非正**必须被 normalize 回落成默认额 —— 绝不许带着 ≤0 进闸， +// 否则闸的「恒放行」会把「回落默认」静默变成「无上限」（构造处已保证闸在归一之后建）。 +func TestNormalizeReviewConfigFillsT1ThrottleDefaults(t *testing.T) { + t.Parallel() + for _, cfg := range []ReviewConfig{ + {}, + {T1PerWindow: -1, T1Window: -time.Second}, + {T1PerWindow: 7}, + {T1Window: 3 * time.Minute}, + } { + got := normalizeReviewConfig(cfg) + if got.T1PerWindow <= 0 { + t.Fatalf("normalize(%+v).T1PerWindow = %d，非正值直达闸会被读成「无上限」", cfg, got.T1PerWindow) + } + if got.T1Window <= 0 { + t.Fatalf("normalize(%+v).T1Window = %s，同上", cfg, got.T1Window) + } + } + if got := normalizeReviewConfig(ReviewConfig{}); got.T1PerWindow != defaultT1PerWindow || + got.T1Window != defaultT1Window { + t.Fatalf("零值配置 = %d/%s, want %d/%s", + got.T1PerWindow, got.T1Window, defaultT1PerWindow, defaultT1Window) + } + if got := normalizeReviewConfig(ReviewConfig{T1PerWindow: 7, T1Window: 3 * time.Minute}); got.T1PerWindow != 7 || + got.T1Window != 3*time.Minute { + t.Fatalf("显式额度被改写：%d/%s", got.T1PerWindow, got.T1Window) + } + + // 构造后的生效值也必须已是默认额（而不是留 0 等运行时兜）。 + store := newTestStore(t) + r := newT1LimiterReviewer(t, store, t1GrayEmbedder(), ReviewConfig{}, nil, nil, nil) + if r.cfg.T1PerWindow != defaultT1PerWindow || r.cfg.T1Window != defaultT1Window { + t.Fatalf("NewPreInsertReviewer 的生效额度 = %d/%s, want 默认 %d/%s", + r.cfg.T1PerWindow, r.cfg.T1Window, defaultT1PerWindow, defaultT1Window) + } + if allowed, first := r.t1Gate.allow(); !allowed || first { + t.Fatal("默认额度下第一次就该放行") + } +}