feat(channels,voice): IM 各渠道语音消息可被读到(全渠道云端 ASR + 临时熔断)
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

采集面:mediafetch 新增 inline_voice.go(音频白名单,与 documents.ExtAllowed 互不污染;
复用既有来源白名单 / 逐跳校验 / 边读边限 / 超时);四个适配器零改动。
消费面:channelbridge 在流水线最前端把「语音字节 → 文本」归一并写回 msg.Content;
inbound.go 的丢空判据放宽为「无文本且无可用媒体」。
微信优先平台自带转写(Content 非空 ⇒ 不调 ASR,不重复付费)。
降级(主人定的两层):新增 voice/outcome.go 的 STTOutcome 识破静默档位;
errors.go 分类鉴权失败/欠费/限流(CircuitWorthy);voice_breaker.go 临时熔断 + 退避(TTL 恢复);
无论是否熔断,消息面都落一行中文诚实提示(绝不静默、绝不留空消息)。
独立短超时:VOICE_STT_TIMEOUT_SEC(默认 20s,同步 .env.example)。
kook 不再把 CDN 地址写进 Content(地址仍在 Media[].URL)。
设计先行:design/06 §1.1/§1.2、design/08 §4.1、design/24 §1.3/§4.2、开源对照清单。
变异 5 条(静默档位不带 Cause / Transcribed 误判 / 熔断恒放行 / ChatText 退化 / 平台文字判据取反)。
诚实边界:实施时官方音频规格尚未抓到 ⇒ MaxVoiceBytes 自定 5 MiB(保守)、采样率仍 16 kHz;
官方规格随后已抓到(10 MB / 5 min;格式含 amr、无 silk),对齐留待下次改动。
This commit is contained in:
AskaEth committed 2026-10-05 08:32:37 +08:00
1 parent 679cee04f7
commit a918819878
20 files changed
+1685 -61

No files matched your search

