Cyrene_v2/internal/channel/adapter/flowgate_async_test.go

package adapter

// 本文件覆盖 M24：把策略四的审查**移出准入路径**，并让节流在失败路径同样生效。
// 每条用例都对应一条可变异点（见 M24 回报里的「预期会红的用例名」）。

import (
	"context"
	"fmt"
	"sync"
	"testing"
	"time"
)

// auditRecorder 收集 OnPruned 的审计回调。审查结论来自后台 goroutine，因此这里
// 必须加锁（-race 会盯）。
type auditRecorder struct {
	mu     sync.Mutex
	events []auditEvent
}

type auditEvent struct {
	content string
	reason  string
}

func (r *auditRecorder) hook() func(UnifiedMessage, string) {
	return func(m UnifiedMessage, reason string) {
		r.mu.Lock()
		r.events = append(r.events, auditEvent{content: m.Content, reason: reason})
		r.mu.Unlock()
	}
}

func (r *auditRecorder) snapshot() []auditEvent {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]auditEvent(nil), r.events...)
}

func (r *auditRecorder) reasons() []string {
	out := make([]string, 0)
	for _, e := range r.snapshot() {
		out = append(out, e.reason)
	}
	return out
}

// waitReviewIdle 等到闸门没有在飞的审查调用，使后台结论已经落回待处理集合。
// reviewInFlight 在**派发时**（Admit 持锁期间）就置位，因此 Admit 返回后立刻
// 调用它不会漏掉刚派发的调用。
func waitReviewIdle(t *testing.T, g *FlowGate) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		g.mu.Lock()
		busy := g.reviewInFlight
		g.mu.Unlock()
		if !busy {
			return
		}
		if time.Now().After(deadline) {
			t.Fatal("reviewer call did not finish within 5s")
		}
		time.Sleep(time.Millisecond)
	}
}

// pendingLen 返回某会话待处理集合的条目数。
func pendingLen(g *FlowGate, key string) int {
	g.mu.Lock()
	defer g.mu.Unlock()
	if st := g.sessions[key]; st != nil {
		return len(st.buffer)
	}
	return 0
}

// keepAll 是「审查者什么都不想丢」的替身：返回全部索引，使待处理集合恒在阈值上，
// 从而让节流/预算成为唯一的抑制因素（否则队列一被修剪就跌破阈值，测不出节流）。
func keepAll(_ context.Context, msgs []UnifiedMessage) ([]int, error) {
	keep := make([]int, len(msgs))
	for i := range msgs {
		keep[i] = i
	}
	return keep, nil
}

// TestFlowGate_AdmitNeverWaitsForSlowReview 是 M24 的核心证据：一个 3–6 s 量级的
// 慢审查替身压住闸门时，入站帧仍然按节奏被放行 —— 读循环不再被模型调用阻塞。
//
// 变异判据：把审查改回同步调用（在 Admit 里直接 g.review(...)）⇒ 帧间隔出现一个
// ≈4 s 的空窗 ⇒ 本用例红。
func TestFlowGate_AdmitNeverWaitsForSlowReview(t *testing.T) {
	t.Parallel()

	release := make(chan struct{})
	started := make(chan struct{}, 1)
	reviewer := func(_ context.Context, _ []UnifiedMessage) ([]int, error) {
		select {
		case started <- struct{}{}:
		default:
		}
		// 4 s 的慢审查替身：实测 noise_review 的 p50 5.3 s / max 9.2 s 同量级。
		select {
		case <-release:
		case <-time.After(4 * time.Second):
		}
		return nil, fmt.Errorf("slow stand-in returned")
	}
	g := NewFlowGate(FlowGateOptions{MaxMessages: 10, Window: time.Minute, Review: reviewer})
	defer close(release)

	const frames = 100
	const cadence = 20 * time.Millisecond
	var maxGap time.Duration
	var slowestAdmit time.Duration
	prev := time.Now()
	for i := 0; i < frames; i++ {
		admitStart := time.Now()
		// 消息与心跳/探活回执在适配器侧走的是同一条 Admit 路，这里逐帧模拟。
		if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed {
			t.Fatalf("frame %d dropped by the gate: %+v", i, v)
		}
		if d := time.Since(admitStart); d > slowestAdmit {
			slowestAdmit = d
		}
		if gap := time.Since(prev); gap > maxGap {
			maxGap = gap
		}
		prev = time.Now()
		time.Sleep(cadence)
	}

	// 先断言「帧没有被拖住」——同步实现会在这里给出可读的红。
	if maxGap > 500*time.Millisecond {
		t.Fatalf("inbound frames stalled behind the reviewer: max gap %v (cadence %v)", maxGap, cadence)
	}
	if slowestAdmit > 100*time.Millisecond {
		t.Fatalf("Admit waited on the reviewer: slowest admission %v", slowestAdmit)
	}

	// 审查确实被派发并且**仍在飞**：证明替身是慢的，上面测到的不是「审查根本没跑」。
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("reviewer was never called: the gate did not schedule the review")
	}
	g.mu.Lock()
	busy := g.reviewInFlight
	g.mu.Unlock()
	if !busy {
		t.Fatal("slow reviewer already returned; the stand-in was not slow enough to prove anything")
	}
}

