AskaEth
ede00d9928
- 迁移 0020_chat_message_voice：按 design/24 §10 与契约 §7.2 已登记的 voice 形状建列/索引
- 语音回合进编排并落库；voice 引用与统计如实回填（转码/转写没成功就如实标未成功，不假装）
- 复用既有 internal/capability/transcode：非 wav（微信 AMR 等）先转码再送 ASR；失败可分类且不丢消息
⚠️ 部分完成（如实登记）：生产端到端未成立（本片只到落库与转码接入），且**未经验真机**。
370 lines
19 KiB
Go
370 lines
19 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/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 是交给参与度判定器的当前角色身份/系统提示词块，
|
||
// 使其能权衡角色的语气。其中不含主人敏感信息。
|
||
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
|
||
// Factory 依据某平台生效的账号配置重建其适配器
|
||
// （channels.json → 环境变量回退）。非 nil 时桥接器支持热重载；
|
||
// 为 nil 时桥接器仅支持启动期装配（Reload 返回 ErrNilFactory）。
|
||
Factory PlatformFactory
|
||
// 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
|
||
|
||
mu sync.RWMutex
|
||
adapters map[string]adapter.PlatformAdapter
|
||
connected map[string]bool
|
||
readCancels map[string]context.CancelFunc
|
||
started bool
|
||
closed bool
|
||
factory PlatformFactory
|
||
// 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{}
|
||
}
|
||
|
||
// 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),
|
||
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,
|
||
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),
|
||
}, nil
|
||
}
|