@@ -149,6 +149,13 @@ type Options struct {
// 被移入某个角色的作用域(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
// DocumentStore 是 IM **入站附件**的落盘配置(design/24 §3;切片 M1):
// 目录、单文件上限与 `documents` 元数据写入面。它由组合根给出(桥接器不猜
// 目录布局),并与 Web 上传共用同一个存储核心。UploadsDir 与 DataDir 都为
@@ -205,6 +212,13 @@ type Bridge struct {
// documentsCfg 是入站附件落盘配置(design/24 §3;切片 M1)。零值 = 未接线:
// 带文件的消息照旧能跑,模型会收到一行「读不到」的诚实提示(绝不静默)。
documentsCfg documents.StoreConfig
// voiceSTT / voiceSTTTimeout / voiceCircuit 是入站语音转写面(切片 IM-ASR;
// design/06 §1.1):注入点、独立短超时,以及按原因分类的**临时熔断**。
// voiceLog 是带 `services/channelbridge/voice` 来源的子功能日志器。
voiceSTT VoiceSTT
voiceSTTTimeout time.Duration
voiceCircuit voiceBreaker
voiceLog *slog.Logger
// proactiveCursor 是主动推送「无信号」兜底时的轮转游标
// (channelbridge/proactive.go),使默认渠道不会总是
// 白名单中的第一个主人渠道。由 mu 保护。
@@ -251,34 +265,37 @@ func New(opts Options) (*Bridge, error) {
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,
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,
cancel: func() {},
sem: make(chan struct{}, maxConcurrent),
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,
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,
voiceSTT: opts.VoiceSTT,
voiceSTTTimeout: voiceSTTTimeout(opts.VoiceSTTTimeout),
voiceLog: logger.WithSource(log, "services/channelbridge/voice"),
cancel: func() {},
sem: make(chan struct{}, maxConcurrent),
}, nil
}
+10 -4
View File
@@ -133,11 +133,17 @@ func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
// 桥接器绝不读取消息内容来判定发送者是谁
// (项目红线)。
func (b *Bridge) processFrom(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) {
// 语音 → 文本归一是入站流水线的**第一件事**(切片 IM-ASR;design/06 §1.1):
// 它必须在「丢空判据」之前发生,否则 telegram 的纯语音(无 caption ⇒ Content
// 为空)会在下面被整条丢掉。归一写回 msg.Content 之后,最小打扰判定、落库与
// 模型可见消息看到的都是**同一份**文本(转写结果,或一行诚实的失败提示)。
b.normalizeInboundVoice(ctx, &msg)
text := strings.TrimSpace(msg.Content)
// 无文本且无文件类附件 ⇒ 没有可处理的内容(与既有行为一致)。带附件的
// 「纯文件消息」必须继续往下走:附件就是它的内容(design/24 §4.4 的
// 「不许静默丢」)。
if text == "" && !hasFileMedia(msg) {
// 无文本且无**任何可用媒体** ⇒ 没有可处理的内容。带附件的「纯文件消息」与带语音的
// 「纯语音消息」必须继续往下走:附件/语音就是它的内容(design/24 §4.4 的
// 「不许静默丢」;语音那一路的归一已保证它要么有文本、要么有诚实提示)。
if text == "" && !hasUsableMedia(msg) {
b.log.Debug("inbound message has no chat text, skipped",
"platform", msg.Platform, "channel", msg.ChannelID)
return
@@ -14,8 +14,9 @@ package channelbridge
// 带来源的记录(只记 name/mime/size,**绝不记正文**)。
//
// 四条边界:
// - **图片/语音不走这里**(design/24 §4.2 末:图片走既有视觉链路),本切片只补
// 文件类;
// - **图片不走这里**(design/24 §4.2 末:图片走既有视觉链路);**语音也不走这里** ——
// 入站语音由 `channelbridge/inbound_voice.go` 在流水线最前端归一成文本写回
// `Content`(切片 IM-ASR;design/06 §1.1),既不落文档存储也不进文档预算;
// - 入站附件是**用户内容** ⇒ 文档部件只挂在 `model.RoleUser` 消息上;
// - 摄取发生在**闸门之后**(会话解析、黑白名单、最小打扰判定都通过之后),
// 被拒的消息因此绝不会在盘上留下文件;
@@ -30,6 +31,7 @@ import (
"unicode/utf8"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/mediafetch"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
@@ -126,6 +128,7 @@ func (b *Bridge) inboundDocumentColumn(msg adapter.UnifiedMessage, attach inboun
}
// hasFileMedia 报告消息是否携带文件类媒体(决定「无文本消息」是否仍要跑流水线)。
// 语音那一半见 inbound_voice.go 的 hasVoiceMedia / hasUsableMedia。
func hasFileMedia(msg adapter.UnifiedMessage) bool {
for _, m := range msg.Media {
if isFileMedia(msg, m) {
@@ -138,11 +141,12 @@ func hasFileMedia(msg adapter.UnifiedMessage) bool {
// isFileMedia 报告一条媒体是不是**文件类**。部分适配器只给内容类型而不给媒体
// Kind(`ContentType == file`,附件就在 Media 里),那种消息也按文件处理,否则
// 附件会被静默忽略 —— 静默忽略正是本切片要消灭的行为。
//
// 判定本身只有一份、在 `mediafetch.IsFileMedia`:桥接器的入站摄取与适配器的字节内联
// 必须对「哪些附件该被处理」给出同一个答案,否则会出现「内联了但没人消费」(切片
// IM-ASR 顺手去重,两处不再是两份实现)。
func isFileMedia(msg adapter.UnifiedMessage, m adapter.Media) bool {
if m.Kind == adapter.MediaFile {
return true
}
return m.Kind == "" && msg.ContentType == adapter.ContentFile
return mediafetch.IsFileMedia(&msg, m)
}
// fileMedia 返回消息里全部文件类媒体(保序)。
@@ -0,0 +1,239 @@
package channelbridge
import (
"context"
"errors"
"fmt"
"log/slog"
"strings"
"time"
"unicode/utf8"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/mediafetch"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice"
)
// 本文件承载**入站语音 → 文本归一**(切片 IM-ASR;design/06 §1.1、design/08 §4.1):
// 在入站流水线的**最前端**(丢空判据之前)把一条语音消息变成普通聊天文本,**只**改写
// `msg.Content`。这样「丢空判据、最小打扰判定、落库、模型可见消息」看到的都是同一份
// 文本(否则纯语音会在丢空判据处被整条丢掉,或落库成空行)。
//
// 三条口径(缺一不可):
//
// 1. **平台自带转写优先**:微信 iLink 的 `voice_item.text` 已经在 `Content` 里
// (真机已验证)⇒ **不调 ASR**(不重复付费)。判据用 `mediafetch.ChatText`,
// 因此 QQ 的 `[语音]` 占位标记不会被误当成「已有正文」。
// 2. **静默降级必须被识破**:语音链末端是**总是成功**的静默档位(空文本 + nil 错),
// 只看文本会把「没有可用的识别后端」当成一次成功的空转写 ⇒ 只认
// `voice.STTOutcome.Transcribed()`。
// 3. **绝不静默、绝不留空消息**:任何一条走完本函数的语音消息,`Content` 要么是转写
// 文本、要么是一行**中文诚实提示**(讲清没转写成功 + 原因方向)。熔断时也照落提示。
// VoiceSTT 是入站语音转写的注入点(组合根注入 `*voice.Chain`)。
//
// 它刻意要 `STTOutcome` 而不是 `STTWithProvider`:后者在静默档位处返回「空文本 + nil
// 错」,「真转写 / 静默降级」无法区分,而**按原因分类的临时熔断**(未配置 / 鉴权失败 /
// 欠费 / 限流)必须能看见被静默掩盖的真实原因(切片 IM-ASR 的硬要求)。
type VoiceSTT interface {
STTOutcome(ctx context.Context, audio voice.Audio) voice.STTOutcome
}
// defaultVoiceSTTTimeout 是一次 IM 入站语音转写的默认超时。它刻意很短:入站流水线的
// 并发槽(`maxConcurrent`)要为这次调用一直举着,上游卡住不能拖死整条流水线。
const defaultVoiceSTTTimeout = 20 * time.Second
// 诚实提示文案:每一条都讲清「没转写成功」与**原因方向**,绝不假装、也绝不静默。
// 参考 `%s` = 原因短语(见下面的 voiceReason* 常量)。
const (
voiceNoticeFailed = "[语音消息:这条语音没有转写(原因方向:%s)。你可以把内容用文字发一遍。]"
voiceNoticeCircuit = "[语音消息:这条语音没有转写(云端语音识别暂时不可用,已临时熔断退避 %s)。你可以把内容用文字发一遍。]"
voiceNoticeEmpty = "[语音消息:识别完成,但这段音频里没有识别出文字(可能是静音、太短或环境噪声)。]"
voiceNoticeNoAudio = "[语音消息:服务端没有拿到音频字节(渠道只给了平台内部引用,或代取失败),这条语音没有转写。]"
voiceNoticeFormat = "[语音消息:这条语音的音频格式不支持转写(需要先转码,当前未做)。]"
)
// 原因短语(用户可见,绝不包含上游错误原文/密钥)。
const (
voiceReasonNotConfigured = "服务端未配置云端语音识别"
voiceReasonAuth = "云端语音识别鉴权失败(API key 被拒)"
voiceReasonQuota = "云端语音识别账户欠费或额度耗尽"
voiceReasonRate = "云端语音识别限流"
voiceReasonTimeout = "云端语音识别超时"
voiceReasonCanceled = "这次转写被取消(渠道断开或回合结束)"
voiceReasonSilent = "语音链路静默降级(没有可用的识别后端)"
voiceReasonFailed = "云端语音识别失败"
)
// normalizeInboundVoice 把一条入站语音消息归一成文本并写回 `msg.Content`。没有语音
// 媒体时是空操作;任何走完本函数的语音消息都会留下文本或一行诚实提示。
func (b *Bridge) normalizeInboundVoice(ctx context.Context, msg *adapter.UnifiedMessage) {
target, found := inboundVoiceTarget(*msg)
if !found {
return
}
if mediafetch.ChatText(msg.Content) != "" {
// 平台已经给了文字(微信自带转写)或这条消息本来就带正文:不调 ASR。
b.voiceLogger().Debug("inbound voice: chat text already present, asr skipped",
"platform", msg.Platform, "channel", msg.ChannelID, "content_type", msg.ContentType)
return
}
audio, ok := voiceAudioOf(target)
if !ok {
msg.Content = voiceNoticeFormat
b.voiceLogger().Warn("inbound voice not transcribed: audio format not transcribable",
"platform", msg.Platform, "channel", msg.ChannelID,
"name", target.Name, "mime", target.MIMEType, "size", target.Size)
return
}
if len(audio.Data) == 0 {
msg.Content = voiceNoticeNoAudio
b.voiceLogger().Warn("inbound voice not transcribed: no audio bytes",
"platform", msg.Platform, "channel", msg.ChannelID,
"format", audio.Format, "mime", target.MIMEType, "size", target.Size)
return
}
b.transcribeInboundVoice(ctx, msg, audio)
}
// transcribeInboundVoice 调一次云端 STT 并把结果写回 `Content`(成功=转写文本,
// 其余一律=诚实提示)。熔断打开的短路路径也在这里:**不再发起上游调用**。
func (b *Bridge) transcribeInboundVoice(ctx context.Context, msg *adapter.UnifiedMessage, audio voice.Audio) {
now := b.now()
if !b.voiceCircuit.allowed(now) {
left := b.voiceCircuit.retryAfter(now)
msg.Content = fmt.Sprintf(voiceNoticeCircuit, left.Truncate(time.Second))
b.voiceLogger().Warn("inbound voice not transcribed: circuit open",
"platform", msg.Platform, "channel", msg.ChannelID, "retry_after", left.String())
return
}
out, ctxErr := b.callVoiceSTT(ctx, audio)
if out.Transcribed() {
// 真转写(文本可能为空:音频里没人说话,那是服务正常的空结果,不是故障)。
b.voiceCircuit.succeed()
if text := strings.TrimSpace(out.Text); text != "" {
msg.Content = text
b.voiceLogger().Info("inbound voice transcribed",
"platform", msg.Platform, "channel", msg.ChannelID,
"provider", out.Provider, "chars", utf8.RuneCountInString(text))
return
}
msg.Content = voiceNoticeEmpty
b.voiceLogger().Info("inbound voice transcribed to empty text",
"platform", msg.Platform, "channel", msg.ChannelID, "provider", out.Provider)
return
}
// 没转写出来:按类别决定是否熔断(只有「再试也是白试」的类别才关掉整面),
// 无论如何都落一行诚实提示 —— 消息**绝不**留空。
if voice.CircuitWorthy(out.Cause) {
b.voiceCircuit.open(now)
}
reason, category := voiceFailureReason(out, ctxErr)
msg.Content = fmt.Sprintf(voiceNoticeFailed, reason)
b.voiceLogger().Warn("inbound voice not transcribed",
"platform", msg.Platform, "channel", msg.ChannelID,
"provider", out.Provider, "category", category, "circuit", !b.voiceCircuit.allowed(now))
}
// callVoiceSTT 调一次注入的 STT,并为这次调用套上**独立短超时**(父 ctx 取消同样立即
// 生效)。未接线时直接返回「未配置」结果:不调上游、不等超时,走同一条诚实提示路径。
//
// 它**不把上游错误原文交给调用方去展示**:错误只用于分类(`voiceFailureReason`),
// 用户可见的正文里永远只有原因方向,绝无上游回显的请求细节。
func (b *Bridge) callVoiceSTT(ctx context.Context, audio voice.Audio) (voice.STTOutcome, error) {
if b.voiceSTT == nil {
return voice.STTOutcome{Cause: voice.ErrNotConfigured}, nil
}
callCtx, cancel := context.WithTimeout(ctx, b.voiceSTTTimeout)
defer cancel()
out := b.voiceSTT.STTOutcome(callCtx, audio)
return out, callCtx.Err()
}
// voiceFailureReason 把「没转写成功」翻成一句原因方向 + 一个日志类别。判据只认哨兵链
// 与超时,**绝不匹配错误文本**;返回的原因短语里不含任何上游错误原文(可能回显请求细节)。
func voiceFailureReason(out voice.STTOutcome, ctxErr error) (string, string) {
switch {
case errors.Is(ctxErr, context.DeadlineExceeded):
return voiceReasonTimeout, "timeout"
case errors.Is(ctxErr, context.Canceled):
return voiceReasonCanceled, "canceled"
case errors.Is(out.Cause, voice.ErrAuthFailed):
return voiceReasonAuth, "auth_failed"
case errors.Is(out.Cause, voice.ErrQuotaExceeded):
return voiceReasonQuota, "quota_exceeded"
case errors.Is(out.Cause, voice.ErrRateLimited):
return voiceReasonRate, "rate_limited"
case errors.Is(out.Cause, voice.ErrNotConfigured):
return voiceReasonNotConfigured, "not_configured"
case out.Provider == voice.SilentProviderName:
return voiceReasonSilent, "silent_degraded"
default:
return voiceReasonFailed, "provider_failed"
}
}
// inboundVoiceTarget 选出这条消息要转写的那一条语音:优先**带字节**的第一条,没有
// 任何一条带字节时退回第一条(好让调用方按它的格式与引用说清「为什么没读到」)。
func inboundVoiceTarget(msg adapter.UnifiedMessage) (adapter.Media, bool) {
var first adapter.Media
found := false
for _, m := range msg.Media {
if !mediafetch.IsVoiceMedia(&msg, m) {
continue
}
if !found {
first, found = m, true
}
if len(m.Data) > 0 {
return m, true
}
}
return first, found
}
// voiceAudioOf 把一条语音媒体转成 STT 的输入。第二个返回值为 false 表示**格式不在
// 可转写白名单内**(与字节内联同一份判据 `mediafetch.AudioFormatOf`)——此时不调上游。
func voiceAudioOf(m adapter.Media) (voice.Audio, bool) {
format, ok := mediafetch.AudioFormatOf(m)
if !ok {
return voice.Audio{}, false
}
return voice.Audio{Format: format, Data: m.Data}, true
}
// hasVoiceMedia 报告消息是否携带语音类媒体(不论有没有字节)——它让「纯语音消息」
// 不再被判成「没有内容」而丢弃(与 `hasFileMedia` 同一条纪律)。
func hasVoiceMedia(msg adapter.UnifiedMessage) bool {
for _, m := range msg.Media {
if mediafetch.IsVoiceMedia(&msg, m) {
return true
}
}
return false
}
// hasUsableMedia 报告消息是否携带**任何**可用媒体(文件或语音):入站丢空判据据此
// 放行「纯媒体消息」——纯文件与纯语音都是它们自己的内容,绝不许静默丢
// (design/24 §4.4、design/06 §1.1)。
func hasUsableMedia(msg adapter.UnifiedMessage) bool {
return hasFileMedia(msg) || hasVoiceMedia(msg)
}
// voiceSTTTimeout 把配置里的超时归一成有效值:≤0 时取 defaultVoiceSTTTimeout。归一放
// 在这里而不是调用方,使「0 表示默认」这条口径只有一份(切片 IM-ASR)。
func voiceSTTTimeout(configured time.Duration) time.Duration {
if configured <= 0 {
return defaultVoiceSTTTimeout
}
return configured
}
// voiceLogger 返回带 `services/channelbridge/voice` 来源的子功能日志器(AGENTS §10:
// 子功能日志必须可追溯;未装配时退回桥接器日志,绝不 panic)。
func (b *Bridge) voiceLogger() *slog.Logger {
if b.voiceLog != nil {
return b.voiceLog
}
return b.log
}
@@ -0,0 +1,454 @@
package channelbridge
// 本文件锁定**入站语音 → 文本归一**(切片 IM-ASR;design/06 §1.1、design/08 §4.1):
//
// - 转写结果写回 `Content` ⇒ 落库与模型可见消息看到的都是**同一份文本**;
// - **平台自带转写优先**:Content 已有真实文本(微信 voice_item.text)⇒ 一次 ASR 都
// 不调;QQ 的 `[语音]` 占位标记不算文字 ⇒ 照样转写;
// - **静默降级被识破**:`silent` 档的「成功」被当失败处理;
// - **临时熔断**:可分类失败(鉴权/欠费/限流/未配置)之后不再调上游,但消息面仍有
// 一行诚实提示;TTL 到期或一次成功即恢复;
// - 拿不到字节 / 格式不支持 / 超时 / 空识别结果都**诚实提示**,绝不静默、绝不留空;
// - 纯语音消息不再被丢空判据丢掉。
import (
"context"
"fmt"
"log/slog"
"strings"
"sync"
"testing"
"time"
"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/runtime/agent"
"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/multimodal/voice"
)
// providerFailureForTest 构造一个「按内容发生」的上游失败(格式不支持一类):它只带
// `ErrProviderFailed`、不带任何可熔断的分类哨兵 —— 用它断言熔断的**分类**是准的。
func providerFailureForTest() error {
return fmt.Errorf("voice: %w: unsupported format", voice.ErrProviderFailed)
}
// fakeVoiceSTT 是可编程的假 STT:记录调用次数与每次的音频(格式+字节),可返回任意
// STTOutcome,也可阻塞到 ctx 结束(超时用例)。
type fakeVoiceSTT struct {
mu sync.Mutex
calls int
seen []voice.Audio
reply voice.STTOutcome
block bool
}
func (f *fakeVoiceSTT) STTOutcome(ctx context.Context, audio voice.Audio) voice.STTOutcome {
f.mu.Lock()
f.calls++
f.seen = append(f.seen, audio)
reply, block := f.reply, f.block
f.mu.Unlock()
if block {
<-ctx.Done()
return voice.STTOutcome{Provider: voice.SilentProviderName, Cause: ctx.Err()}
}
return reply
}
func (f *fakeVoiceSTT) callCount() int {
f.mu.Lock()
defer f.mu.Unlock()
return f.calls
}
func (f *fakeVoiceSTT) lastAudio(t *testing.T) voice.Audio {
t.Helper()
f.mu.Lock()
defer f.mu.Unlock()
if len(f.seen) == 0 {
t.Fatal("STT 从未被调用")
}
return f.seen[len(f.seen)-1]
}
func (f *fakeVoiceSTT) set(out voice.STTOutcome) {
f.mu.Lock()
defer f.mu.Unlock()
f.reply = out
}
// fakeClock 是可推进的墙上时钟(熔断的 TTL 断言必须确定性)。
type fakeClock struct {
mu sync.Mutex
now time.Time
}
func (c *fakeClock) Now() time.Time {
c.mu.Lock()
defer c.mu.Unlock()
return c.now
}
func (c *fakeClock) advance(d time.Duration) {
c.mu.Lock()
defer c.mu.Unlock()
c.now = c.now.Add(d)
}
// voiceEnv 是入站语音用例的装配:真循环(装配期)+ 假模型 + 记录落库的历史 + 假 STT +
// 可推进时钟。
type voiceEnv struct {
bridge *Bridge
model *stubModelAdapter
history *captureHistory
stt *fakeVoiceSTT
clock *fakeClock
logs *safeBuffer
}
func newVoiceEnv(t *testing.T, stt VoiceSTT, timeout time.Duration) *voiceEnv {
t.Helper()
logs := &safeBuffer{}
log := slog.New(slog.NewTextHandler(logs, nil))
modelAdapter := &stubModelAdapter{}
ag, err := agent.New("测试系统提示词", nil, nil, agent.WithModel(modelAdapter, nil))
if err != nil {
t.Fatalf("构造 agent: %v", err)
}
runner := loop.New(loop.Options{Agent: ag, MaxRounds: 4, Log: log})
sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
hist := &captureHistory{}
clock := &fakeClock{now: time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)}
b, err := New(Options{
Log: log, Loop: runner, Sessions: sess, Owner: testOwnerProfile("telegram", "100"),
History: hist, VoiceSTT: stt, VoiceSTTTimeout: timeout, Now: clock.Now,
})
if err != nil {
t.Fatalf("New: %v", err)
}
platform := &fakeAdapter{name: "telegram"}
b.Add(platform)
return &voiceEnv{bridge: b, model: modelAdapter, history: hist, clock: clock, logs: logs}
}
// voiceMessage 构造一条主人私聊的**纯语音**消息(无正文 + 已内联字节)。
func voiceMessage(name, mime string, data []byte) adapter.UnifiedMessage {
msg := directMessage()
msg.Content = ""
msg.ContentType = adapter.ContentVoice
msg.Media = []adapter.Media{{
Kind: adapter.MediaVoice, Name: name, MIMEType: mime, Size: int64(len(data)), Data: data,
}}
return msg
}
// lastUserContent 返回最近一次落库的用户消息正文(没有则空串)。
func lastUserContent(env *voiceEnv) string {
content := ""
for _, m := range env.history.records() {
if m.Role == history.RoleUser {
content = m.Content
}
}
return content
}
// TestInboundVoiceNormalizedIntoSingleVisibleText 覆盖验收:转写结果写回 Content ⇒
// 模型可见视图与落库的用户消息都是**同一份文本**(不是空行、也不是占位标记)。
func TestInboundVoiceNormalizedIntoSingleVisibleText(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{Text: " 你好呀,昔涟 ", Provider: "cloud"}}
env := newVoiceEnv(t, stt, 0)
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
if stt.callCount() != 1 {
t.Fatalf("STT 调用次数 = %d, want 1", stt.callCount())
}
audio := stt.lastAudio(t)
if audio.Format != "ogg" || string(audio.Data) != "bytes" {
t.Fatalf("STT 收到的音频 = %+v, want ogg/bytes", audio)
}
if got := lastUserContent(env); got != "你好呀,昔涟" {
t.Fatalf("落库的用户正文 = %q, want 转写文本(去首尾空白)", got)
}
if visible := env.model.visibleText(); !strings.Contains(visible, "你好呀,昔涟") {
t.Fatalf("模型可见视图里没有转写文本:%q", visible)
}
if log := env.logs.String(); !strings.Contains(log, "inbound voice transcribed") ||
!strings.Contains(log, "source=services/channelbridge/voice") {
t.Fatalf("转写日志缺少子功能来源:%q", log)
}
}
// TestInboundVoicePlatformTextWinsOverASR 覆盖验收「平台文本优先(不调 ASR)」:微信
// 自带转写已经进了 Content ⇒ 一次上游调用都不能发生(不重复付费)。
func TestInboundVoicePlatformTextWinsOverASR(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{Text: "不该被用到"}}
env := newVoiceEnv(t, stt, 0)
msg := voiceMessage("v.silk", "audio/silk", []byte("bytes"))
msg.Content = "你好" // 平台自带转写(weixin_ilink 的 voice_item.text)
env.bridge.process(context.Background(), msg)
if stt.callCount() != 0 {
t.Fatalf("STT 调用次数 = %d, want 0(平台已给文字,绝不重复付费)", stt.callCount())
}
if got := lastUserContent(env); got != "你好" {
t.Fatalf("落库的用户正文 = %q, want 平台自带转写", got)
}
}
// TestInboundVoicePlaceholderTextStillTranscribes 钉住反面:QQ 的 `[语音]` 是**占位
// 标记**、不是人说的话 ⇒ 照样转写,并把占位标记换成真实文本(否则纯语音整条丢掉)。
func TestInboundVoicePlaceholderTextStillTranscribes(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{Text: "语音里的内容", Provider: "cloud"}}
env := newVoiceEnv(t, stt, 0)
msg := voiceMessage("v.amr", "audio/amr", []byte("bytes"))
msg.Content = "[语音]"
env.bridge.process(context.Background(), msg)
if stt.callCount() != 1 {
t.Fatalf("STT 调用次数 = %d, want 1(占位标记不算文字)", stt.callCount())
}
if got := lastUserContent(env); got != "语音里的内容" {
t.Fatalf("落库的用户正文 = %q, want 转写文本", got)
}
}
// TestInboundVoiceSilentDegradationIsFailure 覆盖验收「静默降级被识破」:链路末端
// `silent` 档返回空文本 + 无错误 —— 它必须被当**失败**处理(落一行诚实提示),
// 绝不能当成一次成功的空转写(那样消息就静默了)。
func TestInboundVoiceSilentDegradationIsFailure(t *testing.T) {
t.Parallel()
t.Run("静默档位带原因", func(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{
Text: "", Provider: voice.SilentProviderName, Cause: voice.ErrNotConfigured,
}}
env := newVoiceEnv(t, stt, 0)
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
got := lastUserContent(env)
if !strings.Contains(got, "没有转写") || !strings.Contains(got, "未配置") {
t.Fatalf("落库正文 = %q, want 一行讲清原因的诚实提示", got)
}
if log := env.logs.String(); !strings.Contains(log, "category=not_configured") {
t.Fatalf("日志缺少可分类的原因:%q", log)
}
})
// 「只信文本 / 只信 err == nil」的实现会在这里静默:空文本 + 无错误看着像一次
// 成功的空转写,但档位是静默档 ⇒ 必须仍按失败处理(这就是本切片的硬要求)。
t.Run("静默档位无原因", func(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{Text: "", Provider: voice.SilentProviderName}}
env := newVoiceEnv(t, stt, 0)
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
got := lastUserContent(env)
if !strings.Contains(got, "没有转写") {
t.Fatalf("落库正文 = %q, want 静默降级的诚实提示(绝不留空消息)", got)
}
if log := env.logs.String(); !strings.Contains(log, "category=silent_degraded") {
t.Fatalf("日志缺少静默降级类别:%q", log)
}
})
}
// TestInboundVoiceEmptyResultIsHonestButNotFailure 区分「服务正常但没识别出文字」与
// 「上游不可用」:前者不熔断(provider 真转写,文本为空)。
func TestInboundVoiceEmptyResultIsHonestButNotFailure(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{Text: " ", Provider: "cloud"}}
env := newVoiceEnv(t, stt, 0)
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
if got := lastUserContent(env); !strings.Contains(got, "没有识别出文字") {
t.Fatalf("落库正文 = %q, want 空识别结果的诚实提示", got)
}
// 第二次仍要正常调上游(没有被误开的熔断挡住)。
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
if stt.callCount() != 2 {
t.Fatalf("STT 调用次数 = %d, want 2(空结果不是故障,不得熔断)", stt.callCount())
}
}
// TestInboundVoiceBreakerStopsUpstreamAfterClassifiedFailure 覆盖验收「熔断(连续失败后
// 不再调用上游 + 消息面仍有诚实提示)」:第一次鉴权失败 ⇒ 熔断;第二条语音**一次上游
// 调用都不发生**,但消息面照样落一行提示(绝不静默)。
func TestInboundVoiceBreakerStopsUpstreamAfterClassifiedFailure(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{
Text: "", Provider: voice.SilentProviderName, Cause: voice.ErrAuthFailed,
}}
env := newVoiceEnv(t, stt, 0)
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
first := lastUserContent(env)
if !strings.Contains(first, "鉴权失败") {
t.Fatalf("第一条的落库正文 = %q, want 鉴权失败的原因方向", first)
}
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
second := lastUserContent(env)
if stt.callCount() != 1 {
t.Fatalf("STT 调用次数 = %d, want 1(熔断期间绝不调上游、不白等超时)", stt.callCount())
}
if !strings.Contains(second, "熔断") {
t.Fatalf("第二条的落库正文 = %q, want 熔断退避的诚实提示", second)
}
if strings.TrimSpace(second) == "" || strings.TrimSpace(first) == "" {
t.Fatal("熔断路径留下了空消息 —— 违反「绝不静默」")
}
}
// TestInboundVoiceBreakerRecoversOnTTLAndSuccess 钉住熔断是**临时**的:TTL 到期恢复调用,
// 一次成功把退避与计数清零(下一次立刻能再调)。
func TestInboundVoiceBreakerRecoversOnTTLAndSuccess(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{
Text: "", Provider: voice.SilentProviderName, Cause: voice.ErrRateLimited,
}}
env := newVoiceEnv(t, stt, 0)
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
if stt.callCount() != 1 {
t.Fatalf("首次调用次数 = %d, want 1", stt.callCount())
}
// TTL 未到:短路。
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
if stt.callCount() != 1 {
t.Fatalf("熔断期间调用次数 = %d, want 1", stt.callCount())
}
// TTL 到期:恢复调用,且这次成功。
env.clock.advance(6 * time.Minute)
stt.set(voice.STTOutcome{Text: "恢复了", Provider: "cloud"})
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
if stt.callCount() != 2 {
t.Fatalf("TTL 到期后调用次数 = %d, want 2(熔断必须自动恢复)", stt.callCount())
}
if got := lastUserContent(env); got != "恢复了" {
t.Fatalf("落库正文 = %q, want 恢复后的转写文本", got)
}
// 成功即清零:紧接着的第三条必须还能调上游(不带退避)。
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
if stt.callCount() != 3 {
t.Fatalf("成功之后调用次数 = %d, want 3(成功必须清零熔断)", stt.callCount())
}
}
// TestInboundVoiceFormatRejectionDoesNotTripBreaker 钉住熔断的**分类**:按内容发生的
// 失败(上游说这份音频不行)不该关掉整个转写面 —— 后续语音必须还能被尝试。
func TestInboundVoiceFormatRejectionDoesNotTripBreaker(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{reply: voice.STTOutcome{
Text: "", Provider: voice.SilentProviderName,
Cause: providerFailureForTest(),
}}
env := newVoiceEnv(t, stt, 0)
env.bridge.process(context.Background(), voiceMessage("v.amr", "audio/amr", []byte("bytes")))
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
if stt.callCount() != 2 {
t.Fatalf("STT 调用次数 = %d, want 2(按内容的失败不得熔断整个转写面)", stt.callCount())
}
if got := lastUserContent(env); strings.TrimSpace(got) == "" {
t.Fatal("失败路径留下了空消息")
}
}
// TestInboundVoiceTimeoutIsHonest 覆盖验收「超时诚实」:上游卡住 ⇒ 独立超时把它掐掉,
// 消息面落一行「超时」,绝不留空。
func TestInboundVoiceTimeoutIsHonest(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{block: true}
env := newVoiceEnv(t, stt, 30*time.Millisecond)
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
got := lastUserContent(env)
if !strings.Contains(got, "超时") {
t.Fatalf("落库正文 = %q, want 超时的诚实提示", got)
}
if log := env.logs.String(); !strings.Contains(log, "category=timeout") {
t.Fatalf("日志缺少超时类别:%q", log)
}
}
// TestInboundVoiceWithoutBytesOrFormatIsHonest 覆盖两条「拿不到音频」的路径:没有字节
// (渠道只给平台内部引用 / 代取失败)与格式不支持 —— 都落诚实提示,且**不调上游**。
func TestInboundVoiceWithoutBytesOrFormatIsHonest(t *testing.T) {
t.Parallel()
t.Run("没有字节", func(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{}
env := newVoiceEnv(t, stt, 0)
msg := voiceMessage("v.ogg", "audio/ogg", nil)
env.bridge.process(context.Background(), msg)
if stt.callCount() != 0 {
t.Fatalf("STT 调用次数 = %d, want 0(没有字节可转写)", stt.callCount())
}
if got := lastUserContent(env); !strings.Contains(got, "没有拿到音频字节") {
t.Fatalf("落库正文 = %q, want 字节缺失的诚实提示", got)
}
})
t.Run("格式不支持", func(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{}
env := newVoiceEnv(t, stt, 0)
msg := voiceMessage("v.silk", "audio/silk", []byte("bytes"))
env.bridge.process(context.Background(), msg)
if stt.callCount() != 0 {
t.Fatalf("STT 调用次数 = %d, want 0(格式不在白名单,不调上游)", stt.callCount())
}
if got := lastUserContent(env); !strings.Contains(got, "格式不支持") {
t.Fatalf("落库正文 = %q, want 格式不支持的诚实提示", got)
}
})
}
// TestInboundPureVoiceIsNotDropped 覆盖验收「inbound 放宽后不再丢纯语音」:无正文、
// 无字节的纯语音消息(telegram 的 file_id 取不到字节那种)也必须跑完流水线并留痕。
func TestInboundPureVoiceIsNotDropped(t *testing.T) {
t.Parallel()
stt := &fakeVoiceSTT{}
env := newVoiceEnv(t, stt, 0)
msg := voiceMessage("", "audio/ogg", nil)
env.bridge.process(context.Background(), msg)
if env.model.calls() == 0 {
t.Fatal("纯语音消息没有跑到模型(被丢空判据丢掉了)")
}
if got := lastUserContent(env); strings.TrimSpace(got) == "" {
t.Fatal("纯语音消息落库成空行 —— 违反「绝不静默」")
}
}
// TestInboundVoiceWithoutSTTWiringIsHonest 覆盖「未接线」:组合根没注入 STT 时,带语音的
// 消息仍要落一行「未配置」的诚实提示(绝不静默丢),且不 panic。
func TestInboundVoiceWithoutSTTWiringIsHonest(t *testing.T) {
t.Parallel()
env := newVoiceEnv(t, nil, 0)
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("bytes")))
got := lastUserContent(env)
if !strings.Contains(got, "没有转写") || !strings.Contains(got, "未配置") {
t.Fatalf("落库正文 = %q, want 未配置的诚实提示", got)
}
}
@@ -0,0 +1,76 @@
package channelbridge
import (
"sync"
"time"
)
// 本文件承载 IM 入站语音转写的**临时熔断**(切片 IM-ASR;design/06 §1.1):
// 上游「再试也是白试」的失败(未配置 / 鉴权失败 / 欠费 / 限流)打开它一小段时间,
// 期间**不再发起任何上游调用** —— 既不白等一次超时,也不每条语音都撞一次付费墙。
//
// 三条纪律:
// - **只对可分类的失败熔断**:判据是 `voice.CircuitWorthy`(哨兵链)。「格式不支持」
// 这类按内容发生的失败**不**熔断,否则一条被拒的语音会让后续语音全部降级。
// - **是临时的**:退避时长逐次翻倍、封顶;任意一次**成功**立即清零恢复,TTL 到期也
// 自动恢复。它刻意**没有**「永久禁用」这一档 —— 拿掉一个能力只能由配置决定。
// - **并发安全**:入站流水线并发(`maxConcurrent`),因此状态只经互斥锁访问。
type voiceBreaker struct {
mu sync.Mutex
openUntil time.Time
failures int
}
// 熔断的具名常量:起始 5 分钟、逐次翻倍、上限 1 小时。上限同时防止位移溢出。
const (
voiceBreakerBaseBackoff = 5 * time.Minute
voiceBreakerMaxBackoff = time.Hour
// voiceBreakerMaxShift 是退避翻倍的次数上限(5min << 6 = 320min,已超上限,
// 因此再翻也没有意义;同时避免 failures 无界增长后位移溢出)。
voiceBreakerMaxShift = 6
// voiceBreakerMinRetryAfter 是提示文案里最小可展示的剩余退避(避免出现「0s」)。
voiceBreakerMinRetryAfter = time.Second
)
// allowed 报告现在是否可以发起一次上游调用。
func (b *voiceBreaker) allowed(now time.Time) bool {
b.mu.Lock()
defer b.mu.Unlock()
return !now.Before(b.openUntil)
}
// open 记一次「再试也是白试」的失败,并按退避时长打开/延长熔断。
func (b *voiceBreaker) open(now time.Time) {
b.mu.Lock()
defer b.mu.Unlock()
b.failures++
shift := b.failures - 1
if shift > voiceBreakerMaxShift {
shift = voiceBreakerMaxShift
}
backoff := voiceBreakerBaseBackoff << shift
if backoff > voiceBreakerMaxBackoff || backoff <= 0 {
backoff = voiceBreakerMaxBackoff
}
b.openUntil = now.Add(backoff)
}
// succeed 记一次成功:**立即**恢复正常(熔断是临时的,成功即恢复)。
func (b *voiceBreaker) succeed() {
b.mu.Lock()
defer b.mu.Unlock()
b.failures = 0
b.openUntil = time.Time{}
}
// retryAfter 报告熔断还剩多久(未打开时为 0)。只用于日志与消息里的诚实提示 ——
// 调用方**不**据此做任何调度(恢复由时钟与成功两条路决定)。
func (b *voiceBreaker) retryAfter(now time.Time) time.Duration {
b.mu.Lock()
defer b.mu.Unlock()
left := b.openUntil.Sub(now)
if left < voiceBreakerMinRetryAfter {
return voiceBreakerMinRetryAfter
}
return left
}