// TestFlowGate_ReviewThrottleHoldsOnFailure 固定「节流在失败路径也必须生效」：
// 审查者连续失败时，同一窗口内同一会话只允许一次调用 —— 这正是修复前失守的地方
// （lastReview 只在 err == nil 分支置位，而实测 67–85% 的审查因 BOM 解析失败）。
//
// 变异判据：把 scheduleReviewLocked 里的 `st.lastReview = now` 挪回「成功之后」
// （或删掉）⇒ 11 条消息各派发一次 ⇒ 本用例红。
func TestFlowGate_ReviewThrottleHoldsOnFailure(t *testing.T) {
	t.Parallel()

	now, _ := fixedClock(time.Unix(1000, 0))
	var mu sync.Mutex
	calls := 0
	audit := &auditRecorder{}
	g := NewFlowGate(FlowGateOptions{
		MaxMessages: 4, // reviewAt = 3
		Window:      time.Minute,
		Now:         now,
		Review: func(_ context.Context, _ []UnifiedMessage) ([]int, error) {
			mu.Lock()
			calls++
			mu.Unlock()
			return nil, fmt.Errorf("invalid character 'ï' looking for beginning of value")
		},
		OnPruned: audit.hook(),
	})

	for i := 0; i < 11; i++ {
		if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed {
			t.Fatalf("message %d dropped although the reviewer only failed: %+v", i, v)
		}
		waitReviewIdle(t, g)
	}

	mu.Lock()
	got := calls
	mu.Unlock()
	if got != 1 {
		t.Fatalf("reviewer calls within one window = %d, want 1 (throttle must cover the failure path)", got)
	}
	// 失败方向：既不额外丢弃，也不给队列任何条目「保留」特权 —— 只有上限该有的
	// "capped" 审计，绝不出现 "review_pruned"。
	for _, r := range audit.reasons() {
		if r == "review_pruned" {
			t.Fatalf("a failed review must not prune anything, got audits %v", audit.reasons())
		}
	}
	if n := pendingLen(g, ConversationKey(groupMsg(0))); n != 4 {
		t.Fatalf("pending set = %d, want 4 (cap only, no review pruning)", n)
	}
}

