package backthink import ( "context" "errors" "strconv" "strings" "sync" "testing" "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/restrict" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" ) // 本文件守 design/12 §N C 的**自排本体**(切片 C1-2):WakeNow 的语义(探槽 → // 过闸 → 入槽,同锁)、容量 1 的待跑槽、每窗口上限、worker、审计四条 what。 // 对应 C.8 判据 2 / 3 / 9 / 10 / 11 / 13 / 14。 // // ⚠️ 本片交付的是**服务侧**能力:她**还调不到**(工具面 `request_think` 属 C1-3) // ⇒ 不得表述成「C 已完成」。 // wakeThinker 是自排用例用的假 Thinker:它在**进入**每一轮时记下触发串并发出信号, // 可选地「卡住」自排轮(holdSelfWake 未关闭时阻塞),让测试能构造「worker 正卡在 // 一轮里」这个对撞场景。它尊重 ctx 取消,好让 Stop 能排空挂住的轮次。 // // 为什么信号点在**进入**而不是完成:worker 在调用 Think 之前就已把待跑槽取空 // (C.1 的 take → run),因此「某轮已进入」就是「槽已空」的同步点 —— 用它才能确定地 // 测到窗口闸,而不是撞上「槽还满着」。卡住的轮次计入 triggers,但永不返回。 type wakeThinker struct { mu sync.Mutex triggers []string // holdSelfWake 非 nil 时,自排轮阻塞到它被关闭(启动轮/心跳轮不受影响: // 它们跑在 tickLoop 上,不占自排 worker)。 holdSelfWake chan struct{} entered chan struct{} } func newWakeThinker() *wakeThinker { return &wakeThinker{entered: make(chan struct{}, 64)} } func (w *wakeThinker) Think(ctx context.Context, sessionID, trigger string) (string, error) { if err := ctx.Err(); err != nil { return "", err } w.mu.Lock() w.triggers = append(w.triggers, trigger) w.mu.Unlock() select { case w.entered <- struct{}{}: default: } if w.holdSelfWake != nil && strings.HasPrefix(trigger, SelfWakeSummaryPrefix) { select { case <-w.holdSelfWake: case <-ctx.Done(): return "", ctx.Err() } } return "no push decided", nil } // selfWakeRounds 报告已进入的自排轮数(按 SelfWakeSummaryPrefix 认轮,启动轮与 // 心跳轮不计)。 func (w *wakeThinker) selfWakeRounds() int { w.mu.Lock() defer w.mu.Unlock() n := 0 for _, tr := range w.triggers { if strings.HasPrefix(tr, SelfWakeSummaryPrefix) { n++ } } return n } // total 报告已进入的一轮总数(含启动轮与被卡住的那一轮)。 func (w *wakeThinker) total() int { w.mu.Lock() defer w.mu.Unlock() return len(w.triggers) } // selfWakeSummaries 返回已进入的自排轮触发摘要。它让用例能断言「合并之后这一轮用的是 // **先到**的理由」—— 理由只出现在触发摘要里(落库行只存 Trigger = Source,不存摘要)。 func (w *wakeThinker) selfWakeSummaries() []string { w.mu.Lock() defer w.mu.Unlock() out := make([]string, 0, len(w.triggers)) for _, tr := range w.triggers { if strings.HasPrefix(tr, SelfWakeSummaryPrefix) { out = append(out, tr) } } return out } // waitFor 等到至少 want 轮**已进入**(或超时失败)。它是「槽已空、该轮已在跑」的 // 同步点:保证测的是**闸**,而不是「槽还满着 / worker 还没取槽」。 func (w *wakeThinker) waitFor(t *testing.T, want int, timeout time.Duration) { t.Helper() deadline := time.Now().Add(timeout) for time.Now().Before(deadline) { if w.total() >= want { return } select { case <-w.entered: case <-time.After(5 * time.Millisecond): } } t.Fatalf("等待第 %d 轮进入超时(已进入 %d 轮)", want, w.total()) } // fakeAuditor 记录审计三元组。 type fakeAuditor struct { mu sync.Mutex rows []auditRow } type auditRow struct{ who, what, result string } func (a *fakeAuditor) AuditEvent(who, what, result string) { a.mu.Lock() defer a.mu.Unlock() a.rows = append(a.rows, auditRow{who: who, what: what, result: result}) } // whats 返回按顺序记录的 what 列表。 func (a *fakeAuditor) whats() []string { a.mu.Lock() defer a.mu.Unlock() out := make([]string, 0, len(a.rows)) for _, r := range a.rows { out = append(out, r.what) } return out } // resultOf 返回第一条匹配 what 的 result(没记到则失败)。 func (a *fakeAuditor) resultOf(t *testing.T, what string) string { t.Helper() a.mu.Lock() defer a.mu.Unlock() for _, r := range a.rows { if r.what == what { return r.result } } t.Fatalf("审计里没有 %q:%+v", what, a.rows) return "" } // countOf 返回某个 what 被记了几条。 func (a *fakeAuditor) countOf(what string) int { a.mu.Lock() defer a.mu.Unlock() n := 0 for _, r := range a.rows { if r.what == what { n++ } } return n } func (a *fakeAuditor) len() int { a.mu.Lock() defer a.mu.Unlock() return len(a.rows) } // ownerCtx 是主人环境(工具面与 WakeNow 的第二道身份闸都读它)。 func ownerCtx() context.Context { return loop.WithSessionEnv(context.Background(), loop.OwnerEnv()) } // startSelfWake 构建并启动一个后台思考服务,返回它、假 Thinker 与假审计出口。 // 它**不**等服务启动轮跑完:启动轮恒占 Think 调用 1 次(计数基线),而 w.hold 非 nil 的 // 对撞用例正是要让它卡在那里 —— 等待它反而会死等。自排计数按 SelfWakeSummaryPrefix // 认轮,因此启动轮混在总数里不影响断言。 func startSelfWake(t *testing.T, cfg Config, w *wakeThinker) (*Backthink, *fakeAuditor) { t.Helper() audit := &fakeAuditor{} b, _ := newBackthink(t, cfg, Deps{ Thinker: w, State: &mockState{state: UserState{}}, Audit: audit, Log: discardLog(), }) b.Start(context.Background()) t.Cleanup(func() { _ = b.Stop(context.Background()) }) return b, audit } // TestWakeNow_UnavailableBeforeStart 是 C.8 判据 10(诚实):未 Start 时 WakeNow // 必须返回 ErrSelfWakeUnavailable,**绝不假装排上**(返回 nil + 「已排队」)。 // // 变异:未 Start 时返回 `SelfWakeReceipt{Accepted: true}, nil` ⇒ 本用例变红。 func TestWakeNow_UnavailableBeforeStart(t *testing.T) { t.Parallel() b, _ := newBackthink(t, Config{}, Deps{Thinker: newWakeThinker(), State: &mockState{}}) receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "还没启动"}) if !errors.Is(err, ErrSelfWakeUnavailable) { t.Fatalf("err = %v, want ErrSelfWakeUnavailable", err) } if receipt.Accepted { t.Fatalf("receipt = %+v, want 未被接受(不得假装排上)", receipt) } } // TestWakeNow_UnavailableAfterStop 钉住 Stop 之后同样不可用(进程在收尾,不许再排)。 func TestWakeNow_UnavailableAfterStop(t *testing.T) { t.Parallel() w := newWakeThinker() b, _ := startSelfWake(t, Config{}, w) if err := b.Stop(context.Background()); err != nil { t.Fatalf("Stop: %v", err) } if _, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "太晚了"}); !errors.Is(err, ErrSelfWakeUnavailable) { t.Fatalf("err = %v, want ErrSelfWakeUnavailable", err) } } // TestWakeNow_NotOwnerRefused 是 C.8 判据 7 的服务侧一半(工具面的那一半属 C1-3): // 非主人环境一律拒绝,且**不做任何事**(不消耗额度、不入槽)。 func TestWakeNow_NotOwnerRefused(t *testing.T) { t.Parallel() w := newWakeThinker() b, audit := startSelfWake(t, Config{}, w) ctx := loop.WithSessionEnv(context.Background(), loop.SessionEnv{ActorEnv: restrict.ActorGroupMember}) receipt, err := b.WakeNow(ctx, SelfWakeRequest{Reason: "群里想插一句"}) if !errors.Is(err, ErrSelfWakeNotOwner) { t.Fatalf("err = %v, want ErrSelfWakeNotOwner", err) } if receipt.Accepted { t.Fatalf("receipt = %+v, want 未被接受", receipt) } if got := b.selfWake.pendingDepth(); got != 0 { t.Fatalf("待跑槽占用 = %d, want 0(拒绝时不得占槽)", got) } time.Sleep(50 * time.Millisecond) if got := w.selfWakeRounds(); got != 0 { t.Fatalf("自排轮数 = %d, want 0", got) } if got := audit.countOf(whatSelfWakeDenied); got != 1 { t.Fatalf("denied 审计 = %d 条, want 1(whats=%v)", got, audit.whats()) } if !strings.Contains(audit.resultOf(t, whatSelfWakeDenied), string(restrict.ActorGroupMember)) { t.Fatalf("denied result = %q, want 它带上 actor", audit.resultOf(t, whatSelfWakeDenied)) } } // TestWakeNow_WindowLimitIsTheGate 是 C.8 判据 2:**每次等该轮真跑完再发下一次** // (保证槽是空的 ⇒ 测的是窗口闸而不是槽),limit=2 时第 3 次被拒、只跑 2 轮。 // // 变异:把闸拆掉(恒放行)⇒ 第 3 次仍 Accepted=true ⇒ 本用例变红。 func TestWakeNow_WindowLimitIsTheGate(t *testing.T) { t.Parallel() w := newWakeThinker() b, audit := startSelfWake(t, Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, w) for i := 1; i <= 2; i++ { receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "第几次自排"}) if err != nil { t.Fatalf("第 %d 次 WakeNow: %v", i, err) } if !receipt.Accepted || receipt.Coalesced { t.Fatalf("第 %d 次收据 = %+v, want 排上一轮", i, receipt) } if receipt.Used != i || receipt.Limit != 2 || receipt.WindowSec != 3600 { t.Fatalf("第 %d 次收据 = %+v, want Used=%d Limit=2 WindowSec=3600", i, receipt, i) } w.waitFor(t, 1+i, 3*time.Second) // 槽必须空,下一次才测得到闸 } receipt, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "再来一次"}) if err != nil { t.Fatalf("第 3 次 WakeNow: %v", err) } if receipt.Accepted || receipt.Coalesced { t.Fatalf("第 3 次收据 = %+v, want 被上限拒绝", receipt) } if receipt.Used != 2 || receipt.Limit != 2 { t.Fatalf("第 3 次收据 = %+v, want Used==Limit==2(拒绝不占额度)", receipt) } if receipt.Reason == "" { t.Fatal("拒绝必须给可显示的文案(她要看得见拒绝才会停手)") } time.Sleep(50 * time.Millisecond) if got := w.selfWakeRounds(); got != 2 { t.Fatalf("自排轮数 = %d, want 2(被拒的那次不产生轮次)", got) } if got := audit.countOf(whatSelfWakeCapped); got != 1 { t.Fatalf("capped 审计 = %d 条, want 1(whats=%v)", got, audit.whats()) } if got := audit.countOf(whatSelfWakeQueued); got != 2 { t.Fatalf("queued 审计 = %d 条, want 2(whats=%v)", got, audit.whats()) } } // TestWakeNow_CoalescesIntoPendingSlot 是 C.8 判据 3 + 13 的**可确定**版本:先把 // worker 钉在一轮自排里(假 Thinker 只卡自排轮,启动轮在 tickLoop 上、不占 worker), // 再连发 3 次 —— 第 1 次入槽、第 2/3 次**合并**,而合并**一次都不调用** Allow() // (它成功即消耗 ⇒ 碰一下就白扣额度)。 // // 与 C.8 判据 13 字面的一处**偏离(如实登记)**:它写「①queued 的 Used=1」,那是把 // 「卡住的那一轮」当成**不消耗额度**的启动轮来算的;可启动轮跑在 tickLoop 上,**不占** // 自排 worker(C.1 明确心跳/事件轮可与自排并发)⇒ 钉住 worker 的那一轮必然是自排轮、 // 必然已消耗 1 次额度。因此本用例把「合并扣不扣额度」钉在 **Used 恒为钉住后的值** 上, // 探针性质与判据 13 相同且更锋利(额度只有 2)。详见 `C1-2-report.md`。 // // 变异:① 把过闸挪到探槽**之前**(G1 顺序颠倒)⇒ 第 3 次白扣额度后 capped ⇒ 红; // // ② 待跑槽改无界(每次都入队)⇒ 第 2/3 次不接受合并且跑出 3 轮 ⇒ 红。 func TestWakeNow_CoalescesIntoPendingSlot(t *testing.T) { t.Parallel() w := newWakeThinker() w.holdSelfWake = make(chan struct{}) // 钉住 worker(只卡自排轮) b, audit := startSelfWake(t, Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, w) // ① 先排一轮并等 worker 真的把它取走、卡在里面(槽因此是空的)。 if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "钉住 worker 的那一轮"}); err != nil || !got.Accepted { t.Fatalf("钉住用的 WakeNow = %+v err = %v, want 排上", got, err) } w.waitFor(t, 2, 3*time.Second) // 启动轮 + 被钉住的自排轮 if got := w.selfWakeRounds(); got != 1 { t.Fatalf("自排轮数 = %d, want 1(worker 已取槽并卡在该轮里)", got) } // ② 连发 3 次:第 1 次入槽(额度剩 1 ⇒ Used=2),第 2/3 次合并。 first, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "先到的那一轮"}) if err != nil { t.Fatalf("WakeNow: %v", err) } if !first.Accepted || first.Coalesced || first.Used != 2 { t.Fatalf("入槽的那次收据 = %+v, want 排队且 Used=2", first) } for i, reason := range []string{"第二条理由", "第三条理由"} { got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: reason}) if err != nil { t.Fatalf("WakeNow #%d: %v", i+2, err) } if !got.Accepted || !got.Coalesced { t.Fatalf("第 %d 次收据 = %+v, want 与在队的那一轮合并", i+2, got) } if got.Used != 2 { t.Fatalf("第 %d 次收据 = %+v, want Used 恒 2(合并不消耗预算)", i+2, got) } } if got := audit.countOf(whatSelfWakeCapped); got != 0 { t.Fatalf("capped 审计 = %d 条, want 0(槽满只合并,不拒绝;whats=%v)", got, audit.whats()) } if got := audit.countOf(whatSelfWakeCoalesced); got != 2 { t.Fatalf("coalesced 审计 = %d 条, want 2(whats=%v)", got, audit.whats()) } close(w.holdSelfWake) // 放行:被钉住的那一轮跑完,worker 再取走合并后的那一轮 w.waitFor(t, 3, 3*time.Second) if got := w.selfWakeRounds(); got != 2 { t.Fatalf("自排轮数 = %d, want 2(被钉住的 + 合并后的那一轮)", got) } if got := w.total(); got != 3 { t.Fatalf("Think 总调用 = %d, want 3(启动轮 + 被钉住的 + 合并后的一轮)", got) } // 在队的那一轮必须用**先到**的理由(C.1 步骤 2):后两条理由一个字都不该出现。 summaries := w.selfWakeSummaries() want := SelfWakeSummaryPrefix + "先到的那一轮" if len(summaries) != 2 || summaries[1] != want { t.Fatalf("自排轮摘要 = %#v, want 第二条(合并后的那一轮)是 %q", summaries, want) } } // TestSelfWakeGate_MergeNeverTouchesGate 是 G1(C.8 判据 13 的**确定**探针):直接对闸 // 连发 3 次(不启动 worker ⇒ 槽不会被取走)—— 槽满时**一次都不调用** Allow(),因此 // 「本窗口已用」在三次请求之间恒定。若把过闸挪到探槽之前,第 2 次就会白扣一次额度 // ⇒ Used 变成 2、第 3 次直接 capped ⇒ 本用例变红。 // // 它同时覆盖容量 1:连发 3 次只得到 **1 个**待跑位(第 2/3 次合并),没有无界堆积。 func TestSelfWakeGate_MergeNeverTouchesGate(t *testing.T) { t.Parallel() gate := newSelfWakeGate(2, time.Hour) first, queued := gate.offer(SelfWakeRequest{Reason: "先到"}) if !queued || !first.Accepted || first.Coalesced || first.Used != 1 { t.Fatalf("第 1 次 offer = %+v queued=%v, want 入槽且 Used=1", first, queued) } for i := 0; i < 2; i++ { got, queued := gate.offer(SelfWakeRequest{Reason: "后来的"}) if queued { t.Fatalf("第 %d 次 offer 又入槽了:容量 1 被打破(%+v)", i+2, got) } if !got.Accepted || !got.Coalesced || got.Used != 1 { t.Fatalf("第 %d 次 offer = %+v, want 合并且 Used 恒 1", i+2, got) } } if got := gate.pendingDepth(); got != 1 { t.Fatalf("待跑槽占用 = %d, want 1", got) } // worker 取走槽位后:额度还剩 1 ⇒ 下一次入槽(Used=2),再下一次槽满 ⇒ 仍合并。 if _, ok := gate.take(); !ok { t.Fatal("take 应取到在队的那一轮") } again, queued := gate.offer(SelfWakeRequest{Reason: "取走后第二次"}) if !queued || again.Used != 2 { t.Fatalf("取走后再 offer = %+v queued=%v, want 入槽且 Used=2", again, queued) } if got, _ := gate.offer(SelfWakeRequest{Reason: "再次满槽"}); !got.Coalesced || got.Used != 2 { t.Fatalf("满槽时 offer = %+v, want 合并且 Used=2", got) } // 额度用尽且槽空 ⇒ 拒绝,且**不占额度**(Allow() 返回 false 本就未消耗)。 if _, ok := gate.take(); !ok { t.Fatal("take 应取到第二次入槽的那一轮") } capped, queued := gate.offer(SelfWakeRequest{Reason: "超额"}) if queued || capped.Accepted || capped.Coalesced { t.Fatalf("超额 offer = %+v queued=%v, want 被拒", capped, queued) } if capped.Used != 2 || capped.Limit != 2 { t.Fatalf("超额 offer = %+v, want Used==Limit==2(拒绝不占额度)", capped) } if !strings.Contains(capped.Reason, "已达自我唤醒上限") { t.Fatalf("拒绝文案 = %q, want 可直接显示的中文", capped.Reason) } } // TestWakeNow_ZeroLimitFallsBackToDefault 是 C.8 判据 14(G3 探针):配 // SelfWakeLimit=0(并按同一口径再试一个负值)交给 New ⇒ 闸拿到的是 **normalize 后** // 的默认 4/1h;每次等该轮真跑完再发 ⇒ 前 4 次 Accepted=true(Used=1…4),第 5 次 // Accepted=false + capped(Used==Limit==4)。 // // 变异:把**未 normalize** 的 ≤0 直达 NewRateLimiter ⇒ Allow() 恒 true 且不记录 // ⇒ 第 5 次仍 Accepted=true(静默失败:无报错、无 WARN)⇒ 本用例变红。 func TestWakeNow_ZeroLimitFallsBackToDefault(t *testing.T) { t.Parallel() for _, limit := range []int{0, -1} { t.Run("limit="+strconv.Itoa(limit), func(t *testing.T) { t.Parallel() w := newWakeThinker() b, audit := startSelfWake(t, Config{SelfWakeLimit: limit, SelfWakeWindow: 0}, w) for i := 1; i <= defaultSelfWakeLimit; i++ { got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "回落默认后的第几次"}) if err != nil { t.Fatalf("第 %d 次 WakeNow: %v", i, err) } if !got.Accepted { t.Fatalf("第 %d 次收据 = %+v, want 排上(≤0 必须回落默认,而不是无上限)", i, got) } if got.Limit != defaultSelfWakeLimit || got.WindowSec != int(defaultSelfWakeWindow/time.Second) { t.Fatalf("第 %d 次收据 = %+v, want Limit=%d WindowSec=%d", i, got, defaultSelfWakeLimit, int(defaultSelfWakeWindow/time.Second)) } if got.Used != i { t.Fatalf("第 %d 次收据 = %+v, want Used=%d", i, got, i) } w.waitFor(t, 1+i, 3*time.Second) } got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "超出默认上限"}) if err != nil { t.Fatalf("第 %d 次 WakeNow: %v", defaultSelfWakeLimit+1, err) } if got.Accepted { t.Fatalf("第 %d 次收据 = %+v, want capped(默认上限是 %d)", defaultSelfWakeLimit+1, got, defaultSelfWakeLimit) } if got.Used != defaultSelfWakeLimit || got.Limit != defaultSelfWakeLimit { t.Fatalf("capped 收据 = %+v, want Used==Limit==%d", got, defaultSelfWakeLimit) } if got := audit.countOf(whatSelfWakeCapped); got != 1 { t.Fatalf("capped 审计 = %d 条, want 1(whats=%v)", got, audit.whats()) } }) } } // TestSelfWakeGate_WindowExpiry 覆盖「窗口过期后额度回来」。session.RateLimiter 内部 // 用 time.Now()(design/12 §N C.5 已登记的取舍:不为自排再自造第二套滑窗),因此窗口 // **无法注入假时钟** —— 这里只能把余量放大到与真实机器负载相称,断言口径不变: // 窗口内第二次必被拒、等过窗口必放行且 Used 回到 1。 // // 为什么是 6s(旧值 120ms): // - 「窗口内第二次」这条断言要求「第一次受理 → worker 真的进入两轮(waitFor)→ 第二次 // 调用」整段落在窗口内;waitFor 那一段是纯调度等待,73 个测试包并行抢核时会被拉长, // 120ms 一被拉长就必红(本轮其他证人实测红;本机 12 核上 30~60 次带载复跑未复现, // 所以这里按**余量**加固而不是拿复现驱动)。窗口取 6s ⇒ 给 waitFor 的 4s 上界留下 // 2s 余量:哪怕 waitFor 用满上界,也仍在窗口内,失败点因此只会是「worker 没被调度」 // 而不是一条含义被负载改写的窗口断言。 // - 到期等待不再是盲睡 3×窗口(放大后那是 18s 白等):以**第一次受理时刻**为基准, // 等到「窗口 + 250ms 结算余量」,既保留「等过窗口必放行」的下界语义,又不浪费墙钟。 func TestSelfWakeGate_WindowExpiry(t *testing.T) { t.Parallel() const ( window = 6 * time.Second // scheduleWait 是「worker 已被调度」这个前置件的上界,必须小于 window。 scheduleWait = 4 * time.Second // expirySettle 是窗口到期后的结算余量(RateLimiter 用 t.After(cutoff) 严格比较)。 expirySettle = 250 * time.Millisecond ) w := newWakeThinker() b, _ := startSelfWake(t, Config{SelfWakeLimit: 1, SelfWakeWindow: window}, w) firstAt := time.Now() if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "第一次"}); err != nil || !got.Accepted { t.Fatalf("第一次收据 = %+v err = %v, want 排上", got, err) } w.waitFor(t, 2, scheduleWait) if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "窗口内第二次"}); err != nil || got.Accepted { t.Fatalf("窗口内第二次收据 = %+v err = %v, want 被上限拒绝", got, err) } if wait := window + expirySettle - time.Since(firstAt); wait > 0 { time.Sleep(wait) } got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "窗口过了"}) if err != nil { t.Fatalf("窗口过后 WakeNow: %v", err) } if !got.Accepted { t.Fatalf("窗口过后的收据 = %+v, want 额度已随窗口释放", got) } if got.Used != 1 { t.Fatalf("窗口过后的收据 = %+v, want Used=1(旧记录已过期)", got) } } // TestSelfWakeAudit_FourActions 是 C.8 判据 11:注入 fake Auditor,四个动作各记一条 // 对应的 what,且 who 恒为 backthink。 func TestSelfWakeAudit_FourActions(t *testing.T) { t.Parallel() w := newWakeThinker() w.holdSelfWake = make(chan struct{}) b, audit := startSelfWake(t, Config{SelfWakeLimit: 2, SelfWakeWindow: time.Hour}, w) // ① 入槽并被 worker 取走(Used=1)② 槽空、额度剩 1 ⇒ 再入槽(Used=2) if _, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "排一轮"}); err != nil { t.Fatalf("WakeNow: %v", err) } w.waitFor(t, 2, 3*time.Second) if _, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "再排一轮"}); err != nil { t.Fatalf("WakeNow: %v", err) } // ③ 槽满 ⇒ 合并。 if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "槽满时再来"}); err != nil || !got.Coalesced { t.Fatalf("槽满时收据 = %+v err = %v, want 合并", got, err) } // 放行后等第二轮被取走(槽空、额度已用尽)⇒ 超额请求落到 capped。 close(w.holdSelfWake) w.waitFor(t, 3, 3*time.Second) if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "超额"}); err != nil || got.Accepted { t.Fatalf("超额收据 = %+v err = %v, want 被拒", got, err) } // 非主人环境 ⇒ denied。 ctx := loop.WithSessionEnv(context.Background(), loop.SessionEnv{ActorEnv: restrict.ActorStranger}) if _, err := b.WakeNow(ctx, SelfWakeRequest{Reason: "陌生人想唤醒"}); !errors.Is(err, ErrSelfWakeNotOwner) { t.Fatalf("err = %v, want ErrSelfWakeNotOwner", err) } // 本用例的序列是确定的(worker 被 holdSelfWake 钉住、步进都有 waitFor 同步点): // 两次真的入槽(Used=1、Used=2)⇒ 两条 queued;满槽一次 ⇒ 一条 coalesced; // 额度用尽后槽空 ⇒ 一条 capped;非主人 ⇒ 一条 denied。删掉任一审计调用 ⇒ 变红。 want := map[string]int{ whatSelfWakeQueued: 2, whatSelfWakeCoalesced: 1, whatSelfWakeCapped: 1, whatSelfWakeDenied: 1, } for what, n := range want { if got := audit.countOf(what); got != n { t.Fatalf("what=%q 记了 %d 条, want %d(全部=%v)", what, got, n, audit.whats()) } } audit.mu.Lock() defer audit.mu.Unlock() for _, r := range audit.rows { if r.who != selfWakeAuditWho { t.Fatalf("who = %q, want %q", r.who, selfWakeAuditWho) } if r.result == "" { t.Fatalf("result 为空(what=%q):审计行必须能直接显示", r.what) } } } // TestSelfWake_WorkerRecordsSelfWakeRow 走**生产路径**(WakeNow → worker → think → // Store):落库行的 Trigger 是 SelfWakeSource,落库的是折算后的档位(C.8 判据 9 的 // 自排一半)。C1-1 的用例直接构造 selfWakeTrigger,这里证明「排一次真的跑一轮」。 func TestSelfWake_WorkerRecordsSelfWakeRow(t *testing.T) { t.Parallel() w := newWakeThinker() b, _ := startSelfWake(t, Config{}, w) if got, err := b.WakeNow(ownerCtx(), SelfWakeRequest{Reason: "复盘今天的对话"}); err != nil || !got.Accepted { t.Fatalf("WakeNow = %+v err = %v, want 排上", got, err) } w.waitFor(t, 2, 3*time.Second) // 轮次已进入 ≠ 记录已落库(记录发生在 Think 返回之后),且启动轮与自排轮分属两个 // goroutine ⇒ 这里等那一行真的写进去,再断言它的形状。 var selfWake *Record eventually(t, 3*time.Second, func() bool { recs, err := b.List(context.Background(), 5) if err != nil { t.Fatalf("List: %v", err) } for i := range recs { if recs[i].Trigger == SelfWakeSource { selfWake = &recs[i] return true } } return false }) if selfWake.Urgency != "routine" { t.Fatalf("Record.Urgency = %q, want routine(未声明档位不上抬)", selfWake.Urgency) } } // TestSetAuditSink_LateBinding 钉住审计出口是晚绑定的(组合根在网关组装后调用), // 且换出口不需要重建推送器(审计不在推送路径上)。 func TestSetAuditSink_LateBinding(t *testing.T) { t.Parallel() w := newWakeThinker() b, _ := newBackthink(t, Config{}, Deps{Thinker: w, State: &mockState{}, Log: discardLog()}) before := b.currentPusher() audit := &fakeAuditor{} b.SetAuditSink(audit) if b.currentAuditor() != audit { t.Fatal("SetAuditSink 之后 currentAuditor 不是注入的那个出口") } if b.currentPusher() != before { t.Fatal("SetAuditSink 不该重建推送器(审计不在推送路径上)") } b.SetAuditSink(nil) if b.currentAuditor() != nil { t.Fatal("SetAuditSink(nil) 应摘下出口(审计缺失不阻塞业务)") } }