Merge branch 'feat/m24-flowgate-move-review-off-admission-p' into dev

cmd/cyrene/channels.go

@@ -90,11 +90,62 @@ func buildPlatformAdapters(platform string, accts []config.ChannelAccount, log *
 			log.Warn("channel account build skipped", "platform", platform, "account", acct.ID, "error", berr)
 			continue
 		}
+		logFlowControl(log, built)
 		out = append(out, built)
 	}
 	return out
 }
 
+// logFlowControl 逐账号打一条启动日志，如实报出这个账号此刻被限成什么样（M24 裁决 A）：
+// 闸是否开、每群几条/分、待处理上限、审查触发水位（reviewAt）。
+//
+// 为什么需要它：namedAdapter 的 Config() 无条件暴露四个流控键（含零值），桥的 cfgInt
+// 因此永远认为「键存在」，flowOrDefault 再把 0 映射成保护性默认 —— 于是「一个流控键都
+// 没写的账号，闸门仍然是开着的」。这条事实必须一眼可见，否则「群里为什么有消息被丢」
+// 只能靠读代码解释。判据：只看这一行日志就能说出这个群的限流/审查额度。
+//
+// 判定依据是**构建出来的那个适配器本身**，不是重新读一遍配置：非 namedAdapter 的实例
+// 不暴露流控键，桥按 flowgate.go「键缺失 ⇒ 该限制关闭」处理；这时必须如实报 off，
+// 不能拿保护性默认值假装成「开着」。
+func logFlowControl(log *slog.Logger, built adapter.PlatformAdapter) {
+	if log == nil {
+		log = slog.Default()
+	}
+	log = logger.WithSource(log, "cmd/cyrene/channels")
+	n, ok := built.(*namedAdapter)
+	if !ok {
+		log.Info("channel account ingress flow control",
+			"account", built.Name(), "gate", "off",
+			"why", "adapter exposes no flow keys, every limit stays disabled")
+		return
+	}
+	group := flowLimitOrDefault(n.groupMsgPerMin, adapter.FlowLimitDefault)
+	direct := flowLimitOrDefault(n.directMsgPerMin, adapter.FlowDirectDefault)
+	pending := flowLimitOrDefault(n.maxMessages, adapter.FlowCapDefault)
+	log.Info("channel account ingress flow control",
+		"account", n.Name(), "gate", "on",
+		"group_msg_per_min", group, "group_msg_per_min_raw", n.groupMsgPerMin,
+		"direct_msg_per_min", direct, "direct_msg_per_min_raw", n.directMsgPerMin,
+		"msg_per_min", max(n.msgPerMin, 0),
+		"max_messages", pending, "max_messages_raw", n.maxMessages,
+		"review_at", adapter.ReviewAtFor(pending),
+		"note", "raw 0 (or absent in channels.json) takes the protective default; negative disables")
+}
+
+// flowLimitOrDefault 把账号原始值映射为实际生效值：负值关闭该限制、零值取保护性默认、
+// 正值固定。它与 internal/services/channelbridge/flowgate.go 的 flowOrDefault 逐字同
+// 口径，这里只用于上面那条启动日志 —— 日志必须报生效值，不能报原始值。
+func flowLimitOrDefault(v, def int) int {
+	switch {
+	case v < 0:
+		return 0
+	case v == 0:
+		return def
+	default:
+		return v
+	}
+}
+
 // channelAccountsFromEnv 合成 env 兜底：单账号 CHANNEL_<NAME>_* 适配器，加上
 // CHANNELS_JSON 里的多账号条目。它是 channels.json 缺失或非法时传给
 // runtimeconfig.LoadChannels 的 .env 兜底。.env 只承载必要的连接兜底