// TestFlowGate_ReviewBudgetBoundsCallsPerWindow 把「每适配器每分钟 LLM 审查次数」的
// 硬上界写成可测断言：40 个群在同一窗口内全部越阈，实际调用不超过
// defaultReviewPerWindow（6）。
//
// 变异判据：删掉 reviewBudgetLocked 的调用（或把 reviewBudget 置 0/极大）⇒ 40 次
// ⇒ 本用例红。
func TestFlowGate_ReviewBudgetBoundsCallsPerWindow(t *testing.T) {
	t.Parallel()

	now, _ := fixedClock(time.Unix(1000, 0))
	var mu sync.Mutex
	calls := 0
	g := NewFlowGate(FlowGateOptions{
		MaxMessages: 4, // reviewAt = 3
		Window:      time.Minute,
		Now:         now,
		Review: func(ctx context.Context, msgs []UnifiedMessage) ([]int, error) {
			mu.Lock()
			calls++
			mu.Unlock()
			return keepAll(ctx, msgs)
		},
	})

	const groups = 40
	for gi := 0; gi < groups; gi++ {
		group := fmt.Sprintf("-20%d", gi)
		for i := 0; i < 4; i++ {
			g.Admit(context.Background(), groupMsgIn(group, i))
		}
		waitReviewIdle(t, g)
	}

	mu.Lock()
	got := calls
	mu.Unlock()
	if got > defaultReviewPerWindow {
		t.Fatalf("reviewer calls in one window = %d, want <= %d (per-adapter budget)", got, defaultReviewPerWindow)
	}
	if got != defaultReviewPerWindow {
		t.Fatalf("reviewer calls = %d, want exactly %d (every conversation crossed, budget is the only cap)", got, defaultReviewPerWindow)
	}
}

// TestFlowGate_ReviewSingleInFlightPerAdapter 固定「同一适配器至多一次在飞」：
// 两个群同时越阈，第二个必须让位，不并发起第二次模型调用。
//
// 变异判据：删掉 scheduleReviewLocked 里的 reviewInFlight 判断 ⇒ 第二个群也起一次
// 调用 ⇒ 本用例红。
func TestFlowGate_ReviewSingleInFlightPerAdapter(t *testing.T) {
	t.Parallel()

	release := make(chan struct{})
	started := make(chan string, 4)
	reviewer := func(_ context.Context, msgs []UnifiedMessage) ([]int, error) {
		started <- msgs[0].Content
		<-release
		return keepAll(context.Background(), msgs)
	}
	g := NewFlowGate(FlowGateOptions{MaxMessages: 4, Window: time.Minute, Review: reviewer})
	defer close(release)

	for i := 0; i < 4; i++ {
		g.Admit(context.Background(), groupMsgIn("-100", i))
	}
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("first review never started")
	}
	for i := 0; i < 4; i++ {
		g.Admit(context.Background(), groupMsgIn("-200", i))
	}
	select {
	case extra := <-started:
		t.Fatalf("second review started while one was in flight (%q)", extra)
	case <-time.After(100 * time.Millisecond):
	}
}

// TestFlowGate_ReviewIgnoresMessagesArrivedDuringCall 固定后台审查的落点：结论只作用于
// 审查者**当时看到的那批**（用序号对齐）。审查在飞期间新到的消息没被审查过，绝不能被
// 这次结论连带修剪 —— 否则审计会记出「review_pruned」却都是已经放行的消息。
//
// 变异判据：把 applyReviewSeqs 写成「不在 keep 集合里就丢」（不做快照过滤）⇒ 在飞期间
// 到达的两条也被丢 ⇒ 本用例红。
func TestFlowGate_ReviewIgnoresMessagesArrivedDuringCall(t *testing.T) {
	t.Parallel()

	release := make(chan struct{})
	started := make(chan struct{}, 1)
	audit := &auditRecorder{}
	g := NewFlowGate(FlowGateOptions{
		MaxMessages: 10, // reviewAt = 8
		Window:      time.Minute,
		Review: func(_ context.Context, _ []UnifiedMessage) ([]int, error) {
			started <- struct{}{}
			<-release
			return []int{0}, nil // 只保留快照里的第 0 条，其余视为噪声
		},
		OnPruned: audit.hook(),
	})

	for i := 0; i < 8; i++ {
		g.Admit(context.Background(), groupMsg(i))
	}
	select {
	case <-started:
	case <-time.After(2 * time.Second):
		t.Fatal("review never started")
	}
	// 审查在飞期间又到两条（seq 9、10）：它们不在这次审查的快照里。
	g.Admit(context.Background(), groupMsg(8))
	g.Admit(context.Background(), groupMsg(9))
	close(release)
	waitReviewIdle(t, g)

	// 快照第 0 条（msg 0）保留 + 在飞期间到达的两条 = 3。
	if n := pendingLen(g, ConversationKey(groupMsg(0))); n != 3 {
		t.Fatalf("pending set = %d, want 3 (msg0 + the two messages the reviewer never saw)", n)
	}
	for _, e := range audit.snapshot() {
		if e.reason != "review_pruned" {
			t.Fatalf("audit reason = %q, want review_pruned", e.reason)
		}
		if e.content == "msg 8" || e.content == "msg 9" {
			t.Fatalf("a message that arrived during the call was pruned: %q", e.content)
		}
	}
	if got := len(audit.snapshot()); got != 7 {
		t.Fatalf("review_pruned count = %d, want 7 (snapshot entries 1..7)", got)
	}
}

