AskaEth
c34b9d195d
P0-3 受限车道（群聊/陌生人）的系统框架行并列注入【公域呈现】 （publicPresencePrompt）：它是呈现层投影（AGENTS §9 行为模式=呈现）， 不是人格切换也不是权限门——不称第三方「亲爱的」等主人专属称呼、 不做只对主人的亲近动作、主人的住址/姓名/作息/行程/健康/家人/关系 即使记忆里读到也不外说；主人私聊与 Web 主会话不注入。 P0-4 纵深防御：第三方边界提示词与 loop 受限前导片段各补两条硬规则 （不用主人专属亲昵称呼指代第三方 + 不外说主人私事）。 P1-2 判定器去人格：模型输入只吃「角色名 + 偏好摘要」，整份人格指令 不再进模型（审计 2026-10-30 §2.4 的 reason 亲昵口吻即由此而来）。 P1-3 测试补真：受限期呈现块存在性 + 主人车道不注入 + 判定器输入/ 模型请求均不含人格正文；并纠正 memory_filter 的错误注释 （作用域 ≠ 敏感度，审计 §6 R1）。
444 lines
24 KiB
Go
444 lines
24 KiB
Go
// channelbridge 包把已注册的 IM 渠道适配器接入 Cyrene 运行时：
|
||
// 它连接每个已配置的适配器、装配入站消息处理器，并让每条规范化后的入站消息
|
||
// 依次流经
|
||
// 渠道→会话→actor→循环→回复 流水线（DevDocs/design/08 §0-§2）。
|
||
//
|
||
// 桥接器刻意对 adapter.PlatformAdapter 保持泛化。它掌管
|
||
// 按平台的 Connect 生命周期与「入站→回复」链路；具体适配器
|
||
// 本身（及其来自环境变量的配置）在 cmd/cyrene/wire.go 中组装，
|
||
// 并通过 Add 注入。它绝不读取消息正文来
|
||
// 决定身份——那是 ResolveActor 的职责（actor 元数据 + 主人白名单，
|
||
// 即项目红线）。
|
||
package channelbridge
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"log/slog"
|
||
"math/rand"
|
||
"net/http"
|
||
"sync"
|
||
"time"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/transcode"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/events"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||
)
|
||
|
||
// webhookAdapter 是渠道适配器可选择性暴露的能力，用于
|
||
// 提供 HTTP 入站回调。*webhook.Adapter 与 *kook.Adapter（webhook
|
||
// 模式）实现了它。桥接器上报这些能力，好让 gateway 挂载它们。
|
||
type webhookAdapter interface {
|
||
// InboundHandler 返回 gateway 挂载在路由上、用于
|
||
// 接收外部回调的 http.Handler。
|
||
InboundHandler() http.Handler
|
||
// WebhookPath 返回 gateway 应挂载该处理器的路由。适配器当前
|
||
// 未暴露 webhook 时返回 ""，让
|
||
// 桥接器跳过它（例如 websocket 模式的 KOOK）。
|
||
WebhookPath() string
|
||
}
|
||
|
||
// 调节参数常量。编译期常量，运行路径上就不会出现裸魔法数字
|
||
// （项目规范 §10）。
|
||
const (
|
||
// connectTimeout 限定单次适配器 Connect 的时长，卡住的拨号不会拖住
|
||
// 启动（连接失败只记警告并跳过，绝不致命）。
|
||
connectTimeout = 10 * time.Second
|
||
// maxConcurrent 限定同时运行 agent 循环的入站消息数，
|
||
// 渠道消息突发时不会无限制地派生 goroutine。
|
||
maxConcurrent = 16
|
||
)
|
||
|
||
// ErrNilLoop 表示构建桥接器时未提供对话运行器。
|
||
var ErrNilLoop = errors.New("channelbridge: nil loop runner")
|
||
|
||
// ErrNilSessions 表示构建桥接器时未提供会话存储。
|
||
var ErrNilSessions = errors.New("channelbridge: nil session store")
|
||
|
||
// Chatter 执行单轮 agent。生产实现是 *loop.Loop；
|
||
// 测试注入桩实现。
|
||
type Chatter interface {
|
||
Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error)
|
||
}
|
||
|
||
// SessionStore 解析一条入站消息的「渠道↔会话」绑定。
|
||
// 生产实现是 *session.SessionManager。
|
||
type SessionStore interface {
|
||
GetOrCreate(ctx context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (session.Session, error)
|
||
}
|
||
|
||
// Options 承载桥接器注入的依赖。Loop 与 Sessions 为
|
||
// 必填；Owner 可为 nil（此时不识别任何主人账号，即 design/11 §0 的
|
||
// 安全默认拒绝姿态）。History 与 OnPersisted 可选
|
||
// （design/16 §4.4）：接入历史存储后，主人私聊消息会
|
||
// 被路由进主 Web 会话并打上渠道平台标签，且每条
|
||
// 已持久化消息都会交给 OnPersisted，好让 gateway 广播
|
||
// `message_persisted`（Web 聊天）或 `channel_persisted`（IM 适配器日志）事件；
|
||
// 两者都缺省时，桥接器行为与以往完全一致（隔离的 sess-xxx，
|
||
// 不写 chat_messages）。
|
||
type Options struct {
|
||
Log *slog.Logger
|
||
Loop Chatter
|
||
Sessions SessionStore
|
||
Owner *user.OwnerProfile
|
||
History history.HistoryStore
|
||
OnPersisted func(history.Message)
|
||
// OnRejected 非 nil 时，会在消息被
|
||
// 黑白名单（allow/deny，design/16 §4.6 §6）丢弃后调用，便于组合根
|
||
// 将其写入审计轨迹。调用时传入被拒消息与
|
||
// 简短原因标记（"group_denied"、"user_not_allowed" 等）。Nil = 仅记日志。
|
||
//
|
||
// 它同时被装配为 flow gate 的 OnPruned 审计钩子（design/12 §6 策略三/四），
|
||
// 因此被限流器或审查裁剪丢弃的消息也会进入同一
|
||
// 审计轨迹，原因为 "rate_limited"、"review_pruned" 或 "capped"。
|
||
OnRejected func(adapter.UnifiedMessage, string)
|
||
// Review 是可选的审查模型（design/12 §6 策略四），用于裁剪低价值
|
||
// 队列消息。Nil 时在总量上限处回退为 FIFO 淘汰。用
|
||
// NewLLMReviewer 把 model.Adapter 包装成该签名。
|
||
Review adapter.Reviewer
|
||
// Light 是可选的轻量应答判定器，用于非主人的群聊/陌生人消息
|
||
// （design/12 §6 策略二/五 + design/18 §3）。为 nil 时桥接器使用
|
||
// 内置规则判定器并静默失败关闭，不会仅为「是否打扰主人」而把
|
||
// 聊天消息发给 LLM。它在规则判定器得不出结论、且参与度引擎
|
||
// （design/18）已被触发（角色名提及 / 概率）或处于 ACTIVE 窗口时
|
||
// 被调用。用 NewLLMLightJudge 把 model.Adapter 包成
|
||
// 受成本控制的判定器。
|
||
Light LightJudge
|
||
// TriggerWords 是当前角色的角色名提醒词（design/18 §2）。
|
||
// 群消息只要含有其中任一个，即使没有直接 @ 也会进入参与度 PROBING。
|
||
// 它们绝不代表身份（身份仍由结构决定）。为空则保持
|
||
// 桥接器沿用旧的 design/12 §6 轻量判定。
|
||
TriggerWords []string
|
||
// PersonaPrompt 是当前角色的身份/系统提示词块（装配方传入）。
|
||
// 🔴 LLM 判定器**不**把整份人格指令喂进模型：它只吃由它折算出的
|
||
// 「角色名 + 偏好摘要」（engagejudge.go 的 judgePersonaBrief），
|
||
// 否则判定 reason 会带亲昵口吻去指代第三方（审计 2026-10-30 §2.4 / P1-2）。
|
||
// 字段仍留在输入结构里（装配方与既有调用方都在设它），但内置判定器
|
||
// 刻意不使用它。
|
||
PersonaPrompt string
|
||
// Engage 是参与度引擎配置（概率、活跃窗口、冷却、
|
||
// 上下文大小）。当其 Enabled/Probability 为零且 TriggerWords 为空时，
|
||
// 桥接器沿用旧的轻量判定，永不咨询该引擎。用
|
||
// DefaultEngageConfig 取 design/18 的默认值。
|
||
Engage EngageConfig
|
||
// EngageFor 是可选的按会话覆盖（design/18 §6 的按群参数）。
|
||
// 为 nil 时每个会话都用基准 Engage 配置（P2a）。
|
||
EngageFor func(platform, groupID string) *EngageConfig
|
||
// EngageRand 是概率触发可注入的 [0,1) 随机源。为 nil 时使用
|
||
// 默认的 math/rand 源；测试注入固定源以保证确定性。
|
||
EngageRand func() float64
|
||
// QuietStartHour 与 QuietEndHour 限定免打扰窗口（本地时钟
|
||
// 小时），与 backthink 服务共享，使策略五沿用同一定义。
|
||
// 两个非零且相等的时刻关闭该窗口；都为零则取 backthink 默认值
|
||
// （23-07）。这是跨服务保持单一事实来源的代价。
|
||
QuietStartHour int
|
||
QuietEndHour int
|
||
// Now 是安静时段判定所用的墙上时钟源。为 nil 时用 time.Now，
|
||
// 使依赖时间的行为在测试中可注入。
|
||
Now func() time.Time
|
||
// LaneDepth 是**每会话**回合车道的排队深度（M22 切片 LANEQUEUE，见 lane.go）：
|
||
// 群聊/陌生人车道满时**在门前拒收这一条**（WARN + 计数 + 审计，队列内容不动），
|
||
// 主人私聊永不拒收（溢到 spill 只排队）。零或负数取
|
||
// defaultLaneDepth（那里写了取值依据）。
|
||
LaneDepth int
|
||
// LaneStaleAfter 是**陈旧度闸**的阈值：轮到某条消息时它已等了超过这个时长，
|
||
// 就丢弃不补答（WARN + 计数 + 审计，绝不静默）。零或负数取
|
||
// defaultLaneStaleAfter（那里写了取值依据）。主人私聊与非聊天事件不受该闸约束。
|
||
LaneStaleAfter time.Duration
|
||
// Coalesce 是**静默窗口合并**（M23）的注入面：按车道键取生效窗口参数 +
|
||
// 可注入的定时器工厂。零值 = 走代码默认（群聊窗口 2.0s / 绝对上限 6s / 单批 24 条），
|
||
// 也就是「装配方忘了接线也有合理行为」，而不是把特性静默关掉。
|
||
//
|
||
// 🔴 它**不能**改变车道类型：主人私聊永远是 direct 车道（无窗口、无定时器），
|
||
// 由 actor 元数据决定，配置无权翻转（见 lane.go 的 newDirectLane/newBatchLane）。
|
||
Coalesce CoalesceOptions
|
||
// OwnerLaneFor 是 owner 车道结构化旋钮（design/18 §11「乙 = C」）的**读取**接缝：
|
||
// 按 (platform, groupID) 返回该群生效的参与策略，返回 nil 表示没有覆盖
|
||
// （桥用自己的默认档：不对她说就不开口）。为 nil 时全部走默认档 + 桥内内存覆盖。
|
||
OwnerLaneFor func(platform, groupID string) *OwnerLanePolicy
|
||
// OwnerLaneSet 是旋钮的**落盘**接缝：主人在群里说「少说点」这类诉求被结构化后，
|
||
// 桥通过它把新策略写进外部配置（面板可读可改的那一份）。为 nil 时只更新桥内内存
|
||
// 覆盖（进程重启即回到默认档）。
|
||
OwnerLaneSet func(platform, groupID string, p OwnerLanePolicy) error
|
||
// Factory 依据某平台生效的账号配置重建其适配器
|
||
// （channels.json → 环境变量回退）。非 nil 时桥接器支持热重载；
|
||
// 为 nil 时桥接器仅支持启动期装配（Reload 返回 ErrNilFactory）。
|
||
Factory PlatformFactory
|
||
// Channels 是能力注册表的窄端口（见 ChannelRegistry）：热重载把新适配器写回
|
||
// 同名条目，使管理面板读到的永远是当前实例。为 nil 时热重载照常，只是面板
|
||
// 仍读旧实例（首次同步记一次 WARN）。
|
||
Channels ChannelRegistry
|
||
// ChannelPersona 非 nil 时解析某个渠道账号绑定的角色
|
||
// （channels.json 的 ChannelAccount.PersonaID，design/22 §2 方案 b）：调用时传入
|
||
// 平台与本地账号 id，账号未绑定角色时返回 ""。
|
||
// 它只在每次适配器连接/重载时咨询一次（绝不按消息调用），
|
||
// 且仅针对主人的私聊，因此群聊/陌生人会话绝不会
|
||
// 被移入某个角色的作用域（design/22 §7-4）。为 nil 时所有渠道都使用全局
|
||
// 默认角色。
|
||
ChannelPersona func(platform, accountID string) string
|
||
// VoiceSTT 是**入站语音转写**的注入点（切片 IM-ASR；design/06 §1.1）。组合根注入
|
||
// `*voice.Chain`（与 Web 侧 WS 的 voice_input 同一条降级链）。为 nil 时，带语音的
|
||
// 消息走「服务端未配置云端语音识别」的**诚实提示**路径 —— 绝不静默丢消息。
|
||
VoiceSTT VoiceSTT
|
||
// VoiceSTTTimeout 是一次 IM 入站语音转写的超时上限（≤0 时取默认 20s）。入站流水线
|
||
// 会为这次调用举着一个并发槽，因此上游卡住**不能**拖死整条流水线。
|
||
VoiceSTTTimeout time.Duration
|
||
// VoiceTranscoder 是**入站语音副本的可播归一化**所用的转码能力（design/25 §7.1 第 ④ 步；
|
||
// 组合根注入 `a.transcoder`，与工具面用的是同一个实例）。为 nil 时：
|
||
// 可播格式的语音照旧落副本（不需要转码），**不可播格式**（QQ 的 AMR）走
|
||
// 「不落副本 + 带分类的 WARN」的诚实路径 —— 绝不落一份播不了的字节并假装能播。
|
||
VoiceTranscoder transcode.Transcoder
|
||
// VoiceAudioStore 是入站语音副本的**落盘面**（design/24 §10.9-2）。nil 时按消息构造
|
||
// 共享文档存储并在它实现了音频分支时直接用它（见 VoiceAudioStore 的说明）。
|
||
VoiceAudioStore VoiceAudioStore
|
||
// DocumentStore 是 IM **入站附件**的落盘配置（design/24 §3；切片 M1）：
|
||
// 目录、单文件上限与 `documents` 元数据写入面。它由组合根给出（桥接器不猜
|
||
// 目录布局），并与 Web 上传共用同一个存储核心。UploadsDir 与 DataDir 都为
|
||
// 空即视为**未接线**：带文件的消息照旧能跑（模型会收到一行「读不到」的诚实
|
||
// 提示），只是不落库、不抽取 —— 绝不假装读到了。
|
||
DocumentStore documents.StoreConfig
|
||
// EventStore 是**渠道事件留存**面（design/26 §4/§5；切片 E3）：所有非消息事件
|
||
// （含未知类型与解析失败）经 channelbridge/events.go 的 recordEvent **无条件**
|
||
// 写进它 —— T2/T3 与 unknown 逐条落行、T1 只累加计数。
|
||
//
|
||
// 它只依赖 `events.Store` 接口，因此为 nil 时桥接器走**降级路径**（内存计数 +
|
||
// 首次一条 DEBUG），入站流水线**绝不**因为没有存储而失败或丢帧（fail-visible 而
|
||
// 非 fail-crash，§4.3）。注入必须传**真 nil 接口**而不是装了 nil 指针的接口，
|
||
// 否则这条判据会失效（见 cmd/cyrene/wire_channel_events.go 的 channelEventStore）。
|
||
EventStore events.Store
|
||
}
|
||
|
||
// VoiceAudioStore 是入站语音**可播副本**的落盘面（design/24 §10.9-2 的「共享存储音频
|
||
// 分支」）。它与 `documents.Store.Save` 同形：内容寻址 + 去重 + 元数据行，但走另一份
|
||
// 裁决（音频白名单，见 design/24 §10.6）。
|
||
//
|
||
// 生产装配**可以留 nil**：那时桥接器按消息构造共享文档存储，并在它实现了本接口时直接
|
||
// 用它（`documents.Store.SaveAudio`）—— 也就是说，只要存储侧补上那个方法，无需改组合根。
|
||
// 注入面优先于上面那条发现路径，供测试与「用别的落盘实现」的部署方使用。
|
||
//
|
||
// 两者都没有 ⇒ 走「副本没落成」的诚实路径（带来源 WARN + 不打假 url），绝不假装落成了。
|
||
type VoiceAudioStore interface {
|
||
SaveAudio(ctx context.Context, origName string, data []byte) (documents.Ref, error)
|
||
}
|
||
|
||
// Bridge 掌管渠道适配器生命周期与入站消息流水线。它
|
||
// 可安全并发使用。
|
||
type Bridge struct {
|
||
log *slog.Logger
|
||
run Chatter
|
||
sessions SessionStore
|
||
owner *user.OwnerProfile
|
||
history history.HistoryStore
|
||
onPersisted func(history.Message)
|
||
// onProactivePersisted 是**主动推送**那条实时通知接缝（切片 P0b）：一条主动消息刚
|
||
// 落进主会话时被调用，使客户端能凭结构化判据（帧顶 `proactive:true`）走与
|
||
// proactive_message **同一套**提醒逻辑 —— 「她该不该提醒我」因此与后端走哪条路由
|
||
// 无关。普通消息的通知仍走 onPersisted，两条互不串味（见 proactive_notify.go）。
|
||
onProactivePersisted func(history.Message)
|
||
onRejected func(adapter.UnifiedMessage, string)
|
||
review adapter.Reviewer
|
||
|
||
light LightJudge
|
||
quietStartHour int
|
||
quietEndHour int
|
||
now func() time.Time
|
||
// noticeLog 是系统通知传入前审查（design/12 §6 策略一补充）的子功能日志器：
|
||
// `source=services/channelbridge/notice` 便于跨模块定位。三个通知类日志
|
||
// （notice / flow / event）都在构造期一次性标好来源，使「这条为什么没进她上下文」
|
||
// 「这一帧有没有留下可查痕迹」在日志里可追溯。
|
||
noticeLog *slog.Logger
|
||
// flowLog 是流控解析（design/12 §6 策略三，含账号级 directMsgPerMin）的子功能
|
||
// 日志器：`source=services/channelbridge/flowgate`，使「主人显式关掉某项流控」
|
||
// 在日志里可追溯（见 flowgate.go 的 directLimit）。
|
||
flowLog *slog.Logger
|
||
// eventLog 是渠道事件留存（design/26 §4；切片 E3）的子功能日志器：
|
||
// `source=services/channelbridge/events`。留存成功走 DEBUG（含 NoticeClass 的
|
||
// 裁决结果），未接线 / 落库失败走 DEBUG（首帧）+ WARN（失败），全部带来源。
|
||
eventLog *slog.Logger
|
||
|
||
// 参与度引擎状态（design/18）：按会话的内存状态机，
|
||
// 加上约束它的角色提醒词 / 参与度参数，以及概率触发
|
||
// 可注入的随机源。
|
||
triggerWords []string
|
||
personaPrompt string
|
||
engage EngageConfig
|
||
engageFor func(platform, groupID string) *EngageConfig
|
||
engageRand func() float64
|
||
// bubbleGapFn 是 IM 出站气泡之间可注入的延时（0.5-2s），
|
||
// 默认取包内的 bubbleGap。测试注入固定的短间隔，从而
|
||
// 无需真的 sleep 0.5-2s 就能确定性地断言多气泡延时。
|
||
bubbleGapFn func() time.Duration
|
||
engageMu sync.Mutex
|
||
engageGroups map[string]*engageGroupState
|
||
|
||
// ownerLane 是 owner 车道结构化旋钮（design/18 §11）的**内存覆盖表**，
|
||
// 键 = platform|groupID。ownerLaneFor 从外部配置读取（组合根注入的 owner_lane.json），
|
||
// ownerLaneSet 把改动回写落盘；ownerLaneLast 记录每条 owner 车道回复的时间戳，供
|
||
// 「多频繁」闸使用。四者都由 ownerLaneMu 保护。
|
||
ownerLane map[string]OwnerLanePolicy
|
||
ownerLaneFor func(platform, groupID string) *OwnerLanePolicy
|
||
ownerLaneSet func(platform, groupID string, p OwnerLanePolicy) error
|
||
ownerLaneLast map[string]time.Time
|
||
ownerLaneMu sync.Mutex
|
||
// ownerLaneLog 是 owner 闸门/旋钮的子功能日志器（source=services/channelbridge/ownerlane），
|
||
// 使「这条为什么没回」与「旋钮被改成什么」在日志里可追溯。
|
||
ownerLaneLog *slog.Logger
|
||
|
||
mu sync.RWMutex
|
||
adapters map[string]adapter.PlatformAdapter
|
||
connected map[string]bool
|
||
readCancels map[string]context.CancelFunc
|
||
started bool
|
||
closed bool
|
||
factory PlatformFactory
|
||
// channels 是热重载同步能力注册表的窄端口（reload.go 的 replaceChannelEntry /
|
||
// dropChannelEntry）；为 nil 时面板读到的是保存前的旧实例，首次同步记一次 WARN。
|
||
channels ChannelRegistry
|
||
channelsWarnOnce sync.Once
|
||
// channelPersona 从生效配置解析账号的渠道级角色
|
||
// （design/22 §2 方案 b）；adapterPersona 按已连接的适配器缓存解析结果，
|
||
// 使入站路径永不重读 channels.json。二者
|
||
// 都由 mu 保护。
|
||
channelPersona func(platform, accountID string) string
|
||
adapterPersona map[string]string
|
||
// documentsCfg 是入站附件落盘配置（design/24 §3；切片 M1）。零值 = 未接线：
|
||
// 带文件的消息照旧能跑，模型会收到一行「读不到」的诚实提示（绝不静默）。
|
||
documentsCfg documents.StoreConfig
|
||
// eventStore 是渠道事件留存面（design/26 §4；切片 E3）。**真 nil 接口** = 未接线：
|
||
// 非消息事件走内存计数降级，入站照常（见 events.go 的 recordEvent）。
|
||
// eventFallback 是那条降级路径的计数面（有界，fail-visible）。
|
||
eventStore events.Store
|
||
eventFallback *eventFallback
|
||
// voiceSTT / voiceSTTTimeout / voiceCircuit 是入站语音转写面（切片 IM-ASR；
|
||
// design/06 §1.1）：注入点、独立短超时，以及按原因分类的**临时熔断**。
|
||
// voiceLog 是带 `services/channelbridge/voice` 来源的子功能日志器。
|
||
// voiceTranscoder 是入站语音**副本**的可播归一化所用的转码能力（切片 CODE-VOICE；
|
||
// design/25 §7.1 第 ④ 步）—— 与转写面**彼此独立**：转写只认原始字节，转码只服务回放。
|
||
voiceSTT VoiceSTT
|
||
voiceSTTTimeout time.Duration
|
||
voiceCircuit voiceBreaker
|
||
voiceLog *slog.Logger
|
||
voiceTranscoder transcode.Transcoder
|
||
// voiceAudioStore 是**显式注入**的音频落盘面（切片 CODE-VOICE；design/24 §10.9-2）。
|
||
// nil = 退回「共享文档存储的音频分支」发现路径（见 inboundVoiceStore）。
|
||
voiceAudioStore VoiceAudioStore
|
||
// proactiveCursor 是主动推送「无信号」兜底时的轮转游标
|
||
// （channelbridge/proactive.go），使默认渠道不会总是
|
||
// 白名单中的第一个主人渠道。由 mu 保护。
|
||
proactiveCursor int
|
||
|
||
// gateMu 保护按适配器的流控闸门（design/12 §6 策略三/四）。它们
|
||
// 按账号在首条入站消息时惰性构建，且独立于
|
||
// 适配器/connected 生命周期，因此同一个闸门能跨重连存活。
|
||
gateMu sync.Mutex
|
||
gates map[string]*adapter.FlowGate
|
||
|
||
ctx context.Context
|
||
cancel context.CancelFunc
|
||
wg sync.WaitGroup
|
||
|
||
sem chan struct{}
|
||
// lanes 是**每会话回合车道**的注册表（M22 切片 LANEQUEUE，见 lane.go）：
|
||
// 读循环只投递不跑回合；同一会话内串行 FIFO、不同会话并行；超龄与背压丢弃都留痕。
|
||
// 只在 enqueueInbound 里与 mu 同时出现，锁顺序恒为 mu → lanes.mu。
|
||
lanes *laneSet
|
||
// coalesceFor 是静默窗口参数（M23）的读取接缝（面板配置文件 → 生效值）；
|
||
// coalesceLog 是窗口引擎的子功能日志器（source=services/channelbridge/coalesce）；
|
||
// coalesceStats 是窗口/发车/续跑的计数面（CoalesceStats() 的只读出口）。
|
||
// 三者的类型与语义见 coalesce.go；它们只被 batch 车道的头部用到。
|
||
coalesceFor func(laneKey string) *runtimeconfig.CoalescePolicy
|
||
coalesceLog *slog.Logger
|
||
coalesceStats *coalesceCounters
|
||
}
|
||
|
||
// New 依据 opts 构建 Bridge。当必需依赖缺失时返回错误，
|
||
// 使调用方快速失败，而不是静默丢消息。
|
||
func New(opts Options) (*Bridge, error) {
|
||
if opts.Loop == nil {
|
||
return nil, ErrNilLoop
|
||
}
|
||
if opts.Sessions == nil {
|
||
return nil, ErrNilSessions
|
||
}
|
||
log := opts.Log
|
||
if log == nil {
|
||
log = logger.New(logger.Options{Out: nil})
|
||
}
|
||
// 安静时段默认取 backthink 服务使用的同一窗口，这样两个
|
||
// 服务之间永远不会对免打扰窗口产生分歧（design/12 §6 策略五）。
|
||
qStart, qEnd := opts.QuietStartHour, opts.QuietEndHour
|
||
if qStart == 0 && qEnd == 0 {
|
||
qStart, qEnd = defaultQuietStartHour, defaultQuietEndHour
|
||
}
|
||
now := opts.Now
|
||
if now == nil {
|
||
now = time.Now
|
||
}
|
||
engageRand := opts.EngageRand
|
||
if engageRand == nil {
|
||
engageRand = func() float64 { return rand.Float64() }
|
||
}
|
||
return &Bridge{
|
||
log: logger.WithSource(log, "services/channelbridge"),
|
||
run: opts.Loop,
|
||
sessions: opts.Sessions,
|
||
owner: opts.Owner,
|
||
history: opts.History,
|
||
onPersisted: opts.OnPersisted,
|
||
onRejected: opts.OnRejected,
|
||
review: opts.Review,
|
||
light: opts.Light,
|
||
quietStartHour: qStart,
|
||
quietEndHour: qEnd,
|
||
now: now,
|
||
noticeLog: logger.WithSource(log, "services/channelbridge/notice"),
|
||
flowLog: logger.WithSource(log, flowgateLogSource),
|
||
eventLog: logger.WithSource(log, "services/channelbridge/events"),
|
||
triggerWords: opts.TriggerWords,
|
||
personaPrompt: opts.PersonaPrompt,
|
||
engage: opts.Engage,
|
||
engageFor: opts.EngageFor,
|
||
engageRand: engageRand,
|
||
bubbleGapFn: bubbleGap,
|
||
engageGroups: make(map[string]*engageGroupState),
|
||
ownerLane: make(map[string]OwnerLanePolicy),
|
||
ownerLaneFor: opts.OwnerLaneFor,
|
||
ownerLaneSet: opts.OwnerLaneSet,
|
||
ownerLaneLast: make(map[string]time.Time),
|
||
ownerLaneLog: logger.WithSource(log, ownerLaneLogSource),
|
||
adapters: make(map[string]adapter.PlatformAdapter),
|
||
connected: make(map[string]bool),
|
||
readCancels: make(map[string]context.CancelFunc),
|
||
gates: make(map[string]*adapter.FlowGate),
|
||
factory: opts.Factory,
|
||
channels: opts.Channels,
|
||
channelPersona: opts.ChannelPersona,
|
||
adapterPersona: make(map[string]string),
|
||
documentsCfg: opts.DocumentStore,
|
||
eventStore: opts.EventStore,
|
||
eventFallback: newEventFallback(),
|
||
voiceSTT: opts.VoiceSTT,
|
||
voiceSTTTimeout: voiceSTTTimeout(opts.VoiceSTTTimeout),
|
||
voiceLog: logger.WithSource(log, "services/channelbridge/voice"),
|
||
voiceTranscoder: opts.VoiceTranscoder,
|
||
voiceAudioStore: opts.VoiceAudioStore,
|
||
cancel: func() {},
|
||
sem: make(chan struct{}, maxConcurrent),
|
||
// 每会话车道：深度与陈旧度阈值可经 Options 覆盖（零值取 lane.go 里的默认值）。
|
||
// 定时器工厂来自 Options.Coalesce.After（M23 测试缝）；direct 车道拿不到它。
|
||
lanes: newLaneSet(opts.LaneDepth, opts.LaneStaleAfter, opts.Coalesce.After, log),
|
||
// 静默窗口合并（M23）：参数读取接缝 + 独立日志来源 + 计数面。
|
||
coalesceFor: opts.Coalesce.For,
|
||
coalesceLog: logger.WithSource(log, coalesceLogSource),
|
||
coalesceStats: &coalesceCounters{},
|
||
}, nil
|
||
}
|