@@ -174,8 +225,18 @@ type namedAdapter struct {
 	denyGroups  []string
 	allowUsers  []string
 	denyUsers   []string
-	// 流控限制（design/12 §6 策略三/四）。它们始终经 Config() 暴露，使桥能构建按账号
+	// 流控限制（design/12 §6 策略三/四）。**四个键始终经 Config() 暴露，包括零值**：
-	// 的入口闸门；桥把零值默认映射为保护性取值，把负值映射为「关闭」。
+	// 桥的 cfgInt 因此永远认为「键存在」，flowOrDefault 于是把 0 映射成保护性默认
+	// （每群 6 条/分、待处理上限 50、reviewAt = 40）—— 一个在 channels.json 里一个
+	// 流控键都没写的账号，闸门**仍然是开着的**（M24 只读勘测实测：现存 qq 账号就是
+	// 这种状态）。负值经桥映射为「关闭」。
+	//
+	// 边界的准确说法（M24）：flowgate.go 那句「键缺失 ⇒ 该限制关闭」的向后兼容分支，
+	// 对 CHANNELS_JSON 账号**不可达** —— config.ChannelAccount 用的是普通 int
+	// （omitempty），「未配置」与「显式配 0」解出来都是 0，这里无从区分。要区分就得
+	// 改 config 的类型（*int / 存在性表）或在本题二次解析原始 JSON；而那样做会改掉
+	// 「闸门默认开还是关」这个产品口径（群聊限流/审查默认从开变关），属主人裁决，
+	// 不在本处自行翻转。见 M24 回报里的选项 A/B/C。
 	groupMsgPerMin  int
 	directMsgPerMin int
 	msgPerMin       int

internal/channel/adapter/flowgate.go

@@ -13,21 +13,32 @@ import (
 // It is a pure noise/importance filter (design/12 §6 策略四). Implementations must
 // never decide identity: who the sender is was already resolved structurally by the
 // channel layer. A nil Reviewer makes the gate fall back to a plain FIFO cap.
+//
+// The gate calls it **off the admission path** (see FlowGate.Admit), so an
+// implementation may take as long as the model takes; it is bounded by
+// reviewTimeout and by the per-window budget, never by the caller's patience.
 type Reviewer func(ctx context.Context, msgs []UnifiedMessage) (keep []int, err error)
 
 // GateVerdict reports how the gate classified one inbound message.
 type GateVerdict struct {
 	// Proceed reports whether the message should be delivered downstream.
 	Proceed bool
-	// Reason is a short token describing a drop: "rate_limited", "review_pruned"
+	// Reason is a short token describing a drop. After M24 it is always
-	// or "capped". It is empty when Proceed is true.
+	// "rate_limited" (over the sliding window — dropped at the door, the message
+	// never enters the pending set), because nothing else on the admission path can
+	// refuse a message: the pending cap evicts the **oldest** entries and the review
+	// model runs in the background, so neither can reject the message that just
+	// arrived. The audit hook still reports their drops with "capped" /
+	// "review_pruned"; those are retained-message bookkeeping, not admissions.
+	// Reason is empty when Proceed is true.
 	Reason string
 	// DroppedMsg is the message that failed admission (nil when Proceed is true).
 	DroppedMsg *UnifiedMessage
-	// Pruned lists every message the gate dropped during this admission. On the
+	// Pruned lists every message the gate dropped during this admission: on the cap
-	// review path it is the noise the reviewer rejected; on the cap path it is the
+	// path it is the oldest messages evicted to honor the total cap. Each is recorded
-	// oldest messages evicted to honor the total cap. Each is recorded to the
+	// to the OnPruned audit hook before being dropped, so nothing is lost silently.
-	// OnPruned audit hook before being dropped, so nothing is lost silently.
+	// Backlog drops made by a background review reach the audit hook with
+	// "review_pruned" when that review completes, not through this field.
 	Pruned []UnifiedMessage
 }
 
@@ -56,15 +67,23 @@ type FlowGateOptions struct {
 	MsgPerMin int
 	// MaxMessages is the **per-conversation** pending-message cap (design/12 §6
 	// 策略四 / 裁决 06.4.3). When a conversation's pending set nears it the Reviewer
-	// is invoked to prune noise; with no Reviewer the oldest messages are evicted
+	// is invoked — in the background — to prune noise; with no Reviewer the oldest
-	// FIFO. Zero disables the cap.
+	// messages are evicted FIFO. Zero disables the cap.
 	MaxMessages int
 	// Window is the sliding window over which the per-minute limits are counted.
 	// Zero uses defaultRateWindow.
 	Window time.Duration
 	// Review is the optional review model. When nil the gate enforces the cap with
-	// FIFO eviction instead of model judgment.
+	// FIFO eviction instead of model judgment. It is always called from a background
+	// goroutine, at most once per conversation per Window, with at most one call in
+	// flight per gate.
 	Review Reviewer
+	// ReviewPerMin is the hard per-account ceiling on Reviewer calls started per
+	// Window: the budget that bounds review cost no matter how many conversations
+	// cross their own threshold in the same window. Zero (or negative) uses
+	// defaultReviewPerWindow. It bounds calls started, not calls finished, so the
+	// bound holds independently of model latency.
+	ReviewPerMin int
 	// OnPruned is called once per pruned/dropped message with a short reason token
 	// ("review_pruned" or "capped") so the caller can record an audit trail.
 	// It may be nil (then the caller only sees Pruned on the verdict).
@@ -80,6 +99,13 @@ const (
 	// runs. Design/12 §6 策略四 says "maxMessages 的 80%": this is the cost guard
 	// that keeps review from firing on every message.
 	defaultReviewPct = 80
+	// defaultReviewPerWindow is the hard per-adapter ceiling on reviewer calls per
+	// window (6/minute, the same order of magnitude as FlowLimitDefault). Before it
+	// existed the only bound was "one per conversation per window", which an account
+	// with many conversations could multiply without limit. At p50 5.3 s per call
+	// this still allows ~half a minute of review work per minute per account, so it
+	// bounds the tail without starving normal traffic.
+	defaultReviewPerWindow = 6
 	// FlowLimitDefault is the protective default of 6 group messages per minute
 	// (design/12 §6 策略三). It is applied when an account pins no group rate limit.
 	FlowLimitDefault = 6
@@ -128,6 +154,10 @@ type flowSession struct {
 	groupHits  []time.Time
 	directHits []time.Time
 	buffer     []queuedMsg
+	// lastReview 是**最近一次审查被派发**的时刻（不是完成时刻，也无论成败）：
+	// 写在这里而不是「成功之后」，失败路径才同样被窗口节流（修复前它只在
+	// err == nil 分支置位，而 68% 的审查因对端回带 BOM 解析失败 ⇒ 缓冲满之后
+	// 每条入站帧都付一次 ~6 s 的模型调用）。
 	lastReview time.Time
 	// inFlight 是该会话**正在跑**的回合数（AcquireTurn/ReleaseTurn 维护）。受限车道
 	// 用它把「同一发送者同时最多 1 个在跑的回合」落到实处。
@@ -148,6 +178,14 @@ type flowSession struct {
 //
 // 两个策略的计数都按会话独立（裁决 06.4.3），只有 MsgPerMin 总量护栏按账号汇总。
 //
+// **审查不在准入路径上**：Admit 只读待处理集合里已经存在的结论（限流计数、
+// 上一批已完成审查留下的队列），触发条件满足时只把一次审查**交给后台**，因此
+// 适配器的读 goroutine 永远不会等模型 —— 修复前它会同步等一次 `noise_review`
+// （实测 p50 5.3 s / p95 6.7 s / max 9.2 s），把消息、心跳与探活回执全排在同一条
+// 队上（见 DevDocs/audit/群聊排队与回旧话-2026-10-24.md）。后台审查的三重上界：
+// 同一适配器至多一次在飞（reviewInFlight）、同一会话每 Window 至多一次
+// （lastReview）、整个适配器每 Window 至多 defaultReviewPerWindow 次（reviewHits）。
+//
 // Owner-private messages are not routed through the gate (the caller bypasses it),
 // so the owner is never throttled or pruned. Identity is never examined here.
 // The gate also carries the per-conversation running-turn slot (AcquireTurn /
@@ -160,24 +198,48 @@ type FlowGate struct {
 	reviewAt     int
 	window       time.Duration
 	review       Reviewer
+	reviewBudget int
 	onPruned     func(UnifiedMessage, string)
 	now          func() time.Time
 
 	mu        sync.Mutex
 	totalHits []time.Time
-	sessions  map[string]*flowSession
+	// reviewHits 是本适配器在窗口内**派发**审查的时刻，用于账号级预算
-	admitted  uint64
+	// （defaultReviewPerWindow）。
-	seq       uint64
+	reviewHits []time.Time
+	// reviewInFlight 表示本适配器已有一次审查在后台跑：同一适配器至多一次在飞，
+	// 使后台审查有界（不会因为多个群同时越阈而并发起 N 个模型调用）。
+	reviewInFlight bool
+	sessions       map[string]*flowSession
+	admitted       uint64
+	seq            uint64
 }
 
 // queuedMsg wraps a buffer entry with a stable sequence so pruning decisions can
 // be audited and the fate of the just-admitted message be tracked across index
 // shifts (value equality on UnifiedMessage is unsafe because it holds slices).
+// The sequence also lets a review verdict land on the entries the reviewer actually
+// saw, even when the buffer has moved on while the call was in flight.
 type queuedMsg struct {
 	seq uint64
 	msg UnifiedMessage
 }
 
+// ReviewAtFor 返回给定待处理上限对应的审查触发水位（design/12 §6 策略四：
+// maxMessages 的 80%，至少 1）。它是 NewFlowGate 内部同一处的口径，导出只为让调用方
+// **如实报出生效值**（例如 cmd/cyrene 的逐账号入口流控启动日志），而不是在别处另写
+// 一遍 80% 的算法。非正上限表示上限关闭，返回 0（永不触发审查）。
+func ReviewAtFor(maxMessages int) int {
+	if maxMessages <= 0 {
+		return 0
+	}
+	at := maxMessages * defaultReviewPct / 100
+	if at < 1 {
+		return 1
+	}
+	return at
+}
+
 // NewFlowGate builds a gate from opts. Options that are zero disable their limit,
 // except Window which defaults to one minute when left zero.
 func NewFlowGate(opts FlowGateOptions) *FlowGate {
@@ -188,6 +250,7 @@ func NewFlowGate(opts FlowGateOptions) *FlowGate {
 		maxMessages:  opts.MaxMessages,
 		window:       opts.Window,
 		review:       opts.Review,
+		reviewBudget: opts.ReviewPerMin,
 		onPruned:     opts.OnPruned,
 		now:          opts.Now,
 		sessions:     make(map[string]*flowSession),
@@ -195,22 +258,24 @@ func NewFlowGate(opts FlowGateOptions) *FlowGate {
 	if g.window <= 0 {
 		g.window = defaultRateWindow
 	}
-	if g.maxMessages > 0 {
+	if g.reviewBudget <= 0 {
-		g.reviewAt = g.maxMessages * defaultReviewPct / 100
+		g.reviewBudget = defaultReviewPerWindow
-		if g.reviewAt < 1 {
-			g.reviewAt = 1
-		}
 	}
+	g.reviewAt = ReviewAtFor(g.maxMessages)
 	if g.now == nil {
 		g.now = time.Now
 	}
 	return g
 }
 
-// Admit applies the rate limits then the per-conversation cap/review gate to one
+// Admit applies the rate limits then the per-conversation cap gate to one inbound
-// inbound message. It returns a verdict telling the caller whether to proceed;
+// message. It returns a verdict telling the caller whether to proceed; messages
-// messages that the gate drops are returned (and audited) so nothing is lost
+// that the gate drops are returned (and audited) so nothing is lost silently.
-// silently.
+//
+// **它从不同步调用审查模型**：限流与上限是 O(1) 的判定，审查只是一次「后台修剪」
+// 的派发。修复前这里同步等 `noise_review`（p50 5.3 s），而五路入站里有三路
+// （qq-ws、kook-ws、weixin 长轮询）是「单读 goroutine + 同步派发」，于是消息、
+// 心跳、探活回执共享同一条 6 s 队列 —— 主人口中「消息一直排队」的机制。
 //
 // 计数按会话独立（裁决 06.4.3）：群聊滑窗与待处理上限都落在 ConversationKey 对应的
 // flowSession 上（两个群并发互不挤占），只有 MsgPerMin 这一账号级总量护栏仍按适配器汇总。
@@ -218,48 +283,55 @@ func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict {
 	now := g.now()
 	key := ConversationKey(msg)
 	g.mu.Lock()
-	defer g.mu.Unlock()
 	g.maybeGCLocked(now)
+	verdict, job := g.admitLocked(msg, key, now)
+	g.mu.Unlock()
 
+	if job != nil {
+		// 派发在解锁之后：后台审查自己会加锁把结论落回待处理集合，绝不能让它
+		// 与这次准入抢同一把锁（也绝不能让准入等它）。
+		go g.runReview(ctx, job)
+	}
+	return verdict
+}
+
+// admitLocked runs the whole admission decision under g.mu and returns the verdict
+// plus, when due, the background review job the caller must launch after unlocking.
+// It never calls the Reviewer itself.
+func (g *FlowGate) admitLocked(msg UnifiedMessage, key string, now time.Time) (GateVerdict, *reviewJob) {
 	// 策略三（会话级）: sliding-window rate limits. 群聊上限按每个群独立计数；MsgPerMin
 	// 是账号级的非主人消息总量护栏。Over-limit messages are dropped at the door
-	// (downgraded) instead of blocking.
+	// (downgraded) instead of blocking. 这一段的语义与修复前**一字不改**：
+	// 被限流的消息根本不进待处理集合，也永远不会因为审查被「捞回来」。
 	if g.groupPerMin > 0 && msg.ChannelType == ChannelGroup {
 		st := g.sessionLocked(key)
 		st.groupHits = g.trimHits(st.groupHits, now)
 		if len(st.groupHits) >= g.groupPerMin {
-			return g.drop("rate_limited", msg)
+			return g.drop("rate_limited", msg), nil
 		}
 		st.groupHits = append(st.groupHits, now)
 	}
 	if g.msgPerMin > 0 {
 		g.totalHits = g.trimHits(g.totalHits, now)
 		if len(g.totalHits) >= g.msgPerMin {
-			return g.drop("rate_limited", msg)
+			return g.drop("rate_limited", msg), nil
 		}
 		g.totalHits = append(g.totalHits, now)
 	}
 
 	if g.maxMessages <= 0 {
-		return GateVerdict{Proceed: true}
+		return GateVerdict{Proceed: true}, nil
 	}
 
-	// 策略四（会话级）: add to this conversation's pending set, then prune when it
+	// 策略四（会话级）: add to this conversation's pending set, then enforce the cap
-	// nears the cap.
+	// synchronously. 审查只被**派发**（下面的 scheduleReviewLocked），结论稍后由
+	// 后台落回 —— 准入读的是「当前已有结论」。
 	st := g.sessionLocked(key)
 	g.seq++
 	seq := g.seq
 	st.buffer = append(st.buffer, queuedMsg{seq: seq, msg: msg})
 
 	var dropped []queuedMsg
-	if g.review != nil && len(st.buffer) >= g.reviewAt && !g.reviewRecently(st, now) {
-		keep, err := g.review(ctx, queueMessages(st.buffer))
-		if err == nil {
-			st.buffer, dropped = g.applyReview(st.buffer, keep)
-			st.lastReview = now
-			g.auditDropped(dropped, "review_pruned")
-		}
-	}
 	if len(st.buffer) > g.maxMessages {
 		n := len(st.buffer) - g.maxMessages
 		var evicted []queuedMsg
@@ -267,18 +339,12 @@ func (g *FlowGate) Admit(ctx context.Context, msg UnifiedMessage) GateVerdict {
 		dropped = append(dropped, evicted...)
 		g.auditDropped(evicted, "capped")
 	}
+	job := g.scheduleReviewLocked(st, key, now)
 
-	if !bufferHasSeq(st.buffer, seq) {
+	// 本条一定还在缓冲里：上限按新旧顺序丢最旧，而它是最新的（maxMessages ≥ 1）。
-		reason := "capped"
+	// 于是修复后准入路径上唯一能让消息 Proceed=false 的判据就是上面的限流 —— 审查
-		if containsSeq(dropped, seq) {
+	// （已移到后台）与上限（丢最旧）都只影响**已经放行过的**历史条目，只留审计痕迹。
-			reason = "review_pruned"
+	return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)}, job
-		}
-		v := GateVerdict{Proceed: false, Reason: reason, Pruned: queueMessages(dropped)}
-		c := msg
-		v.DroppedMsg = &c
-		return v
-	}
-	return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)}
 }
 
 // AllowDirect enforces the per-sender sliding-window rate limit for a **direct
@@ -385,15 +451,6 @@ func (g *FlowGate) maybeGCLocked(now time.Time) {
 	}
 }
 
-// reviewRecently throttles review so it does not run on every message: it fires at
-// most once per window per conversation, which is the cost guard for the review model.
-func (g *FlowGate) reviewRecently(st *flowSession, now time.Time) bool {
-	if st == nil || st.lastReview.IsZero() {
-		return false
-	}
-	return now.Sub(st.lastReview) < g.window
-}
-
 // trimHits drops event timestamps older than the sliding window.
 func (g *FlowGate) trimHits(hits []time.Time, now time.Time) []time.Time {
 	cutoff := now.Add(-g.window)
@@ -407,27 +464,6 @@ func (g *FlowGate) trimHits(hits []time.Time, now time.Time) []time.Time {
 	return hits[:keep]
 }
 
-// applyReview keeps only the entries at the returned indices (validated, de-duped,
-// ascending) and returns the kept slice plus the dropped entries in original order.
-func (g *FlowGate) applyReview(buffer []queuedMsg, keep []int) (kept, dropped []queuedMsg) {
-	seen := make(map[int]bool, len(keep))
-	for _, i := range keep {
-		if i >= 0 && i < len(buffer) {
-			seen[i] = true
-		}
-	}
-	out := make([]queuedMsg, 0, len(buffer))
-	drop := make([]queuedMsg, 0)
-	for i, entry := range buffer {
-		if seen[i] {
-			out = append(out, entry)
-		} else {
-			drop = append(drop, entry)
-		}
-	}
-	return out, drop
-}
-
 // dropFront removes n entries from the front (oldest) and returns the rest plus the
 // dropped ones.
 func (g *FlowGate) dropFront(buffer []queuedMsg, n int) (kept, dropped []queuedMsg) {
@@ -437,27 +473,6 @@ func (g *FlowGate) dropFront(buffer []queuedMsg, n int) (kept, dropped []queuedM
 	return buffer[n:], append([]queuedMsg(nil), buffer[:n]...)
 }
 
-// bufferHasSeq reports whether an entry with the given sequence is still buffered
-// in the conversation's pending set.
-func bufferHasSeq(buffer []queuedMsg, seq uint64) bool {
-	for _, e := range buffer {
-		if e.seq == seq {
-			return true
-		}
-	}
-	return false
-}
-
-// containsSeq reports whether seq appears among the dropped entries.
-func containsSeq(dropped []queuedMsg, seq uint64) bool {
-	for _, e := range dropped {
-		if e.seq == seq {
-			return true
-		}
-	}
-	return false
-}
-
 // auditDropped records every dropped entry to the OnPruned audit hook.
 func (g *FlowGate) auditDropped(dropped []queuedMsg, reason string) {
 	for _, e := range dropped {

internal/channel/adapter/flowgate_async_test.go

@@ -0,0 +1,411 @@
+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)
+}

internal/channel/adapter/flowgate_parse.go

@@ -0,0 +1,94 @@
+package adapter
+
+import (
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+)
+
+// utf8BOM 是 UTF-8 的字节序标记。对端（审查模型网关）会把它原样回带，而
+// encoding/json 见到它报 `invalid character 'ï' looking for beginning of value` ——
+// 实测 9 天 1,909 次审查里 1,288 次（67%）因此解析失败，每一次都白烧一次模型调用
+// 并把「每窗口一次」的节流打穿（见 DevDocs/audit/群聊排队与回旧话-2026-10-24.md）。
+const utf8BOM = "\ufeff"
+
+// reviewExcerptLen 是解析失败时附带在错误里的模型输出长度（字节）。它只含模型
+// 返回的编号数组（数字/方括号/逗号/围栏），足以定位「又换了什么壳」，不至于把
+// 消息内容带进日志。
+const reviewExcerptLen = 40
+
+// ParseReviewIndices decodes the keep-index array a review model returned, rising
+// above the shapes real gateways produce instead of requiring a bare JSON array:
+//
+//   - a UTF-8 BOM prefix (`\ufeff`) — the actual cause of the 67% failure rate;
+//   - leading/trailing whitespace and newlines;
+//   - a Markdown code fence (` ```json … ``` ` or ` ``` … ``` `);
+//   - surrounding prose ("好的，保留以下编号：[0, 2]") — the outermost `[ … ]`
+//     span is used.
+//
+// The decoded indices are validated the same way the previous strict parser did:
+// out-of-range and duplicate values are dropped and the rest sorted ascending, so a
+// hallucinated number can never evict a real message. An empty or absent array is
+// NOT an error here — the caller decides; a missing/non-array answer is.
+func ParseReviewIndices(s string, bound int) ([]int, error) {
+	body, ok := reviewArraySpan(s)
+	if !ok {
+		return nil, fmt.Errorf("no JSON array in reviewer output %q", reviewExcerpt(s))
+	}
+	var idx []int
+	if err := json.Unmarshal([]byte(body), &idx); err != nil {
+		return nil, fmt.Errorf("decode reviewer indices %q: %w", reviewExcerpt(body), err)
+	}
+	seen := make(map[int]bool, len(idx))
+	out := make([]int, 0, len(idx))
+	for _, i := range idx {
+		if i >= 0 && i < bound && !seen[i] {
+			seen[i] = true
+			out = append(out, i)
+		}
+	}
+	sort.Ints(out)
+	return out, nil
+}
+
+// reviewArraySpan strips the decorations a model puts around its answer (BOM,
+// whitespace, code fence, prose) and returns the outermost JSON array span.
+func reviewArraySpan(s string) (string, bool) {
+	s = strings.TrimSpace(strings.TrimPrefix(s, utf8BOM))
+	s = strings.TrimSpace(unwrapCodeFence(s))
+	start := strings.IndexByte(s, '[')
+	end := strings.LastIndexByte(s, ']')
+	if start < 0 || end < start {
+		return "", false
+	}
+	return s[start : end+1], true
+}
+
+// unwrapCodeFence drops a leading ``` fence line and the closing ``` so the fenced
+// body is left for the array scan. Anything without a leading fence is returned
+// unchanged; a fence with no body is returned as-is and fails the array scan.
+func unwrapCodeFence(s string) string {
+	if !strings.HasPrefix(s, "```") {
+		return s
+	}
+	nl := strings.IndexByte(s, '\n')
+	if nl < 0 {
+		return s
+	}
+	s = s[nl+1:]
+	if close := strings.LastIndex(s, "```"); close >= 0 {
+		s = s[:close]
+	}
+	return s
+}
+
+// reviewExcerpt truncates text for an error message so the next parse failure is
+// diagnosable without logging the whole completion.
+func reviewExcerpt(s string) string {
+	s = strings.TrimSpace(s)
+	if len(s) <= reviewExcerptLen {
+		return s
+	}
+	return s[:reviewExcerptLen] + "…"
+}

internal/channel/adapter/flowgate_parse_test.go

@@ -0,0 +1,128 @@
+package adapter
+
+// 本文件用**对端原样输出**（含 UTF-8 BOM 字节）做夹具，固定审查输出的解析韧性。
+// 生产症状（`~/cy.log`，9 天）：`flow gate review output unparsable` 1,301 次，其中
+// 1,288 次的错误串逐字是 `invalid character 'ï' looking for beginning of value` ——
+// 对端在 JSON 数组前带了 BOM，而闸门用的 json.Unmarshal 不接受它。
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+// bomReviewFixture 是审查模型网关的实际回包形状：UTF-8 BOM（EF BB BF）紧接一个
+// JSON 数组。写死字节而不是用 \ufeff，是为了让「夹具被谁悄悄去掉 BOM」本身就会红。
+const bomReviewFixture = "\xef\xbb\xbf[0, 1, 2, 3, 4, 5]"
+
+// TestReviewBOMFixtureStillReproducesTheFailure 先钉住夹具的真实性：它必须以 BOM
+// 开头，且严格的 json.Unmarshal 必须在它上面失败，错误串必须就是生产日志里那条。
+//
+// 变异判据：把夹具里的 BOM 去掉 ⇒ 本用例红（同时也说明 BOM 相关用例已失去意义）。
+func TestReviewBOMFixtureStillReproducesTheFailure(t *testing.T) {
+	t.Parallel()
+
+	if !strings.HasPrefix(bomReviewFixture, "\xef\xbb\xbf") {
+		t.Fatal("BOM fixture lost its BOM bytes: it no longer reproduces the production failure")
+	}
+	var idx []int
+	err := json.Unmarshal([]byte(bomReviewFixture), &idx)
+	if err == nil {
+		t.Fatal("strict json.Unmarshal accepted the BOM fixture: the regression it models is gone")
+	}
+	if !strings.Contains(err.Error(), "invalid character 'ï'") {
+		t.Fatalf("strict parse error = %v, want the production symptom \"invalid character 'ï'\"", err)
+	}
+}
+
+// TestParseReviewIndices_ToleratesBOM 是修复的正面证据：同一份带 BOM 的原样回包
+// 现在能解析出编号。
+//
+// 变异判据：去掉 ParseReviewIndices 里的 reviewArraySpan（回到裸 json.Unmarshal）
+// ⇒ 本用例红。
+func TestParseReviewIndices_ToleratesBOM(t *testing.T) {
+	t.Parallel()
+
+	keep, err := ParseReviewIndices(bomReviewFixture, 6)
+	if err != nil {
+		t.Fatalf("ParseReviewIndices(BOM fixture) error = %v, want nil", err)
+	}
+	want := []int{0, 1, 2, 3, 4, 5}
+	if len(keep) != len(want) {
+		t.Fatalf("keep = %v, want %v", keep, want)
+	}
+	for i := range want {
+		if keep[i] != want[i] {
+			t.Fatalf("keep = %v, want %v", keep, want)
+		}
+	}
+}
+
+// TestParseReviewIndices_ToleratesDecorations 覆盖模型常见的几种外壳：首尾空白与换行、
+// Markdown 代码围栏（带或不带语言标签）、以及围在数组外面的解说文字。
+func TestParseReviewIndices_ToleratesDecorations(t *testing.T) {
+	t.Parallel()
+
+	cases := map[string]string{
+		"newlines":        "\n\n[0, 2]\n\n",
+		"fence":           "```\n[0, 2]\n```",
+		"fence with lang": "```json\n[0, 2]\n```",
+		"prose prefix":    "好的，保留以下编号：[0, 2]",
+		"prose suffix":    "[0, 2] 以上是保留的编号。",
+		"bom plus fence":  "\xef\xbb\xbf```json\n[0, 2]\n```",
+		"spaced values":   "[ 0 , 2 ]",
+	}
+	for name, raw := range cases {
+		keep, err := ParseReviewIndices(raw, 3)
+		if err != nil {
+			t.Fatalf("%s: error = %v, want nil", name, err)
+		}
+		if len(keep) != 2 || keep[0] != 0 || keep[1] != 2 {
+			t.Fatalf("%s: keep = %v, want [0 2]", name, keep)
+		}
+	}
+}
+
+// TestParseReviewIndices_FiltersHallucinations 固定过滤口径与修复前一致：越界、负数、
+// 重复的编号一律丢弃（模型幻觉出的数字绝不能淘汰真实消息），结果升序。
+func TestParseReviewIndices_FiltersHallucinations(t *testing.T) {
+	t.Parallel()
+
+	keep, err := ParseReviewIndices("\xef\xbb\xbf[2, 2, 99, -1, 0]", 3)
+	if err != nil {
+		t.Fatalf("error = %v, want nil", err)
+	}
+	if len(keep) != 2 || keep[0] != 0 || keep[1] != 2 {
+		t.Fatalf("keep = %v, want [0 2]", keep)
+	}
+}
+
+// TestParseReviewIndices_EmptyArrayIsNotAnError：模型明确回空数组 = 「一条都不保留」，
+// 这是有效结论（全部当噪声），不是解析失败 —— 失败方向必须留给真正的畸形输出。
+func TestParseReviewIndices_EmptyArrayIsNotAnError(t *testing.T) {
+	t.Parallel()
+
+	keep, err := ParseReviewIndices("\xef\xbb\xbf[]", 5)
+	if err != nil {
+		t.Fatalf("error = %v, want nil", err)
+	}
+	if len(keep) != 0 {
+		t.Fatalf("keep = %v, want empty", keep)
+	}
+}
+
+// TestParseReviewIndices_RejectsGarbage：真正畸形的输出仍然报错（调用方据此走
+// 「不额外丢弃、也不提权」的安全方向），错误里带一小段原文便于定位下一次变种。
+func TestParseReviewIndices_RejectsGarbage(t *testing.T) {
+	t.Parallel()
+
+	for name, raw := range map[string]string{
+		"no array":     "抱歉，我无法判断。",
+		"empty":        "",
+		"array of str": `["a", "b"]`,
+	} {
+		if _, err := ParseReviewIndices(raw, 5); err == nil {
+			t.Fatalf("%s: error = nil, want a decode error", name)
+		}
+	}
+}

internal/channel/adapter/flowgate_review.go

@@ -0,0 +1,171 @@
+package adapter
+
+// 本文件是策略四的「审查」半边：把它从准入路径上摘下来的派发器（reviewJob /
+// scheduleReviewLocked / runReview）与三重上界（同适配器在飞、每会话窗口、每适配器
+// 窗口预算）。解析韧性在 flowgate_parse.go，准入判定仍在 flowgate.go。
+//
+// 为什么必须是后台：`noise_review` 实测 p50 5,294 ms / p95 6,651 ms / max 9,204 ms
+// （`~/cy.log`，9 天 1,900 次），而五路入站里 qq-ws、kook-ws、weixin 长轮询都是
+// 「单读 goroutine + 同步派发」—— 同步跑在读循环上等于把消息、心跳、探活回执全排在
+// 一次 6 秒的模型调用后面（DevDocs/audit/群聊排队与回旧话-2026-10-24.md）。
+//
+// # 与修复前的语义差异（哪一条变了、哪一条没变）
+//
+//   - `rate_limited`（策略三，门口丢）：**一字不改**。超窗消息在准入当次就被拒绝、
+//     根本不进待处理集合，也不经过审查。
+//   - 上限修剪（策略四的硬部分）：**一字不改**。待处理集合超过 maxMessages 时，最旧的
+//     条目仍在**准入当次同步**被挤掉并审计为 "capped"；触发水位仍是上限的 80%。
+//   - 审查修剪（策略四的模型部分）：**判据不变，时机变了**。触发条件（≥80% 上限、
+//     同一会话每窗口至多一次、返回保留编号、越界/重复编号丢弃）逐条沿用；变的是结论
+//     落地时点 —— 它不再决定「触发它的那条消息」是否放行，而是修剪**它当时看到的**
+//     那批历史条目（审计为 "review_pruned"）。
+//     直接后果：`review_pruned` 不再从门口丢消息（实测原本 41–74 条/日），那些消息
+//     现在会进入流水线 —— 完整性↑、噪声↑，是否开口仍由 engage_judge / 冷避 / owner
+//     闸判（这三处判据一行未动）。
+//
+// 为什么这个差异可以接受：design/12 §6 策略四给审查的职责是「保护 loop 不被不重要的
+// 消息塞满」的历史积压去噪，不是逐条消息的准入判决；把一次 6 秒的模型调用钉在准入
+// 路径上，代价（读循环阻塞 → 消息/心跳/探活全排队）远大于收益（少进 41–74 条噪声/日）。
+// 真正决定「进不进 loop」的两道判据 —— 限流与上限 —— 都留在准入路径上，且未改口径。
+
+import (
+	"context"
+	"time"
+)
+
+const (
+	// reviewTimeout bounds one background reviewer call. The in-flight slot is held
+	// for the whole call, so an unbounded call would also freeze every other
+	// conversation's review for that account. 上一版没有上界，最长一次实测 9.2 s；
+	// 30 s 留足余量而不至于让一次挂死的调用长期占位。
+	reviewTimeout = 30 * time.Second
+)
+
+// reviewJob is one scheduled reviewer call: the snapshot of the pending set the
+// reviewer is asked about, plus the sequence numbers that identify those entries
+// when the answer comes back (the buffer keeps moving while the call is in flight).
+type reviewJob struct {
+	key  string
+	msgs []UnifiedMessage
+	seqs []uint64
+}
+
+// scheduleReviewLocked reports whether this admission should start a background
+// review and, when it should, marks that review as started and returns its job.
+//
+// 三重上界都在这里（调用方须持有 g.mu）：
+//  1. 同一适配器至多一次在飞（reviewInFlight）—— 后台审查有界，不会因为多个群
+//     同时越阈而并发起 N 个模型调用；
+//  2. 同一会话每 window 至多一次（lastReview）—— 修复前它只在成功分支置位，
+//     于是 68% 解析失败时「每窗口一次」退化成「每条消息一次」；
+//  3. 整个适配器每 window 至多 reviewBudget 次（reviewHits）—— 与模型耗时无关的
+//     硬上界，会话再多也乘不上去。
+//
+// lastReview 在这里（**派发时**）写入，而不是等调用返回：成功与失败两条路径因此
+// 同样被节流，这是「每条入站帧都付一次 6 秒」的正面修复。
+func (g *FlowGate) scheduleReviewLocked(st *flowSession, key string, now time.Time) *reviewJob {
+	if g.review == nil || g.maxMessages <= 0 || len(st.buffer) < g.reviewAt {
+		return nil
+	}
+	if g.reviewInFlight || g.reviewRecently(st, now) || !g.reviewBudgetLocked(now) {
+		return nil
+	}
+	st.lastReview = now
+	g.reviewInFlight = true
+	g.reviewHits = append(g.reviewHits, now)
+	return &reviewJob{key: key, msgs: queueMessages(st.buffer), seqs: bufferSeqs(st.buffer)}
+}
+
+// runReview executes one scheduled review off the admission path and folds its
+// verdict back into the conversation's pending set. It is the only place the
+// Reviewer is called.
+//
+// 失败方向（安全侧，与修复前 `err != nil` 分支同口径）：审查失败**既不额外丢弃**
+// （失败绝不被解释成「低价值 ⇒ 丢掉」），**也不给任何条目保留特权**（失败绝不被
+// 解释成「高价值 ⇒ 提权」）；待处理集合原样保留，只由既有的 FIFO 上限兜底。
+// 唯一的代价是这一批噪声没被修剪，而窗口节流保证下一次尝试要等到下个窗口。
+func (g *FlowGate) runReview(ctx context.Context, job *reviewJob) {
+	rctx, cancel := reviewContext(ctx)
+	defer cancel()
+	keep, err := g.review(rctx, job.msgs)
+
+	g.mu.Lock()
+	g.reviewInFlight = false
+	var dropped []queuedMsg
+	if err == nil {
+		// 会话可能在审查期间被 GC 掉（空闲清理），那时结论无处可落，直接丢弃。
+		if st := g.sessions[job.key]; st != nil {
+			st.buffer, dropped = applyReviewSeqs(st.buffer, job.seqs, keep)
+		}
+	}
+	g.mu.Unlock()
+
+	// 审计在锁外：钩子由 channelbridge 接（落库/面板），不该占着闸门的锁跑 I/O。
+	g.auditDropped(dropped, "review_pruned")
+}
+
+// reviewContext detaches the reviewer call from the admission caller's cancellation
+// — the inbound handler returns the moment the message is admitted, and on several
+// adapters its ctx dies with the frame — while still bounding the call so a hung
+// model cannot pin the single in-flight slot forever.
+func reviewContext(ctx context.Context) (context.Context, context.CancelFunc) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return context.WithTimeout(context.WithoutCancel(ctx), reviewTimeout)
+}
+
+// reviewRecently throttles review so it does not run on every message: it fires at
+// most once per window per conversation, which is the cost guard for the review model.
+func (g *FlowGate) reviewRecently(st *flowSession, now time.Time) bool {
+	if st == nil || st.lastReview.IsZero() {
+		return false
+	}
+	return now.Sub(st.lastReview) < g.window
+}
+
+// reviewBudgetLocked enforces the per-adapter ceiling on reviewer calls started in
+// the current window. 调用方须持有 g.mu。
+func (g *FlowGate) reviewBudgetLocked(now time.Time) bool {
+	g.reviewHits = g.trimHits(g.reviewHits, now)
+	return len(g.reviewHits) < g.reviewBudget
+}
+
+// applyReviewSeqs keeps only the buffered entries the reviewer chose. Entries are
+// matched by sequence number, so:
+//   - entries evicted by the cap while the call was in flight simply do not appear
+//     in the buffer any more (they were audited as "capped" at eviction time);
+//   - entries that arrived **after** the snapshot are never judged by this review
+//     and always survive (the reviewer never saw them).
+func applyReviewSeqs(buffer []queuedMsg, seqs []uint64, keep []int) (kept, dropped []queuedMsg) {
+	reviewed := make(map[uint64]bool, len(seqs))
+	for _, s := range seqs {
+		reviewed[s] = true
+	}
+	keepSet := make(map[uint64]bool, len(keep))
+	for _, i := range keep {
+		if i >= 0 && i < len(seqs) {
+			keepSet[seqs[i]] = true
+		}
+	}
+	kept = make([]queuedMsg, 0, len(buffer))
+	dropped = make([]queuedMsg, 0)
+	for _, e := range buffer {
+		if !reviewed[e.seq] || keepSet[e.seq] {
+			kept = append(kept, e)
+			continue
+		}
+		dropped = append(dropped, e)
+	}
+	return kept, dropped
+}
+
+// bufferSeqs extracts the sequence numbers of buffered entries in order, so a
+// review verdict can be applied to exactly the entries it was asked about.
+func bufferSeqs(buffer []queuedMsg) []uint64 {
+	out := make([]uint64, len(buffer))
+	for i, e := range buffer {
+		out[i] = e.seq
+	}
+	return out
+}

internal/channel/adapter/flowgate_test.go

@@ -2,6 +2,7 @@ package adapter
 
 import (
 	"context"
+	"sync"
 	"testing"
 	"time"
 )
@@ -118,10 +119,16 @@ func TestFlowGate_MsgPerMinCapsAllNonOwner(t *testing.T) {
 	}
 }
 
-func TestFlowGate_ReviewPrunesWhenNearCap(t *testing.T) {
+// TestFlowGate_ReviewPrunesBacklogAfterAdmission 固定策略四的新口径：审查**不在准入
+// 路径上**。越阈那一条消息照常放行（它的到达只负责派发审查），审查完成后只修剪
+// **它当时看到的那批**待处理条目，并以 "review_pruned" 入审计。
+//
+// 变异判据：把审查改回在 Admit 里同步执行并按结论丢「刚到的那条」⇒ 本用例的
+// `v.Proceed` 断言红。
+func TestFlowGate_ReviewPrunesBacklogAfterAdmission(t *testing.T) {
 	t.Parallel()
 	now, _ := fixedClock(time.Unix(1000, 0))
-	var audited []string
+	audit := &auditRecorder{}
 	g := NewFlowGate(FlowGateOptions{
 		MaxMessages: 10,
 		Window:      time.Minute,
@@ -130,9 +137,7 @@ func TestFlowGate_ReviewPrunesWhenNearCap(t *testing.T) {
 			// Keep only the first two messages; drop the rest as noise.
 			return []int{0, 1}, nil
 		},
-		OnPruned: func(_ UnifiedMessage, reason string) {
+		OnPruned: audit.hook(),
-			audited = append(audited, reason)
-		},
 	})
 
 	// Seven messages stay below the 80% threshold (reviewAt = 8), no review fires.
@@ -141,69 +146,87 @@ func TestFlowGate_ReviewPrunesWhenNearCap(t *testing.T) {
 			t.Fatalf("message %d should be admitted before threshold: %+v", i, v)
 		}
 	}
-	if len(audited) != 0 {
+	waitReviewIdle(t, g)
-		t.Fatalf("review fired below threshold: %v", audited)
+	if n := len(audit.snapshot()); n != 0 {
+		t.Fatalf("review fired below threshold: %d audits", n)
 	}
 
-	// The 8th message crosses the 80% threshold → review runs and drops msg2..7,
+	// The 8th message crosses the 80% threshold → the review is dispatched in the
-	// including the just-admitted message 7.
+	// background; the message itself is still admitted (admission never waits).
 	v := g.Admit(context.Background(), groupMsg(7))
-	if v.Proceed {
+	if !v.Proceed {
-		t.Fatal("just-admitted message 7 must be pruned by the reviewer")
+		t.Fatalf("the message that crosses the threshold must still be admitted, got %+v", v)
 	}
-	if v.Reason != "review_pruned" {
+	if v.Reason != "" {
-		t.Fatalf("reason = %q, want review_pruned", v.Reason)
+		t.Fatalf("reason = %q, want empty", v.Reason)
 	}
-	if len(v.Pruned) != 6 {
+	waitReviewIdle(t, g)
-		t.Fatalf("pruned = %d, want 6 (msg2..msg7)", len(v.Pruned))
+
+	// The review keeps 0 and 1, so msg2..7 are pruned from the pending set and audited.
+	if n := len(audit.snapshot()); n != 6 {
+		t.Fatalf("audit count = %d, want 6 (msg2..msg7)", n)
 	}
-	if len(audited) != 6 {
+	for _, e := range audit.snapshot() {
-		t.Fatalf("audit count = %d, want 6", len(audited))
+		if e.reason != "review_pruned" {
-	}
+			t.Fatalf("audit reason = %q, want review_pruned", e.reason)
-	for _, r := range audited {
-		if r != "review_pruned" {
-			t.Fatalf("audit reason = %q, want review_pruned", r)
 		}
 	}
+	if n := pendingLen(g, ConversationKey(groupMsg(0))); n != 2 {
+		t.Fatalf("pending set = %d, want 2 (only the kept entries)", n)
+	}
 }
 
+// TestFlowGate_ReviewThrottledWithinWindow 固定「每会话每窗口至多一次审查」。
+// 审查者保留全部条目，使待处理集合恒在阈值上 —— 于是节流（而不是队列长度）是唯一的
+// 抑制因素；这正是修复前失守的那条判据（lastReview 只在成功分支置位）。
 func TestFlowGate_ReviewThrottledWithinWindow(t *testing.T) {
 	t.Parallel()
 	now, clock := fixedClock(time.Unix(1000, 0))
+	var mu sync.Mutex
 	reviews := 0
 	g := NewFlowGate(FlowGateOptions{
 		MaxMessages: 3,
 		Window:      time.Minute,
 		Now:         now,
-		Review: func(_ context.Context, msgs []UnifiedMessage) ([]int, error) {
+		Review: func(ctx context.Context, msgs []UnifiedMessage) ([]int, error) {
+			mu.Lock()
 			reviews++
-			return []int{0}, nil
+			mu.Unlock()
+			return keepAll(ctx, msgs)
 		},
 	})
+	count := func() int {
+		mu.Lock()
+		defer mu.Unlock()
+		return reviews
+	}
 
 	// First threshold crossing triggers a review.
 	for i := 0; i < 3; i++ {
 		g.Admit(context.Background(), groupMsg(i))
 	}
-	if reviews != 1 {
+	waitReviewIdle(t, g)
-		t.Fatalf("reviews after crossing threshold = %d, want 1", reviews)
+	if got := count(); got != 1 {
+		t.Fatalf("reviews after crossing threshold = %d, want 1", got)
 	}
 
 	// Within the same window the gate must not re-review even when it refills.
-	for i := 3; i < 6; i++ {
+	for i := 3; i < 9; i++ {
 		g.Admit(context.Background(), groupMsg(i))
 	}
-	if reviews != 1 {
+	waitReviewIdle(t, g)
-		t.Fatalf("reviews within window = %d, want 1 (throttled)", reviews)
+	if got := count(); got != 1 {
+		t.Fatalf("reviews within window = %d, want 1 (throttled)", got)
 	}
 
 	// A fresh window allows a review again.
 	*clock = clock.Add(2 * time.Minute)
-	for i := 6; i < 9; i++ {
+	for i := 9; i < 12; i++ {
 		g.Admit(context.Background(), groupMsg(i))
 	}
-	if reviews != 2 {
+	waitReviewIdle(t, g)
-		t.Fatalf("reviews after window roll = %d, want 2", reviews)
+	if got := count(); got != 2 {
+		t.Fatalf("reviews after window roll = %d, want 2", got)
 	}
 }
 

internal/services/channelbridge/flowgate.go

@@ -2,10 +2,8 @@ package channelbridge
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"log/slog"
-	"sort"
 	"strings"
 
 	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
@@ -201,22 +199,14 @@ func truncateText(s string, n int) string {
 	return string(runes[:n]) + "…"
 }
 
-// parseKeepIndices 解码审查模型返回的 JSON 索引数组，并按升序
+// parseKeepIndices 解码审查模型返回的编号数组，并按升序只保留范围内、去重后的值。
-// 只保留范围内、去重后的值。越界或重复的
+// 越界或重复的索引被丢弃，因此模型幻觉出的数字绝不会淘汰真实消息。
-// 索引被丢弃，因此模型幻觉出的数字绝不会淘汰真实消息。
+//
+// 解析本身在 adapter.ParseReviewIndices（解析属于闸门的契约，和 Reviewer 类型同一个包），
+// 它容忍对端回带的东西：**UTF-8 BOM**、首尾空白/换行、Markdown 代码围栏、数组外的解说
+// 文字。修复前这里是裸 json.Unmarshal，实测 9 天 1,909 次审查里 1,301 次解析失败
+// （其中 1,288 次错误串逐字是 `invalid character 'ï' looking for beginning of value`），
+// 每次失败都白烧一次模型调用 —— 见 DevDocs/audit/群聊排队与回旧话-2026-10-24.md。
 func parseKeepIndices(s string, bound int) ([]int, error) {
-	var idx []int
+	return adapter.ParseReviewIndices(s, bound)
-	if err := json.Unmarshal([]byte(s), &idx); err != nil {
-		return nil, err
-	}
-	seen := make(map[int]bool, len(idx))
-	out := make([]int, 0, len(idx))
-	for _, i := range idx {
-		if i >= 0 && i < bound && !seen[i] {
-			seen[i] = true
-			out = append(out, i)
-		}
-	}
-	sort.Ints(out)
-	return out, nil
 }

internal/services/channelbridge/flowgate_test.go

@@ -3,6 +3,7 @@ package channelbridge
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"log/slog"
 	"strings"
 	"testing"
@@ -329,3 +330,88 @@ func TestParseKeepIndices(t *testing.T) {
 		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)
+	}
+}