From 76d972f0f299003916dd47cede2c74d978409223 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 24 Oct 2026 19:43:57 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(channel/flowgate):=20=E5=AE=A1=E6=9F=A5?= =?UTF-8?q?=E7=A7=BB=E5=87=BA=E5=87=86=E5=85=A5=E8=B7=AF=E5=BE=84=E3=80=81?= =?UTF-8?q?=E8=8A=82=E6=B5=81=E8=A6=86=E7=9B=96=E5=A4=B1=E8=B4=A5=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E3=80=81BOM=20=E8=A7=A3=E6=9E=90=E9=9F=A7=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Admit 不再同步调用审查模型：触发时只把审查**派发**到后台 goroutine，准入只读 当前已有结论。三重上界：同一适配器至多一次在飞（reviewInFlight）、同一会话每窗口 至多一次（lastReview）、整个适配器每窗口至多 defaultReviewPerWindow=6 次（reviewHits， 写成可测断言）。 - st.lastReview 改为在**派发时**置位 ⇒ 成功与失败两条路径同样被节流。修复前它只在 err == nil 分支置位，而实测 68% 的审查因对端回带 BOM 解析失败 ⇒ 缓冲满后每条入站帧 都付一次 ~6s 的模型调用（qq-ws/kook-ws/weixin-poll 三路单读 goroutine 全被钉住）。 - 新增 adapter.ParseReviewIndices：容忍 UTF-8 BOM、首尾空白/换行、Markdown 代码围栏、 数组外解说文字；越界/重复编号照旧丢弃。channelbridge.parseKeepIndices 改为委派它 （仅函数体 + 注释，未动限流/丢弃/审查触发语义）。 - 语义差异（已写进 flowgate_review.go 文件头）：rate_limited 的门口丢与 FIFO 上限 一字不改；审查修剪降级为后台积压去噪，只修剪它当时看到的那批历史条目 —— review_pruned 不再从门口丢消息（实测 41–74 条/日），完整性↑、噪声↑，是否开口仍由 engage_judge/冷避/owner 闸判（三处判据未动）。 - cmd/cyrene：逐账号启动日志如实报出生效的入口流控（塔裁决 A：零行为改动、不翻默认）， 让「没配流控键的账号闸门仍开着」一眼可见。 - 测试：4s 慢审查替身证明入站帧不被阻塞（变异回同步 ⇒ 红）、节流在失败路径生效 （摘掉派发时置位 ⇒ 红）、每适配器窗口预算上界（摘掉预算 ⇒ 红）、同适配器单飞 （摘掉 in-flight ⇒ 红）、在飞期间到达的消息不被连带修剪、带 BOM 字节的夹具 （夹具去掉 BOM ⇒ 红）。闸门用例 -race 通过。 --- cmd/cyrene/channels.go | 65 ++- internal/channel/adapter/flowgate.go | 211 ++++----- .../channel/adapter/flowgate_async_test.go | 411 ++++++++++++++++++ internal/channel/adapter/flowgate_parse.go | 94 ++++ .../channel/adapter/flowgate_parse_test.go | 128 ++++++ internal/channel/adapter/flowgate_review.go | 171 ++++++++ internal/channel/adapter/flowgate_test.go | 85 ++-- internal/services/channelbridge/flowgate.go | 28 +- 8 files changed, 1043 insertions(+), 150 deletions(-) create mode 100644 internal/channel/adapter/flowgate_async_test.go create mode 100644 internal/channel/adapter/flowgate_parse.go create mode 100644 internal/channel/adapter/flowgate_parse_test.go create mode 100644 internal/channel/adapter/flowgate_review.go diff --git a/cmd/cyrene/channels.go b/cmd/cyrene/channels.go index aeb68fe..38ccf0a 100644 --- a/cmd/cyrene/channels.go +++ b/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__* 适配器，加上 // 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 diff --git a/internal/channel/adapter/flowgate.go b/internal/channel/adapter/flowgate.go index 7ba0914..e91323c 100644 --- a/internal/channel/adapter/flowgate.go +++ b/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" - // or "capped". It is empty when Proceed is true. + // Reason is a short token describing a drop. After M24 it is always + // "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 - // review path it is the noise the reviewer rejected; on the cap path it is the - // oldest messages evicted to honor the total cap. Each is recorded to the - // OnPruned audit hook before being dropped, so nothing is lost silently. + // Pruned lists every message the gate dropped during this admission: on the cap + // path it is the oldest messages evicted to honor the total cap. Each is recorded + // to the 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 - // FIFO. Zero disables the cap. + // is invoked — in the background — to prune noise; with no Reviewer the oldest + // 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 - admitted uint64 - seq uint64 + // reviewHits 是本适配器在窗口内**派发**审查的时刻，用于账号级预算 + // （defaultReviewPerWindow）。 + 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 { - g.reviewAt = g.maxMessages * defaultReviewPct / 100 - if g.reviewAt < 1 { - g.reviewAt = 1 - } + if g.reviewBudget <= 0 { + g.reviewBudget = defaultReviewPerWindow } + 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 -// inbound message. It returns a verdict telling the caller whether to proceed; -// messages that the gate drops are returned (and audited) so nothing is lost -// silently. +// Admit applies the rate limits then the per-conversation cap gate to one inbound +// message. It returns a verdict telling the caller whether to proceed; messages +// that the gate drops are returned (and audited) so nothing is lost 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 - // nears the cap. + // 策略四（会话级）: add to this conversation's pending set, then enforce 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) { - reason := "capped" - if containsSeq(dropped, seq) { - reason = "review_pruned" - } - v := GateVerdict{Proceed: false, Reason: reason, Pruned: queueMessages(dropped)} - c := msg - v.DroppedMsg = &c - return v - } - return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)} + // 本条一定还在缓冲里：上限按新旧顺序丢最旧，而它是最新的（maxMessages ≥ 1）。 + // 于是修复后准入路径上唯一能让消息 Proceed=false 的判据就是上面的限流 —— 审查 + // （已移到后台）与上限（丢最旧）都只影响**已经放行过的**历史条目，只留审计痕迹。 + return GateVerdict{Proceed: true, Pruned: queueMessages(dropped)}, job } // 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 { diff --git a/internal/channel/adapter/flowgate_async_test.go b/internal/channel/adapter/flowgate_async_test.go new file mode 100644 index 0000000..55eb86f --- /dev/null +++ b/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) +} diff --git a/internal/channel/adapter/flowgate_parse.go b/internal/channel/adapter/flowgate_parse.go new file mode 100644 index 0000000..dc0fe59 --- /dev/null +++ b/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] + "…" +} diff --git a/internal/channel/adapter/flowgate_parse_test.go b/internal/channel/adapter/flowgate_parse_test.go new file mode 100644 index 0000000..f0d48dc --- /dev/null +++ b/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) + } + } +} diff --git a/internal/channel/adapter/flowgate_review.go b/internal/channel/adapter/flowgate_review.go new file mode 100644 index 0000000..2ab64d7 --- /dev/null +++ b/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 +} diff --git a/internal/channel/adapter/flowgate_test.go b/internal/channel/adapter/flowgate_test.go index 9b83e4d..184536b 100644 --- a/internal/channel/adapter/flowgate_test.go +++ b/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) { - audited = append(audited, reason) - }, + OnPruned: audit.hook(), }) // 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 { - t.Fatalf("review fired below threshold: %v", audited) + waitReviewIdle(t, g) + 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, - // including the just-admitted message 7. + // The 8th message crosses the 80% threshold → the review is dispatched in the + // background; the message itself is still admitted (admission never waits). v := g.Admit(context.Background(), groupMsg(7)) - if v.Proceed { - t.Fatal("just-admitted message 7 must be pruned by the reviewer") + if !v.Proceed { + t.Fatalf("the message that crosses the threshold must still be admitted, got %+v", v) } - if v.Reason != "review_pruned" { - t.Fatalf("reason = %q, want review_pruned", v.Reason) + if v.Reason != "" { + t.Fatalf("reason = %q, want empty", v.Reason) } - if len(v.Pruned) != 6 { - t.Fatalf("pruned = %d, want 6 (msg2..msg7)", len(v.Pruned)) + waitReviewIdle(t, g) + + // 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 { - t.Fatalf("audit count = %d, want 6", len(audited)) - } - for _, r := range audited { - if r != "review_pruned" { - t.Fatalf("audit reason = %q, want review_pruned", r) + for _, e := range audit.snapshot() { + if e.reason != "review_pruned" { + t.Fatalf("audit reason = %q, want review_pruned", e.reason) } } + 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 { - t.Fatalf("reviews after crossing threshold = %d, want 1", reviews) + waitReviewIdle(t, g) + 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 { - t.Fatalf("reviews within window = %d, want 1 (throttled)", reviews) + waitReviewIdle(t, g) + 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 { - t.Fatalf("reviews after window roll = %d, want 2", reviews) + waitReviewIdle(t, g) + if got := count(); got != 2 { + t.Fatalf("reviews after window roll = %d, want 2", got) } } diff --git a/internal/services/channelbridge/flowgate.go b/internal/services/channelbridge/flowgate.go index bb6135b..a38739c 100644 --- a/internal/services/channelbridge/flowgate.go +++ b/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 - 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 + return adapter.ParseReviewIndices(s, bound) } From a25e0c8870e7e3469036b25411dd912ae428e515 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sat, 24 Oct 2026 19:49:58 +0800 Subject: [PATCH 2/2] =?UTF-8?q?test(channelbridge):=20=E9=94=81=E4=BD=8F?= =?UTF-8?q?=E5=AE=A1=E6=9F=A5=E8=A7=A3=E6=9E=90=E7=9A=84=E6=8E=A5=E7=BA=BF?= =?UTF-8?q?=EF=BC=88BOM=20=E9=9F=A7=E6=80=A7=E5=BF=85=E9=A1=BB=E5=9C=A8?= =?UTF-8?q?=E7=94=9F=E4=BA=A7=E4=BE=A7=E7=94=9F=E6=95=88=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 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），未改动既有用例的任何断言。 --- .../services/channelbridge/flowgate_test.go | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) diff --git a/internal/services/channelbridge/flowgate_test.go b/internal/services/channelbridge/flowgate_test.go index bc5f94d..2fe93a1 100644 --- a/internal/services/channelbridge/flowgate_test.go +++ b/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) + } +}