// TestFlowGate_CapStillEvictsOnAdmission 固定上限那一半仍是**同步**的：队列超出
// maxMessages 时最旧的条目在准入当次就被挤掉并审计 "capped"，不依赖审查。
//
// 变异判据：把上限修剪也挪进后台 ⇒ v.Pruned 为空 ⇒ 本用例红。
func TestFlowGate_CapStillEvictsOnAdmission(t *testing.T) {
	t.Parallel()

	audit := &auditRecorder{}
	g := NewFlowGate(FlowGateOptions{
		MaxMessages: 3,
		Review:      keepAll,
		OnPruned:    audit.hook(),
	})

	for i := 0; i < 3; i++ {
		if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed {
			t.Fatalf("message %d should fit under the cap: %+v", i, v)
		}
	}
	v := g.Admit(context.Background(), groupMsg(3))
	if !v.Proceed {
		t.Fatalf("new message should be admitted after FIFO eviction, got %+v", v)
	}
	if len(v.Pruned) != 1 || v.Pruned[0].Content != "msg 0" {
		t.Fatalf("pruned = %+v, want [msg 0]", v.Pruned)
	}
	found := false
	for _, r := range audit.reasons() {
		if r == "capped" {
			found = true
		}
	}
	if !found {
		t.Fatalf("cap eviction not audited as capped: %v", audit.reasons())
	}
}

// TestFlowGate_RateLimitDropsAtTheDoor 固定 rate_limited 的「门口丢」语义一字不改：
// 被滑窗挡下的消息根本不进待处理集合，审查也绝不会把它「捞回来」。
func TestFlowGate_RateLimitDropsAtTheDoor(t *testing.T) {
	t.Parallel()

	now, _ := fixedClock(time.Unix(1000, 0))
	g := NewFlowGate(FlowGateOptions{
		GroupMsgPerMin: 2,
		MaxMessages:    3,
		Window:         time.Minute,
		Now:            now,
		Review:         keepAll,
	})

	for i := 0; i < 2; i++ {
		if v := g.Admit(context.Background(), groupMsg(i)); !v.Proceed {
			t.Fatalf("message %d should pass the group window: %+v", i, v)
		}
	}
	v := g.Admit(context.Background(), groupMsg(2))
	if v.Proceed {
		t.Fatal("third message in the window must be rate limited")
	}
	if v.Reason != "rate_limited" {
		t.Fatalf("reason = %q, want rate_limited", v.Reason)
	}
	if v.DroppedMsg == nil || v.DroppedMsg.Content != "msg 2" {
		t.Fatalf("dropped msg = %+v, want msg 2", v.DroppedMsg)
	}
	if n := pendingLen(g, ConversationKey(groupMsg(0))); n != 2 {
		t.Fatalf("pending set = %d, want 2: a rate-limited message never enters it", n)
	}
	waitReviewIdle(t, g)
}