package channelbridge import ( "context" "fmt" "log/slog" "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) // gateFor 返回适配器的按账号流控闸门,首次使用时依据适配器的 // Config 快照惰性构建。闸门以适配器名称为键,使每个账号 // (platform:confId)保有自己、可跨适配器重连存活的闸门; // **闸门内部的节流计数器按会话隔离** // (adapter.ConversationKey:platform+channelID+groupID/peerID),因此两个群绝不会 // 相互挤占(design/12 §6 策略三/四,裁决 06.4.3)。 func (b *Bridge) gateFor(a adapter.PlatformAdapter) *adapter.FlowGate { if a == nil { // 便捷路径(process 用平台名找适配器)在平台上未注册任何账号时给出 nil: // 没有账号就没有闸门,调用方按 nil 跳过(生产路径的 a 恒非 nil)。 return nil } name := a.Name() b.gateMu.Lock() defer b.gateMu.Unlock() if g, ok := b.gates[name]; ok { return g } g := adapter.NewFlowGate(b.gateOptions(a)) b.gates[name] = g return g } // gateOptions 把适配器 Config 快照中的按账号流控限制解析进 // FlowGate。群聊/总量/上限键缺失(单账号环境变量适配器)时限制关闭, // 既有渠道行为不变;键存在且值为 0 时采用保护性默认值 // (每群每分钟 6 条、上限 50),负值则关闭限制—— // 这是对 CHANNELS_JSON 账号的向后兼容约定(design/12 §6 实现要点)。 // // **私聊(陌生人)限流走 directMsgPerMin**(主人 2026-10-05 裁决③与裁决 13 // 「流控时间要可配置」):它与群聊同一条读法,但缺省/0 一律取保护性默认 // FlowDirectDefault —— 原本没有任何配置的账号(如 qq-obv11)对陌生人私聊是完全 // 敞口,而裁决要求「陌生人必须有流控」,所以这里绝不能因为「没配」而静默关掉。 // 主人私聊在 onMessage 就被绕过本闸门,不受影响。 func (b *Bridge) gateOptions(a adapter.PlatformAdapter) adapter.FlowGateOptions { cfg := a.Config() opts := adapter.FlowGateOptions{ Review: b.review, OnPruned: b.onRejected, Now: b.now, DirectMsgPerMin: directLimit(cfg, b.flowLog), } if grp, ok := cfgInt(cfg, "groupMsgPerMin"); ok { opts.GroupMsgPerMin = flowOrDefault(grp, adapter.FlowLimitDefault) } if tot, ok := cfgInt(cfg, "msgPerMin"); ok && tot > 0 { opts.MsgPerMin = tot } if max, ok := cfgInt(cfg, "maxMessages"); ok { opts.MaxMessages = flowOrDefault(max, adapter.FlowCapDefault) } return opts } // directLimit 解析账号级 directMsgPerMin(非主人私聊 = 陌生人的每分钟每发送者上限), // 语义与 flowOrDefault 逐字同口径:**键缺失、非整数或 0 ⇒ 保护性默认 // adapter.FlowDirectDefault;正值 ⇒ 该值;负值 ⇒ 0(关闭该限制)**。 // // 与群聊的唯一区别是缺省侧:缺省绝不解释成「关闭」,否则一个漏配的键就会 // 把陌生人私聊重新变成全敞口(主人 2026-10-05 裁决③的既定裁决)。负值关闭 // 只有手写负数才能到达,属主人的**显式**选择,因此这里留一条带 source 的 WARN // 日志 —— 流控被关必须在日志里可追溯,不许静默发生。 func directLimit(cfg map[string]any, log *slog.Logger) int { v, ok := cfgInt(cfg, directMsgPerMinKey) if !ok { return adapter.FlowDirectDefault } if v < 0 { if log != nil { log.Warn("non-owner private rate limit turned off by account config", "key", directMsgPerMinKey, "value", v) } return 0 } return flowOrDefault(v, adapter.FlowDirectDefault) } // flowOrDefault 把配置的限制值映射为实际生效的限制:负值 // 关闭限制(0),零采用保护性默认值,正值则 // 固定为该值。这就是「没有该字段的旧账号保持原有行为或 // 取安全默认值」的约定。 func flowOrDefault(v, def int) int { switch { case v < 0: return 0 case v == 0: return def default: return v } } // cfgInt 从适配器配置 map 读取整数字段,键缺失或类型不符时 // 返回 (0,false)(因此缺失的字段保持「关闭」)。 func cfgInt(m map[string]any, key string) (int, bool) { if m == nil { return 0, false } switch v := m[key].(type) { case int: return v, true case int64: return int(v), true case float64: return int(v), true default: return 0, false } } const ( // directMsgPerMinKey 是账号级「非主人私聊(陌生人)每分钟每发送者上限」的配置键 // (design/12 §6 策略三;主人 2026-10-05 裁决 13)。它与 groupMsgPerMin 同一条读法: // 由 cmd/cyrene 的 namedAdapter 从账号字段透出到 Config(),桥在这里解析。 directMsgPerMinKey = "directMsgPerMin" // flowgateLogSource 是本文件日志的来源标记,使「流控解析结果」跨模块可定位 // (项目规范 §五:禁止无来源裸日志)。 flowgateLogSource = "services/channelbridge/flowgate" // reviewMaxTokens 限定审查模型的输出,使冗长的队列清单绝不会 // 产出超长补全(design/12 §6 策略四 成本护栏)。 reviewMaxTokens = 256 // reviewSnippetLen 是展示给审查者的每条消息内容字节数, // 足以判断重要性而不泄露整个队列。 reviewSnippetLen = 80 // reviewPrompt 指示审查模型挑选哪些队列消息重要。 // 它只判断噪声/重要性——绝不判断发送者是谁(项目 // 红线:身份来自 actor 元数据,而非 LLM)。 reviewPrompt = "你是群聊消息去噪助手。下面列出若干进入队列的消息(每条带编号)。" + "请判断哪些消息没有保留价值(灌水、无关、低信息量),只返回你认为【有必要保留】的编号 JSON 数组。" + "只判断信息价值,不判断发送者身份;不要编造编号;只输出 JSON 数组。" ) // NewLLMReviewer 把 model.Adapter 包装成 adapter.Reviewer(design/12 §6 策略四)。 // 它用一次轻量单轮提示问 LLM 哪些队列消息值得保留, // 并把答案回送给闸门。闸门只在待处理集合接近上限时 // 调用它,因此绝不会按消息调用 LLM。 func NewLLMReviewer(llm model.Adapter, log *slog.Logger) adapter.Reviewer { if log == nil { log = slog.Default() } return func(ctx context.Context, msgs []adapter.UnifiedMessage) ([]int, error) { resp, err := llm.Chat(ctx, model.Request{ Messages: []model.Message{ model.TextMessage(model.RoleSystem, reviewPrompt), model.TextMessage(model.RoleUser, buildReviewContent(msgs)), }, Purpose: "noise_review", MaxTokens: reviewMaxTokens, }, model.Options{}) if err != nil { return nil, fmt.Errorf("flowgate: review model: %w", err) } keep, err := parseKeepIndices(resp.Content, len(msgs)) if err != nil { log.Warn("flow gate review output unparsable, falling back to FIFO", "error", err) return nil, err } return keep, nil } } // buildReviewContent 把队列消息渲染为带编号的截断清单, // 便于审查者低成本判断重要性。这里不做身份判定或 LLM 判断 // ——只生成紧凑、带索引标签的文本摘要。 func buildReviewContent(msgs []adapter.UnifiedMessage) string { var b strings.Builder for i, m := range msgs { name := senderDisplayName(m, m.Sender) if name == "" { name = "(未标注)" } fmt.Fprintf(&b, "[%d] %s: %s\n", i, name, truncateText(m.Content, reviewSnippetLen)) } return b.String() } // truncateText 为展示把 s 截短到至多 n 字节(按 rune 计数)。 func truncateText(s string, n int) string { if n <= 0 { return "" } runes := []rune(s) if len(runes) <= n { return s } return string(runes[:n]) + "…" } // 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) { return adapter.ParseReviewIndices(s, bound) }