AskaEth
a25e0c8870
- 新增 TestParseKeepIndices_ToleratesBOMWiring：喂对端**原样字节**（\xef\xbb\xbf + JSON 数组） 给生产路径的 parseKeepIndices，必须解析成功且升序；同时断言夹具真的以 BOM 开头、且裸 json.Unmarshal 在它上面仍然失败 —— 把委派改回裸 json.Unmarshal 即红。 - 新增 TestParseKeepIndices_StillFiltersAfterDelegation：委派之后过滤口径不变（越界/负数/ 重复编号丢弃、结果升序）。 - 新增 TestNewLLMReviewer_ToleratesBOMOutput：走整条生产路径（NewLLMReviewer 产出的 Reviewer），回带 BOM 时给出保留编号而不是退回 FIFO。 - 注释里写明 M24 已确认的语义代价：审查移出准入路径后 review_pruned 不再从门口丢消息 （41–74 条/日 会进流水线），而 rate_limited 门口丢与 FIFO 上限修剪一字不改（由既有用例钉住）。 只加用例与必要注释（含一个为断言所需的 encoding/json import），未改动既有用例的任何断言。
418 lines
16 KiB
Go
418 lines
16 KiB
Go
package channelbridge
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"log/slog"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||
)
|
||
|
||
func itoa(n int) string {
|
||
if n == 0 {
|
||
return "0"
|
||
}
|
||
var b [20]byte
|
||
i := len(b)
|
||
for n > 0 {
|
||
i--
|
||
b[i] = byte('0' + n%10)
|
||
n /= 10
|
||
}
|
||
return string(b[i:])
|
||
}
|
||
|
||
func groupSender(i int) adapter.UnifiedMessage {
|
||
return adapter.UnifiedMessage{
|
||
Platform: "kook",
|
||
ChannelID: "-100",
|
||
ChannelType: adapter.ChannelGroup,
|
||
GroupID: "-100",
|
||
SenderID: "u" + itoa(i),
|
||
Content: "spam " + itoa(i),
|
||
ContentType: adapter.ContentText,
|
||
}
|
||
}
|
||
|
||
func groupOwnerMsg(i int) adapter.UnifiedMessage {
|
||
m := groupSender(i)
|
||
m.Sender = adapter.Actor{Role: adapter.RoleOwner}
|
||
return m
|
||
}
|
||
|
||
// TestFlowGate_RateLimitDropsNonOwnerAndAudits 验证闸门已接入
|
||
// 入口（onMessage），且被限流的非主人消息会被丢弃并经
|
||
// OnRejected 审计钩子暴露原因。丢弃路径是同步的
|
||
// （不会派生工作 goroutine），因此断言是确定性的。
|
||
func TestFlowGate_RateLimitDropsNonOwnerAndAudits(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "hi"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}}
|
||
var rejected []string
|
||
b, err := New(Options{
|
||
Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{},
|
||
OnRejected: func(_ adapter.UnifiedMessage, reason string) { rejected = append(rejected, reason) },
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
ad := &fakeAdapter{name: "kook", cfg: map[string]any{"bot_token": "x", "msgPerMin": 1}}
|
||
b.Add(ad)
|
||
|
||
// 第一条消息被放行（异步工作项），只对第二条入审计，
|
||
// 它被每分钟总量上限拒绝。
|
||
b.onMessage(context.Background(), groupSender(0), ad)
|
||
b.onMessage(context.Background(), groupSender(1), ad)
|
||
|
||
if len(rejected) != 1 || rejected[0] != "rate_limited" {
|
||
t.Fatalf("rejected = %v, want [rate_limited]", rejected)
|
||
}
|
||
}
|
||
|
||
// TestFlowGate_OwnerPrivateBypassesGate 验证主人消息完全绕过流控
|
||
// 闸门：即使限流很激进，主人消息也绝不会以
|
||
// rate_limited / review_pruned / capped 被丢弃。
|
||
func TestFlowGate_OwnerPrivateBypassesGate(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "hi"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
|
||
var rejected []string
|
||
b, err := New(Options{
|
||
Loop: chat, Sessions: sess, Owner: testOwnerProfile("kook", "u0", "u1", "u2"),
|
||
OnRejected: func(_ adapter.UnifiedMessage, reason string) { rejected = append(rejected, reason) },
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
ad := &fakeAdapter{name: "kook", cfg: map[string]any{"bot_token": "x", "groupMsgPerMin": 1, "maxMessages": 1}}
|
||
b.Add(ad)
|
||
|
||
// 三条主人消息：若闸门生效，第 2、3 条会被限流
|
||
// / 触及上限并到达 OnRejected。主人绕过则审计为空。
|
||
for i := 0; i < 3; i++ {
|
||
b.onMessage(context.Background(), groupOwnerMsg(i), ad)
|
||
}
|
||
if len(rejected) != 0 {
|
||
t.Fatalf("owner messages must bypass the gate, got rejected %v", rejected)
|
||
}
|
||
}
|
||
|
||
// TestFlowGate_GateOptionsDefaults 钉住「配置→限制」的解析规则，
|
||
// 它保证旧的 CHANNELS_JSON 账号向后兼容：键缺失 → 关闭，0 → 保护性
|
||
// 默认值，负值 → 关闭，正值 → 固定为该值。**私聊（陌生人）限流同一口径，
|
||
// 但缺省侧相反**：键缺失/非整数/0 一律取保护性默认 6（裁决③「陌生人必须有流控」
|
||
// ＋裁决 13「流控时间要可配置」）——绝不因为「没配」而静默关掉。
|
||
func TestFlowGate_GateOptionsDefaults(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
b, err := New(Options{Loop: &stubChatter{}, Sessions: &stubSessions{}})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
|
||
// direct 列用**字面量 6**（该常量由 adapter/flowgate_direct_test.go 的
|
||
// TestFlowDirectDefaultIsPinned 钉住）：缺键时若解析成 0（= 关掉流控）本用例必须红。
|
||
cases := []struct {
|
||
name string
|
||
cfg map[string]any
|
||
group, max int
|
||
direct int
|
||
}{
|
||
{"absent keys off", map[string]any{"bot_token": "x"}, 0, 0, 6},
|
||
{"group 0 defaults 6", map[string]any{"bot_token": "x", "groupMsgPerMin": 0}, 6, 0, 6},
|
||
{"group -1 off", map[string]any{"bot_token": "x", "groupMsgPerMin": -1}, 0, 0, 6},
|
||
{"group 10 pinned", map[string]any{"bot_token": "x", "groupMsgPerMin": 10}, 10, 0, 6},
|
||
{"max 0 defaults 50", map[string]any{"bot_token": "x", "maxMessages": 0}, 0, 50, 6},
|
||
{"max -1 off", map[string]any{"bot_token": "x", "maxMessages": -1}, 0, 0, 6},
|
||
{"max 80 pinned", map[string]any{"bot_token": "x", "maxMessages": 80}, 0, 80, 6},
|
||
{"direct 0 defaults 6", map[string]any{"bot_token": "x", "directMsgPerMin": 0}, 0, 0, 6},
|
||
{"direct -1 off", map[string]any{"bot_token": "x", "directMsgPerMin": -1}, 0, 0, 0},
|
||
{"direct 12 pinned", map[string]any{"bot_token": "x", "directMsgPerMin": 12}, 0, 0, 12},
|
||
{"direct non-numeric defaults 6", map[string]any{"bot_token": "x", "directMsgPerMin": "abc"}, 0, 0, 6},
|
||
}
|
||
for _, c := range cases {
|
||
t.Run(c.name, func(t *testing.T) {
|
||
opts := b.gateOptions(&fakeAdapter{name: "kook", cfg: c.cfg})
|
||
if opts.GroupMsgPerMin != c.group {
|
||
t.Fatalf("GroupMsgPerMin = %d, want %d", opts.GroupMsgPerMin, c.group)
|
||
}
|
||
if opts.MaxMessages != c.max {
|
||
t.Fatalf("MaxMessages = %d, want %d", opts.MaxMessages, c.max)
|
||
}
|
||
if opts.DirectMsgPerMin != c.direct {
|
||
t.Fatalf("DirectMsgPerMin = %d, want %d（缺失/0 走保护性默认 6，负值=关闭，正值=该值）",
|
||
opts.DirectMsgPerMin, c.direct)
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// TestFlowGate_DirectLimitFromAccountConfig 验证 directMsgPerMin 从账号配置到闸门的
|
||
// **整条链**：键缺失 ⇒ 保护性默认 6 条/分真的拦下第 7 条；正值 ⇒ 该值；负值 ⇒ 关闭
|
||
// （全部放行）。计数按发送者独立（AllowDirect 的会话键含 sender）。
|
||
func TestFlowGate_DirectLimitFromAccountConfig(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
cases := []struct {
|
||
name string
|
||
cfg map[string]any
|
||
// allowed 是限制内会被放行的条数；off = 限制被显式关闭（第 allowed+1 条不会被拒）。
|
||
allowed int
|
||
off bool
|
||
}{
|
||
{"absent key keeps the protective default 6", map[string]any{"bot_token": "x"}, 6, false},
|
||
{"pinned 2", map[string]any{"bot_token": "x", "directMsgPerMin": 2}, 2, false},
|
||
{"negative disables the limit", map[string]any{"bot_token": "x", "directMsgPerMin": -1}, 12, true},
|
||
}
|
||
for _, c := range cases {
|
||
c := c
|
||
t.Run(c.name, func(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
b, ad := newGateBridge(t, c.cfg)
|
||
gate := b.gateFor(ad)
|
||
if gate == nil {
|
||
t.Fatal("gateFor = nil, want a gate for the configured adapter")
|
||
}
|
||
for i := 0; i < c.allowed; i++ {
|
||
if !gate.AllowDirect(strangerDirectMsg("777", "第 "+itoa(i)+" 句")) {
|
||
t.Fatalf("额度内第 %d 条被拒，want allowed", i+1)
|
||
}
|
||
}
|
||
// 限制生效时第 allowed+1 条必须被拒（off=false）；关闭时（off=true）必须放行。
|
||
if over := gate.AllowDirect(strangerDirectMsg("777", "第 N+1 句")); over != c.off {
|
||
t.Fatalf("第 %d 条的放行结果 = %v（off=%v），want %v",
|
||
c.allowed+1, over, c.off, c.off)
|
||
}
|
||
// 另一个发送者有独立额度：不被前一个发送者挤占。
|
||
if !gate.AllowDirect(strangerDirectMsg("888", "我是另一个人")) {
|
||
t.Fatal("另一个发送者的额度被挤占")
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// newGateBridge 构建一条只用于流控解析的桥接器（telegram 适配器 + 给定账号配置），
|
||
// 使 directMsgPerMin 走与生产完全相同的「Config() → gateOptions」链。
|
||
func newGateBridge(t *testing.T, cfg map[string]any) (*Bridge, *fakeAdapter) {
|
||
t.Helper()
|
||
b, err := New(Options{Loop: &stubChatter{}, Sessions: &stubSessions{}})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
ad := &fakeAdapter{name: "telegram", cfg: cfg}
|
||
b.Add(ad)
|
||
return b, ad
|
||
}
|
||
|
||
// TestFlowGate_DirectLimitOffIsLogged 钉住「负值关闭必须留痕」：主人显式关掉熟人/
|
||
// 陌生人私聊流控时，日志里必须出现一条带 source 的 WARN（不许静默关闭）；而正值或
|
||
// 缺省（保护性默认）绝不产生这条告警（否则主人会以为流控被关了）。
|
||
func TestFlowGate_DirectLimitOffIsLogged(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
var buf bytes.Buffer
|
||
log := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))
|
||
at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
|
||
b, _, ad := newStrangerBridge(t, &stubChatter{}, &captureHistory{}, at, log)
|
||
|
||
ad.cfg = map[string]any{"bot_token": "x", "directMsgPerMin": -1}
|
||
if opts := b.gateOptions(ad); opts.DirectMsgPerMin != 0 {
|
||
t.Fatalf("DirectMsgPerMin = %d, want 0（负值 = 关闭）", opts.DirectMsgPerMin)
|
||
}
|
||
logged := buf.String()
|
||
if !strings.Contains(logged, "non-owner private rate limit turned off") ||
|
||
!strings.Contains(logged, "source=services/channelbridge/flowgate") ||
|
||
!strings.Contains(logged, "directMsgPerMin") {
|
||
t.Fatalf("负值关闭流控没有留下带 source 的可见告警：%q", logged)
|
||
}
|
||
|
||
// 正值：不告警，且额度就是该值。
|
||
buf.Reset()
|
||
ad.cfg = map[string]any{"bot_token": "x", "directMsgPerMin": 12}
|
||
if opts := b.gateOptions(ad); opts.DirectMsgPerMin != 12 {
|
||
t.Fatalf("DirectMsgPerMin = %d, want 12", opts.DirectMsgPerMin)
|
||
}
|
||
if got := buf.String(); got != "" {
|
||
t.Fatalf("正值不该产生「流控被关闭」告警：%q", got)
|
||
}
|
||
|
||
// 缺省（保护性默认）：同样不告警 —— 默认不是「主人关掉了它」。
|
||
buf.Reset()
|
||
ad.cfg = map[string]any{"bot_token": "x"}
|
||
if opts := b.gateOptions(ad); opts.DirectMsgPerMin != adapter.FlowDirectDefault {
|
||
t.Fatalf("DirectMsgPerMin = %d, want the protective default %d",
|
||
opts.DirectMsgPerMin, adapter.FlowDirectDefault)
|
||
}
|
||
if got := buf.String(); got != "" {
|
||
t.Fatalf("缺省不该产生「流控被关闭」告警：%q", got)
|
||
}
|
||
}
|
||
|
||
// mockLLM 是返回固定补全的 model.Adapter 桩。
|
||
type mockLLM struct {
|
||
content string
|
||
err error
|
||
}
|
||
|
||
func (m *mockLLM) Name() string { return "mock-llm" }
|
||
func (m *mockLLM) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) {
|
||
if m.err != nil {
|
||
return model.Response{}, m.err
|
||
}
|
||
return model.Response{Content: m.content, Message: model.TextMessage(model.RoleAssistant, m.content)}, nil
|
||
}
|
||
func (m *mockLLM) ChatStream(_ context.Context, _ model.Request, _ func(model.StreamEvent)) (model.Response, error) {
|
||
return model.Response{}, nil
|
||
}
|
||
|
||
// TestNewLLMReviewer 验证审查模型被包装成 Reviewer：向
|
||
// LLM 询问哪些队列消息该保留，并过滤返回的索引。
|
||
func TestNewLLMReviewer(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
llm := &mockLLM{content: `[0,1,99,0]`}
|
||
review := NewLLMReviewer(llm, nil)
|
||
|
||
msgs := []adapter.UnifiedMessage{
|
||
{SenderID: "a", Content: "x", ContentType: adapter.ContentText},
|
||
{SenderID: "b", Content: "y", ContentType: adapter.ContentText},
|
||
{SenderID: "c", Content: "z", ContentType: adapter.ContentText},
|
||
}
|
||
keep, err := review(context.Background(), msgs)
|
||
if err != nil {
|
||
t.Fatalf("review: %v", err)
|
||
}
|
||
// 越界索引 99 与重复的 0 被丢弃；保留 = [0,1]。
|
||
if len(keep) != 2 || keep[0] != 0 || keep[1] != 1 {
|
||
t.Fatalf("keep = %v, want [0 1]", keep)
|
||
}
|
||
|
||
if llm.err != nil {
|
||
t.Fatal("unexpected mock error")
|
||
}
|
||
}
|
||
|
||
// TestNewLLMReviewerError 验证审查模型失败会暴露给闸门，
|
||
// 闸门随后回退为 FIFO 淘汰，而不是伪造一个成功。
|
||
func TestNewLLMReviewerError(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
review := NewLLMReviewer(&mockLLM{err: context.DeadlineExceeded}, nil)
|
||
if _, err := review(context.Background(), []adapter.UnifiedMessage{
|
||
{SenderID: "a", Content: "x", ContentType: adapter.ContentText},
|
||
}); err == nil {
|
||
t.Fatal("expected error from failed review model")
|
||
}
|
||
}
|
||
|
||
func TestParseKeepIndices(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
got, err := parseKeepIndices(`[2,0,0,5]`, 3)
|
||
if err != nil {
|
||
t.Fatalf("parse: %v", err)
|
||
}
|
||
if len(got) != 2 || got[0] != 0 || got[1] != 2 {
|
||
t.Fatalf("keep = %v, want [0 2] (out-of-range/dup dropped)", got)
|
||
}
|
||
if _, err := parseKeepIndices(`not json`, 3); err == nil {
|
||
t.Fatal("expected error for non-JSON")
|
||
}
|
||
}
|
||
|
||
// ── M24：审查解析的**接线**（BOM 韧性只有在这里才算真的生效）────────────────
|
||
//
|
||
// parseKeepIndices 现在是 adapter.ParseReviewIndices 的一行委派；adapter 侧有自己的
|
||
// 用例，但「委派还在不在」只有本文件的用例能锁住 —— 谁把它改回裸 json.Unmarshal，
|
||
// 下面这几条就会红。生产症状（`~/cy.log`，9 天）：1,909 次审查里 1,301 次解析失败，
|
||
// 其中 1,288 次错误串逐字是 `invalid character 'ï' looking for beginning of value`
|
||
// （对端在数组前回带了 UTF-8 BOM），每一次都白烧一次模型调用。
|
||
//
|
||
// 另一条已确认接受的语义代价（M24，塔裁决）：审查移出准入路径后 review_pruned 不再
|
||
// 从门口丢消息（实测 41–74 条/日 现在会进流水线）—— 这是「不阻塞读循环」的必然代价。
|
||
// rate_limited 的门口丢与 FIFO 上限修剪**一字不改**：前者由本文件既有的
|
||
// TestFlowGate_RateLimitDropsNonOwnerAndAudits 钉住，后者由 adapter 侧的
|
||
// TestFlowGate_CapStillEvictsOnAdmission 钉住。
|
||
|
||
// bomReviewOutput 是审查模型网关的**原样**回包：UTF-8 BOM 紧接 JSON 数组。写死字节
|
||
// （而不是 \ufeff 转义或运行时拼装），使「夹具被谁悄悄去掉 BOM」本身就会红。
|
||
const bomReviewOutput = "\xef\xbb\xbf[0, 1, 2, 3, 4, 5]"
|
||
|
||
// TestParseKeepIndices_ToleratesBOMWiring 锁住接线：生产路径的 parseKeepIndices 必须
|
||
// 吃下带 BOM 的对端回包。
|
||
func TestParseKeepIndices_ToleratesBOMWiring(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
// 夹具真实性：它必须以 BOM 字节开头，且裸 json.Unmarshal 在它上面必须失败 ——
|
||
// 否则这条用例就不再模拟任何生产故障。
|
||
if !strings.HasPrefix(bomReviewOutput, "\xef\xbb\xbf") {
|
||
t.Fatal("fixture lost its BOM bytes: this case no longer models the production failure")
|
||
}
|
||
var raw []int
|
||
if err := json.Unmarshal([]byte(bomReviewOutput), &raw); err == nil {
|
||
t.Fatal("strict json.Unmarshal accepted the BOM fixture; the case is meaningless now")
|
||
}
|
||
|
||
got, err := parseKeepIndices(bomReviewOutput, 6)
|
||
if err != nil {
|
||
t.Fatalf("parseKeepIndices(BOM) error = %v, want nil (delegation to adapter.ParseReviewIndices lost?)", err)
|
||
}
|
||
if len(got) != 6 {
|
||
t.Fatalf("keep = %v, want [0 1 2 3 4 5]", got)
|
||
}
|
||
for i := range got {
|
||
if got[i] != i {
|
||
t.Fatalf("keep = %v, want ascending [0 1 2 3 4 5]", got)
|
||
}
|
||
}
|
||
}
|
||
|
||
// TestParseKeepIndices_StillFiltersAfterDelegation 委派之后过滤口径不变：越界、负数、
|
||
// 重复的编号仍被丢弃，结果按升序（模型幻觉出的数字绝不能淘汰真实消息）。
|
||
func TestParseKeepIndices_StillFiltersAfterDelegation(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
got, err := parseKeepIndices("\xef\xbb\xbf[3, -1, 3, 1, 99]", 4)
|
||
if err != nil {
|
||
t.Fatalf("parse: %v", err)
|
||
}
|
||
if len(got) != 2 || got[0] != 1 || got[1] != 3 {
|
||
t.Fatalf("keep = %v, want [1 3] (out-of-range/negative/dup dropped, ascending)", got)
|
||
}
|
||
}
|
||
|
||
// TestNewLLMReviewer_ToleratesBOMOutput 走**整条生产路径**：审查模型回带 BOM ⇒
|
||
// NewLLMReviewer 产出的 Reviewer 必须给出保留编号，而不是报 unparsable 退回 FIFO。
|
||
// （它证明的是密封在函数里的 wiring，而不只是 parseKeepIndices 这一个符号。）
|
||
func TestNewLLMReviewer_ToleratesBOMOutput(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
review := NewLLMReviewer(&mockLLM{content: bomReviewOutput}, nil)
|
||
msgs := make([]adapter.UnifiedMessage, 0, 6)
|
||
for i := 0; i < 6; i++ {
|
||
msgs = append(msgs, adapter.UnifiedMessage{
|
||
SenderID: "u",
|
||
Content: "m",
|
||
ContentType: adapter.ContentText,
|
||
})
|
||
}
|
||
keep, err := review(context.Background(), msgs)
|
||
if err != nil {
|
||
t.Fatalf("review with BOM output: %v, want nil error", err)
|
||
}
|
||
if len(keep) != 6 {
|
||
t.Fatalf("keep = %v, want all six indices", keep)
|
||
}
|
||
}
|