diff --git a/.env.example b/.env.example index a58752c..75a5537 100644 --- a/.env.example +++ b/.env.example @@ -169,6 +169,10 @@ STT_PROVIDER=dashscope # DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/api/v1 # VOICE_STT_MODEL=qwen-audio-3.0-asr-flash # VOICE_STT_LANGUAGE=zh +# IM 入站语音转写的超时（秒，默认 20）：桥接器会为这次调用举着一个入站并发槽， +# 因此上游卡住不能拖死整条流水线。≤0 或未配置时用桥接器默认值 20s。 +# 只影响 IM 渠道的入站语音；Web 客户端的语音输入不受它约束。 +# VOICE_STT_TIMEOUT_SEC=20 # 已实测：cosyvoice-v3.5-plus / cosyvoice-v2 走 HTTP SpeechSynthesizer 会回 418(引擎拒绝)； # 可用组合为 cosyvoice-v3-flash + 对应 v3 音色(longxiaochun_v3 龙小淳·女助手等)。 # 各模型音色不可混用（v2 名如 longxiaochun 用于 v3 模型会 418）。 diff --git a/cmd/cyrene/wire_channels.go b/cmd/cyrene/wire_channels.go index 0cb56be..c67b18d 100644 --- a/cmd/cyrene/wire_channels.go +++ b/cmd/cyrene/wire_channels.go @@ -6,6 +6,7 @@ import ( "log/slog" "path/filepath" "strings" + "time" "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" @@ -118,6 +119,14 @@ func (a *app) wireChannels() error { // ChannelPersona 从工厂构建所用的同一份配置里解析账号的渠道级角色 // （design/22 §2 方案 b）；bridge 按已连接的适配器缓存它，并在每次重载时刷新。 ChannelPersona: factory.Persona, + // VoiceSTT 把进程内语音链注入入站流水线（切片 IM-ASR；design/06 §1.1）： + // IM 语音在没有平台自带转写时由它在**流水线最前端**归一成文本（写回 + // msg.Content），因此丢空判据 / 最小打扰判定 / 落库 / 模型可见消息看到的是同 + // 一份文本。它在「静默档位」处**不会被当成成功**——语音链末端总是成功，所以 + // bridge 用的是 `STTOutcome`（带档位与被掩盖的真实原因），并据此做临时熔断。 + // a.voice 同时服务 Web 侧 WS 的 voice_input（同一条降级链，不另建一条）。 + VoiceSTT: a.voice, + VoiceSTTTimeout: time.Duration(a.cfg.Voice.VoiceSTTTimeoutSec) * time.Second, }) if berr != nil { return fmt.Errorf("channel bridge: %w", berr) diff --git a/internal/channel/mediafetch/inline.go b/internal/channel/mediafetch/inline.go index 0bfe96f..248da68 100644 --- a/internal/channel/mediafetch/inline.go +++ b/internal/channel/mediafetch/inline.go @@ -1,9 +1,12 @@ package mediafetch -// 本文件承载**四个适配器共享的入站附件内联循环**（切片 IM-1）。四家的差异只在 +// 本文件承载**四个适配器共享的入站附件内联循环**（切片 IM-1）：四家的差异只在 // 「怎么把一个平台媒体引用取回成字节」（telegram 要先 getFile 换路径、weixin 要多 // 一层 AES），预筛、限额、超时、日志与诚实降级的口径完全相同 —— 因此只实现一次， // 由各适配器实现 Fetcher 注入（三次复制必须抽取，AGENTS §10）。 +// +// 语音/音频附件由同目录的 `inline_voice.go` 承载（切片 IM-ASR）：它复用这里的取回 +// 与日志口径，但用一份**独立**的音频白名单（不得污染 `documents.ExtAllowed`）。 import ( "context" @@ -30,11 +33,29 @@ type Fetcher interface { Fetch(ctx context.Context, media adapter.Media) ([]byte, error) } -// Inline 把消息里**文件类**媒体在受支持格式与限额内的字节内联进 `Media.Data`， -// 使下游（channelbridge 的入站摄取）拿到字节。它是四个适配器共享的唯一实现： +// Inline 把消息里**文件类**与**语音类**媒体在受支持格式与限额内的字节内联进 +// `Media.Data`，使下游（channelbridge 的入站摄取 / 语音归一）拿到字节。它是四个适配器 +// 共享的唯一实现： +// +// - 文件类走 `inlineFiles`（图片仍走各自既有链路：图片进视觉链路，design/24 §4.2 末）； +// - 语音类走 `inlineVoices`（切片 IM-ASR，design/06 §1.1、design/08 §4.1）：它用一份 +// **独立的音频白名单**，绝不与 `documents.ExtAllowed` 混淆——后者是 Web 文档上传的 +// 口径，往里加音频格式会污染文档白名单； +// - 两条分支共用同一份取回口径（来源白名单、逐跳重定向校验、边读边限、超时、日志只记 +// 来源不记完整 URL），因此四个适配器零改动即可拿到语音字节。 +// +// 它**不返回错误**：入站附件的失败由「留空字节 + 日志 + 下游诚实提示」表达，绝不 +// 让一条读不到的附件拖垮整个回合（失败不炸）。 +func Inline(ctx context.Context, log *slog.Logger, msg *adapter.UnifiedMessage, f Fetcher) { + if msg == nil || f == nil || log == nil { + return + } + inlineFiles(ctx, log, msg, f) + inlineVoices(ctx, log, msg, f) +} + +// inlineFiles 内联**文件类**媒体： // -// - 只处理**文件类**媒体（图片/语音走各自既有链路：图片进视觉链路，design/24 -// §4.2 末），本切片不改动它们； // - 先按平台给的文件名**预筛扩展名**（`documents.ExtAllowed`，全仓唯一名单）： // 不在名单里 ⇒ **不发起任何下载**，只留一条 Debug 日志（下游会照 M1 既有口径 // 给模型一行诚实提示）； @@ -46,16 +67,13 @@ type Fetcher interface { // **消息照收、回合继续**； // - 已经带字节的媒体（本地产出）原样跳过，绝不重复下载。 // -// 它**不返回错误**：入站附件的失败由「留空字节 + 日志 + 下游诚实提示」表达，绝不 -// 让一条读不到的附件拖垮整个回合（失败不炸）。 -func Inline(ctx context.Context, log *slog.Logger, msg *adapter.UnifiedMessage, f Fetcher) { - if msg == nil || f == nil || log == nil { - return - } +// 语音附件由 `inlineVoices` 处理：两份计数各管一条链路（同一条消息里的文件与语音 +// 互不吃掉对方的额度）。 +func inlineFiles(ctx context.Context, log *slog.Logger, msg *adapter.UnifiedMessage, f Fetcher) { seen := 0 for i := range msg.Media { m := &msg.Media[i] - if !isFileMedia(msg, *m) { + if !IsFileMedia(msg, *m) { continue } if seen >= documents.MaxPerMessage { @@ -98,10 +116,11 @@ func Inline(ctx context.Context, log *slog.Logger, msg *adapter.UnifiedMessage, } } -// isFileMedia 报告一条媒体是不是**文件类**。判定与 channelbridge 的入站摄取同形 +// IsFileMedia 报告一条媒体是不是**文件类**。判定与 channelbridge 的入站摄取同形 // （`MediaFile`，或「只给了内容类型、附件就在 Media 里」的 `ContentFile`）：两边 -// 必须对「哪些附件该被内联」有一致的答案，否则会出现「内联了但没人消费」。 -func isFileMedia(msg *adapter.UnifiedMessage, m adapter.Media) bool { +// 必须对「哪些附件该被内联」有一致的答案，否则会出现「内联了但没人消费」—— +// 因此判定只有这一份，桥接器侧直接复用（切片 IM-ASR 去重）。 +func IsFileMedia(msg *adapter.UnifiedMessage, m adapter.Media) bool { if m.Kind == adapter.MediaFile { return true } diff --git a/internal/channel/mediafetch/inline_test.go b/internal/channel/mediafetch/inline_test.go index 8fe1170..8e55fc2 100644 --- a/internal/channel/mediafetch/inline_test.go +++ b/internal/channel/mediafetch/inline_test.go @@ -197,8 +197,10 @@ func TestInlineSkipsMediaThatAlreadyHasBytes(t *testing.T) { } } -// TestInlineIgnoresNonFileMedia 钉住边界：图片/语音媒体不走本循环（图片走视觉链路）。 -func TestInlineIgnoresNonFileMedia(t *testing.T) { +// TestInlineIgnoresImageAndNonAudioVoice 钉住边界：图片媒体不走本循环（图片走视觉 +// 链路），**格式不可转写**的语音也不取字节（切片 IM-ASR 起，白名单格式的语音**会** +// 被取回，见 inline_voice_test.go）。 +func TestInlineIgnoresImageAndNonAudioVoice(t *testing.T) { t.Parallel() f := &countingFetcher{data: []byte("x")} msg := &adapter.UnifiedMessage{ diff --git a/internal/channel/mediafetch/inline_voice.go b/internal/channel/mediafetch/inline_voice.go new file mode 100644 index 0000000..b8aa8c5 --- /dev/null +++ b/internal/channel/mediafetch/inline_voice.go @@ -0,0 +1,183 @@ +package mediafetch + +// 本文件承载**入站语音/音频附件的字节内联**（切片 IM-ASR，design/06 §1.1、 +// design/08 §4.1）：四个适配器（telegram / qq-obv11 / weixin-ilink / kook）共用同一份 +// `Inline`，因此语音只需在这里补一条分支，渠道侧**零改动**即可把语音字节送进 +// `Media.Data`。 +// +// 四条口径（每条都是本切片刻意钉死的）： +// +// 1. **音频白名单独立于文档白名单**：`documents.ExtAllowed` 是 Web 文档上传的口径 +// （23 项纯文本类 + PDF），**绝不能**往里加音频格式 —— 那会让 Web 也接受 .amr +// 之类的上传，属于悄悄扩大另一个面。这里的放行集合只用于「入站语音能不能转写」。 +// +// 2. **平台已给文字就不取字节**：微信 iLink 的语音自带转写（`voice_item.text`）已经 +// 进了 `Content`，此时既不该重复付费调 ASR，也不该白下载一份用不上的音频。 +// 判据是 `ChatText`（剥掉适配器的媒体占位标记后的真实文本），**不是**「Content +// 非空」—— QQ 的 `[语音]` 是占位标记、不是文字。 +// +// 3. **引用的形态由适配器的 Fetcher 负责解释**：各渠道给出的引用并不都是 URL +// （telegram 给 `file_id`、weixin 给 CDN 相对引用），因此这里**不**做「像不像 +// http 地址」的预判 —— 预判会把 telegram 的语音整条挡掉。取不到字节时由 Fetcher +// 报错（错误只带来源，可安全进日志），下游照实说「没拿到音频字节」。 +// +// 4. **体积按语音口径收紧**：语音是短音频，超限的不下载（ASR 上游按短音频设计， +// base64 还会把请求体放大 4/3）。上限是具名常量 `MaxVoiceBytes`。 + +import ( + "context" + "log/slog" + "strings" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// MaxVoiceBytes 是**单条入站语音**内联的字节上限。比文档上限（20 MiB）小：语音是短 +// 音频，且它要被整段 base64 送进 ASR（请求体膨胀 4/3 后仍要落在同步接口的常见体型内）。 +const MaxVoiceBytes = 5 << 20 + +// mediaPlaceholders 是适配器为「**没有正文**的媒体消息」写进 `Content` 的占位标记 +// （QQ 的 `[图片]`/`[语音]`/…，见 `plugins/channels/qq_obv11/message.go`）。它们描述 +// 的是「这条消息带了什么媒体」，**不是**人说的话：判「平台有没有给文字」时必须剥掉， +// 否则一条纯语音会被当成「已有正文」而整条丢掉转写。 +// +// 全仓只此一份：字节内联（本文件）与桥接器的语音归一（`internal/services/channelbridge`） +// 都用 `ChatText`，两边对「有没有真实文本」必须给出同一个答案。 +// +// ChatText 返回消息正文里**真正的聊天文本**：剥掉媒体占位标记并去掉首尾空白。空串 +// 表示「这条消息没有人类可读的文字」（纯媒体消息，或只有占位标记）。 +func ChatText(content string) string { + s := content + for _, marker := range [...]string{"[图片]", "[语音]", "[视频]", "[文件]", "[表情]"} { + s = strings.ReplaceAll(s, marker, "") + } + return strings.TrimSpace(s) +} + +// AudioFormatOf 报告一条语音媒体的音频格式名（`wav`/`mp3`/…）以及它是否属于**可转写** +// 的放行格式。扩展名优先（平台给的文件名更具体），平台声明的音频 MIME 兜底 —— +// telegram 的 voice 只有 `file_id` 与 `audio/ogg`，**连文件名都没有**，只看扩展名会把 +// 它全数拒掉。两条都拿不到 ⇒ false（deny-by-default：不去下载一条类型不明的音频）。 +func AudioFormatOf(m adapter.Media) (string, bool) { + if ext := documents.ExtOf(displayName(m)); audioExtAllowed(ext) { + return strings.TrimPrefix(ext, "."), true + } + if format := audioMIMEFormat(m.MIMEType); format != "" { + return format, true + } + return "", false +} + +// audioExtAllowed 报告扩展名（含点、小写）是否在**语音**放行集合内。 +// +// 集合口径：① 云端 ASR 常见的压缩容器（mp3/ogg/opus/m4a/aac/amr/flac/webm）； +// ② 裸 PCM（wav）。**不含 SILK**：它需要转码，而微信那条路径用平台自带转写 +// （design/06 §1.1），不引入转码依赖 —— 与其把 SILK 送上去换一个必然的拒绝， +// 不如诚实地说「这个格式不支持」。 +// +// ⚠️ 未确证：DashScope 对 AMR 等格式的接受度未拿到官方口径（「音频规格」页未抓到）， +// 因此这里只保证「我们尝试并如实报告失败」，不保证每种格式都被上游接受。 +func audioExtAllowed(ext string) bool { + switch strings.ToLower(ext) { + case ".wav", ".mp3", ".ogg", ".oga", ".opus", ".m4a", ".aac", ".amr", ".flac", ".webm": + return true + default: + return false + } +} + +// audioMIMEFormat 把平台声明的音频 MIME 归一成格式名；不认识时返回空串。参数部分 +// （`; codecs=…`）先剥掉，大小写不敏感。 +func audioMIMEFormat(mime string) string { + base := strings.ToLower(strings.TrimSpace(mime)) + if i := strings.IndexByte(base, ';'); i >= 0 { + base = strings.TrimSpace(base[:i]) + } + switch base { + case "audio/wav", "audio/wave", "audio/x-wav", "audio/vnd.wave": + return "wav" + case "audio/mpeg", "audio/mp3": + return "mp3" + case "audio/ogg", "audio/opus": + return "ogg" + case "audio/mp4", "audio/m4a", "audio/x-m4a": + return "m4a" + case "audio/aac": + return "aac" + case "audio/amr": + return "amr" + case "audio/flac", "audio/x-flac": + return "flac" + case "audio/webm": + return "webm" + default: + return "" + } +} + +// IsVoiceMedia 报告一条媒体是不是**语音类**。判定与文件类同形（`MediaVoice`，或 +// 「只给了内容类型、附件就在 Media 里」的 `ContentVoice`）。 +func IsVoiceMedia(msg *adapter.UnifiedMessage, m adapter.Media) bool { + if m.Kind == adapter.MediaVoice { + return true + } + return m.Kind == "" && msg.ContentType == adapter.ContentVoice +} + +// inlineVoices 内联**语音类**媒体。它只在「平台没给文字」时才取字节（口径 2）， +// 且每条语音最多取前 `documents.MaxPerMessage` 条（与文件附件同源的上限常量，错误 +// 只是不取字节 —— 下游会照实说「没读到音频字节」）。 +func inlineVoices(ctx context.Context, log *slog.Logger, msg *adapter.UnifiedMessage, f Fetcher) { + if ChatText(msg.Content) != "" { + // 平台已经给了文字（微信自带转写）或这条消息本来就有正文：不需要转写。 + log.Debug("inbound voice not inlined: message already carries chat text", + "platform", msg.Platform, "media", len(msg.Media)) + return + } + seen := 0 + for i := range msg.Media { + m := &msg.Media[i] + if !IsVoiceMedia(msg, *m) { + continue + } + if seen >= documents.MaxPerMessage { + log.Warn("inbound voice attachments beyond per-message limit not inlined", + "platform", msg.Platform, "media", len(msg.Media), "limit", documents.MaxPerMessage) + return + } + seen++ + if len(m.Data) > 0 { + continue + } + format, ok := AudioFormatOf(*m) + if !ok { + // 格式不可转写：**一个字节都不发出去**。这不是失败，是「这个格式本来就不读」。 + log.Debug("inbound voice not inlined: audio format not transcribable", + "platform", msg.Platform, "name", displayName(*m), "mime", m.MIMEType, "size", m.Size) + continue + } + if m.Size > MaxVoiceBytes { + log.Warn("inbound voice not inlined: declared size over limit", + "platform", msg.Platform, "format", format, "mime", m.MIMEType, "size", m.Size, + "limit", MaxVoiceBytes) + continue + } + data, err := f.Fetch(ctx, *m) + if err != nil { + // 只记来源（HostOf），绝不记 URL 原串：平台资产地址可能带签名凭据。 + log.Warn("inbound voice inline failed", + "platform", msg.Platform, "format", format, "mime", m.MIMEType, "size", m.Size, + "host", HostOf(m.URL), "err", err) + continue + } + if len(data) == 0 { + log.Warn("inbound voice inline returned no bytes", + "platform", msg.Platform, "format", format, "mime", m.MIMEType, "host", HostOf(m.URL)) + continue + } + m.Data = data + log.Debug("inbound voice inlined", + "platform", msg.Platform, "format", format, "mime", m.MIMEType, "size", len(data)) + } +} diff --git a/internal/channel/mediafetch/inline_voice_test.go b/internal/channel/mediafetch/inline_voice_test.go new file mode 100644 index 0000000..ff9133a --- /dev/null +++ b/internal/channel/mediafetch/inline_voice_test.go @@ -0,0 +1,238 @@ +package mediafetch + +// 本文件锁定**入站语音字节内联**的口径（切片 IM-ASR）：音频白名单命中/未命中、平台 +// 已给文字就不取字节、占位标记不算文字、体积与条数上限、失败不炸，以及**音频白名单 +// 绝不污染文档白名单**（后者是 Web 文档上传的口径）。 + +import ( + "context" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" +) + +// voiceMessage 构造一条「无正文的纯语音」入站消息（telegram/QQ 的形状）。 +func voiceMessage(url string, m adapter.Media) *adapter.UnifiedMessage { + return &adapter.UnifiedMessage{ + Platform: "test", + ContentType: adapter.ContentVoice, + Media: []adapter.Media{m}, + } +} + +// TestInlineVoiceFetchesAllowedAudio 是正向用例：白名单格式的语音被内联进 Data。 +func TestInlineVoiceFetchesAllowedAudio(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("audio")} + msg := voiceMessage("https://cdn.example.com/a.mp3", adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/a.mp3", Name: "a.mp3", MIMEType: "audio/mpeg", + }) + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 1 { + t.Fatalf("取回次数 = %d, want 1", f.calls()) + } + if string(msg.Media[0].Data) != "audio" { + t.Fatalf("Data = %q, want 取回的字节", msg.Media[0].Data) + } + if format, ok := AudioFormatOf(msg.Media[0]); !ok || format != "mp3" { + t.Fatalf("AudioFormatOf = %q/%v, want mp3/true", format, ok) + } +} + +// TestInlineVoiceAcceptsMIMEOnlyReference 钉住 telegram 的形状：语音只有 `file_id` 与 +// 平台声明的 `audio/ogg`，**连文件名都没有** —— 只看扩展名会把 telegram 语音全数拒掉。 +func TestInlineVoiceAcceptsMIMEOnlyReference(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("ogg")} + msg := voiceMessage("AwACAgIAAxkBAAI", adapter.Media{ + Kind: adapter.MediaVoice, URL: "AwACAgIAAxkBAAI", MIMEType: "audio/ogg", + }) + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 1 { + t.Fatalf("取回次数 = %d, want 1（MIME 兜底必须生效）", f.calls()) + } + format, ok := AudioFormatOf(msg.Media[0]) + if !ok || format != "ogg" { + t.Fatalf("AudioFormatOf = %q/%v, want ogg/true", format, ok) + } +} + +// TestInlineVoiceSkipsUnsupportedAudioFormat 钉住预筛：不在音频白名单内的**一个字节都不 +// 发出去**（SILK 需要转码、可执行文件更不该取）。 +func TestInlineVoiceSkipsUnsupportedAudioFormat(t *testing.T) { + t.Parallel() + for _, name := range []string{"v.silk", "v.exe", "voice", ""} { + t.Run(name, func(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("x")} + msg := voiceMessage("https://cdn.example.com/whatever", adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/whatever", Name: name, + }) + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 0 { + t.Fatalf("格式 %q 不在音频白名单里却发起了取回", name) + } + if len(msg.Media[0].Data) != 0 { + t.Fatalf("格式 %q 不该被内联", name) + } + }) + } +} + +// TestAudioWhitelistDoesNotPolluteDocumentWhitelist 钉住切片的硬红线：音频白名单是 +// **独立**的一份，`documents.ExtAllowed`（Web 文档上传口径）绝不能被音频格式污染。 +func TestAudioWhitelistDoesNotPolluteDocumentWhitelist(t *testing.T) { + t.Parallel() + for _, ext := range []string{".wav", ".mp3", ".ogg", ".oga", ".opus", ".m4a", ".aac", ".amr", ".flac", ".webm"} { + if !audioExtAllowed(ext) { + t.Fatalf("audioExtAllowed(%q) = false, want true", ext) + } + if documents.ExtAllowed(ext) { + t.Fatalf("documents.ExtAllowed(%q) = true —— 音频格式污染了文档白名单", ext) + } + } +} + +// TestInlineVoiceSkipsWhenChatTextPresent 钉住「平台自带转写优先、不重复付费」： +// `Content` 已有真实文本（微信的 voice_item.text）⇒ **连字节都不取**。 +// +// 同一用例里钉住反面：QQ 的 `[语音]` 是**占位标记**、不是文字 ⇒ 仍然要取字节与转写， +// 否则纯语音会被当成「已有正文」而整条丢掉。 +func TestInlineVoiceSkipsWhenChatTextPresent(t *testing.T) { + t.Parallel() + cases := []struct { + name string + content string + want int + }{ + {"平台自带转写", "你好呀", 0}, + {"用户正文", "看看这条语音", 0}, + {"QQ 占位标记", "[语音]", 1}, + {"QQ 占位标记+空白", " [语音] ", 1}, + {"空正文", "", 1}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("x")} + msg := voiceMessage("https://cdn.example.com/a.ogg", adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/a.ogg", Name: "a.ogg", + }) + msg.Content = tc.content + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != tc.want { + t.Fatalf("取回次数 = %d, want %d（正文 %q）", f.calls(), tc.want, tc.content) + } + }) + } +} + +// TestChatTextStripsMediaPlaceholders 钉住「有没有真实文本」的判定只有一份。 +func TestChatTextStripsMediaPlaceholders(t *testing.T) { + t.Parallel() + cases := []struct { + content string + want string + }{ + {"[语音]", ""}, + {"[图片][语音]", ""}, + {" [文件] ", ""}, + {"@300 look [图片][语音]", "@300 look"}, + {"你好", "你好"}, + {"", ""}, + } + for _, tc := range cases { + if got := ChatText(tc.content); got != tc.want { + t.Fatalf("ChatText(%q) = %q, want %q", tc.content, got, tc.want) + } + } +} + +// TestInlineVoiceSkipsExistingBytesAndOversize 钉住两条省流：已有字节不重复下载； +// 平台声明的体积超语音上限时连请求都不发。 +func TestInlineVoiceSkipsExistingBytesAndOversize(t *testing.T) { + t.Parallel() + t.Run("已有字节", func(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("new")} + msg := voiceMessage("https://cdn.example.com/a.ogg", adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/a.ogg", Name: "a.ogg", Data: []byte("local"), + }) + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 0 || string(msg.Media[0].Data) != "local" { + t.Fatalf("已有字节的语音被重复下载或覆盖：calls=%d data=%q", f.calls(), msg.Media[0].Data) + } + }) + t.Run("声明超限", func(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("x")} + msg := voiceMessage("https://cdn.example.com/big.ogg", adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/big.ogg", Name: "big.ogg", + Size: MaxVoiceBytes + 1, + }) + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 0 { + t.Fatalf("声明超限的语音仍被取回 %d 次, want 0", f.calls()) + } + }) +} + +// TestInlineVoiceCapsPerMessage 钉住条数上限（与文件附件同源的常量）。 +func TestInlineVoiceCapsPerMessage(t *testing.T) { + t.Parallel() + f := &countingFetcher{data: []byte("x")} + msg := voiceMessage("https://cdn.example.com/0.ogg", adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/0.ogg", Name: "0.ogg", + }) + for i := 1; i <= documents.MaxPerMessage+2; i++ { + msg.Media = append(msg.Media, adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/x.ogg", Name: "x.ogg", + }) + } + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != documents.MaxPerMessage { + t.Fatalf("取回次数 = %d, want %d", f.calls(), documents.MaxPerMessage) + } +} + +// TestInlineVoiceFailureKeepsMessageAndContinues 钉住「失败不炸」：取不到字节时消息照旧、 +// 后续语音继续尝试，且**绝不**改动正文（下游据此说「没拿到音频字节」）。 +func TestInlineVoiceFailureKeepsMessageAndContinues(t *testing.T) { + t.Parallel() + f := &countingFetcher{err: ErrBadStatus} + msg := voiceMessage("https://cdn.example.com/a.ogg", adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/a.ogg", Name: "a.ogg", + }) + msg.Media = append(msg.Media, adapter.Media{ + Kind: adapter.MediaVoice, URL: "https://cdn.example.com/b.ogg", Name: "b.ogg", + }) + + Inline(context.Background(), testLogger(t), msg, f) + + if f.calls() != 2 { + t.Fatalf("取回次数 = %d, want 2（一次失败不阻断第二次）", f.calls()) + } + for i, m := range msg.Media { + if len(m.Data) != 0 { + t.Fatalf("Media[%d] 在失败时被写入了字节", i) + } + } + if msg.Content != "" { + t.Fatalf("失败路径改动了正文：%q（诚实提示由桥接器负责，取回层不写正文）", msg.Content) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 2919e0c..a2c3d8f 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -181,6 +181,12 @@ type VoiceConfig struct { // STTModel is the DashScope speech-transcription model. STTModel string `env:"VOICE_STT_MODEL" default:"qwen-audio-3.0-asr-flash"` STTLanguage string `env:"VOICE_STT_LANGUAGE" default:"zh"` + // VoiceSTTTimeoutSec bounds one **IM inbound** voice transcription call + // (design/06 §1.1; slice IM-ASR). The channel bridge holds one of its inbound + // concurrency slots for the whole call, so a stuck upstream must not be able to + // drag the pipeline down. Zero or negative falls back to the bridge default + // (20s). It does not affect the Web client's voice_input path. + VoiceSTTTimeoutSec int `env:"VOICE_STT_TIMEOUT_SEC" default:"20"` // TTSModel is the DashScope speech-synthesis model. cosyvoice-v3.5-plus and // cosyvoice-v2 reject the HTTP SpeechSynthesizer call (engine 418); verified diff --git a/internal/runtime/multimodal/voice/dashscope.go b/internal/runtime/multimodal/voice/dashscope.go index f744b45..511eeb6 100644 --- a/internal/runtime/multimodal/voice/dashscope.go +++ b/internal/runtime/multimodal/voice/dashscope.go @@ -395,11 +395,47 @@ func dashscopePost(ctx context.Context, client *http.Client, apiKey, url string, return nil, fmt.Errorf("voice: %w: read: %v", ErrProviderFailed, err) } if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("voice: %w: dashscope status %d: %s", ErrProviderFailed, resp.StatusCode, strings.TrimSpace(string(body))) + return nil, classifyDashscopeStatus(resp.StatusCode, body) } return body, nil } +// classifyDashscopeStatus 把一个非 200 的 DashScope 响应翻成**可分类**的失败：鉴权失败 / +// 欠费额度 / 限流各自成一条哨兵链（都以 ErrProviderFailed 为底），其余原样归入 +// ErrProviderFailed（切片 IM-ASR 的熔断据此判定「再试是不是白试」）。 +// +// 判据只用**状态码 + 错误码/消息里的两类关键字**（阿里云官方口径：`Arrearage`、 +// `Throttling.*`、额度上限），绝不匹配自由文本描述——把「格式不支持」误判成「欠费」 +// 会让一条被拒的语音关掉整个转写面。 +func classifyDashscopeStatus(status int, body []byte) error { + detail := strings.TrimSpace(string(body)) + if err := dashscopeClassifiedError(status, detail); err != nil { + return err + } + return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrProviderFailed, status, detail) +} + +// dashscopeClassifiedError 给出状态码对应的**分类**失败，无法分类时返回 nil。 +func dashscopeClassifiedError(status int, detail string) error { + switch { + case status == http.StatusUnauthorized || status == http.StatusForbidden: + return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrAuthFailed, status, detail) + case status == http.StatusTooManyRequests: + return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrRateLimited, status, detail) + } + // 欠费/额度耗尽：阿里云用 HTTP 400（少数情况 402）配上 Arrearage / + // AllocatedQuotaExceeded 这类错误码，因此按错误码关键字归类（不按状态码）。 + lowered := strings.ToLower(detail) + if strings.Contains(lowered, "arrearage") || strings.Contains(lowered, "quota") || + strings.Contains(lowered, "throttling") { + if strings.Contains(lowered, "throttling") { + return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrRateLimited, status, detail) + } + return fmt.Errorf("voice: %w: dashscope status %d: %s", ErrQuotaExceeded, status, detail) + } + return nil +} + // extractSTTText 从 DashScope ASR 响应中取出转写。ASR // 的输出形状随模型而异，因此按多个候选路径依次尝试。 func extractSTTText(body []byte) (string, error) { @@ -427,14 +463,27 @@ func extractSTTText(body []byte) (string, error) { } // mimeOf 把音频格式映射为其 MIME 类型，供 data: URI / Content-Type 使用。 +// +// 覆盖 IM 入站语音放行的全部格式（`mediafetch` 的音频白名单）：data: URI 的 MIME 是 +// 上游唯一能看到的类型线索，把它写错（例如把 AMR 写成 audio/wav）本身就是一种假话。 +// ⚠️ 未确证：DashScope 对 AMR/SILK 等格式的接受度、以及 `parameters.sample_rate` 对 +// 压缩容器是否有影响，官方「音频规格」页尚未抓到（design/06 §1.2 已标注）。 func mimeOf(format string) string { switch strings.ToLower(format) { case "mp3", "mpeg": return "audio/mpeg" - case "ogg", "oga": + case "ogg", "oga", "opus": return "audio/ogg" case "webm": return "audio/webm" + case "m4a", "mp4": + return "audio/mp4" + case "aac": + return "audio/aac" + case "amr": + return "audio/amr" + case "flac": + return "audio/flac" default: return "audio/wav" } diff --git a/internal/runtime/multimodal/voice/errors.go b/internal/runtime/multimodal/voice/errors.go index 52790ae..bb8afd3 100644 --- a/internal/runtime/multimodal/voice/errors.go +++ b/internal/runtime/multimodal/voice/errors.go @@ -1,6 +1,9 @@ package voice -import "errors" +import ( + "errors" + "fmt" +) // voice 包返回的哨兵错误。每个错误都以 %w 包装， // 使调用方能按类别判断，同时保留原始原因。 @@ -15,4 +18,41 @@ var ( ErrEmptyText = errors.New("voice: empty text") // ErrClosed 在 Close 之后仍有服务调用到达时返回。 ErrClosed = errors.New("voice: service closed") + + // 下面三条是**可分类的上游失败**（切片 IM-ASR 的熔断需要）。它们各自以 + // ErrProviderFailed 为底（`%w`），因此既有的 `errors.Is(err, ErrProviderFailed)` + // 判定逐字不变，而需要「这条失败再试也是白试吗」的调用方可以再细分一层。 + // + // 分类依据是 DashScope 的 HTTP 状态码与错误码（官方口径），**绝不**匹配自由文本 + // 描述：把「格式不支持」误判成「欠费」会让一条被拒的语音关掉整个转写面。 + // + // 归入这三类的失败打开**临时熔断**（未配置同理，见 CircuitWorthy）：它们是 + // 「账户/配额/权限」层面的问题，逐条重试只会白等超时并反复撞付费墙。 + // + // ErrAuthFailed 报告云端鉴权失败（HTTP 401/403，例如 API key 无效或被吊销）。 + ErrAuthFailed = fmt.Errorf("%w: 云端鉴权失败（API key 被拒）", ErrProviderFailed) + // ErrQuotaExceeded 报告云端欠费或额度耗尽（例如阿里云的 Arrearage / + // AllocatedQuotaExceeded）。 + ErrQuotaExceeded = fmt.Errorf("%w: 云端账户欠费或额度耗尽", ErrProviderFailed) + // ErrRateLimited 报告云端限流（HTTP 429 / Throttling.*）。 + ErrRateLimited = fmt.Errorf("%w: 云端限流", ErrProviderFailed) ) + +// CircuitWorthy 报告一次失败是否属于「**再试也是白试**」的类别：未配置 / 鉴权失败 / +// 欠费 / 限流。只有这些才值得熔断退避。 +// +// 「这份音频格式不支持」「解码失败」「音频为空」这类**按内容发生**的失败返回 false： +// 它们不该关掉整个转写面（否则一条被拒的语音会让后续语音全部降级）。判据只认哨兵链。 +func CircuitWorthy(err error) bool { + switch { + case err == nil: + return false + case errors.Is(err, ErrNotConfigured), + errors.Is(err, ErrAuthFailed), + errors.Is(err, ErrQuotaExceeded), + errors.Is(err, ErrRateLimited): + return true + default: + return false + } +} diff --git a/internal/runtime/multimodal/voice/outcome.go b/internal/runtime/multimodal/voice/outcome.go new file mode 100644 index 0000000..a75019a --- /dev/null +++ b/internal/runtime/multimodal/voice/outcome.go @@ -0,0 +1,70 @@ +package voice + +import "context" + +// 本文件承载一次链上 STT 的**完整诚实结果**（切片 IM-ASR）。既有的 +// `STTWithProvider` 只报告「谁最后成功了」，而链路末端是**总是成功**的静默档位 +// （空文本 + `err=nil`）——真实失败原因因此被吞掉。IM 入站语音要拿它做两件事： +// +// 1. **识破静默降级**：静默档位返回的「成功」不是转写成功，绝不能当成功； +// 2. **按原因分类**：未配置 / 鉴权失败 / 欠费 / 限流才值得**临时熔断**——否则每条 +// 语音都会去撞一次付费墙、白等一次超时。 +// +// 这里只**新增**报告面：STT/TTS 的既有语义与 `STTWithProvider` 的返回逐字不变， +// 既有调用方（WS 的 voice_input、`/audio/transcribe`）无需改动。 + +// STTOutcome 是一次链上 STT 的结果。 +type STTOutcome struct { + // Text 是转写文本；空串表示这一趟没有拿到文字。 + Text string + // Provider 是产出该结果的档位名（cloud / local / edge-tts / espeak / + // SilentProviderName）。空串表示链上没有任何 STT 方向可用。 + Provider string + // Cause 是**静默档位掩盖的真实原因**：非 nil 表示这一趟**没有真的转写成功** + // （例如云端 401 / 欠费 / 限流，或整个链路未配置）。nil 且 Provider 不是静默档位 + // 才是真转写——此时 Text 仍可能为空（音频里没人说话），那是**服务正常**的空结果。 + Cause error +} + +// Transcribed 报告这一趟是不是**真转写**（不是静默降级掩盖的成功）。文本本身可能为空 +// （静音/噪声），那属于「服务正常但没识别出文字」，与「上游不可用」必须分开。 +func (o STTOutcome) Transcribed() bool { + return o.Cause == nil && o.Provider != "" && o.Provider != SilentProviderName +} + +// STTOutcome 在链上做一次 STT，并如实报告结果来源与（静默档位掩盖的）真实原因。 +// 它与 STTWithProvider 走同一条降级顺序，只是**不把静默档位当成功**。 +func (c *Chain) STTOutcome(ctx context.Context, audio Audio) STTOutcome { + if len(c.steps) == 0 { + return STTOutcome{Cause: ErrProviderFailed} + } + var lastErr error + for _, p := range c.steps { + stt := p.STT() + if stt == nil { + continue + } + text, err := stt.STT(ctx, audio) + if err != nil { + lastErr = err + continue + } + out := STTOutcome{Text: text, Provider: p.Name()} + if p.Name() != SilentProviderName { + return out + } + // 静默档位：它「成功」只说明链路没崩，真实原因看它前面那一级。 + out.Cause = silenceCause(lastErr) + return out + } + return STTOutcome{Cause: silenceCause(lastErr)} +} + +// silenceCause 把「静默档位为什么成功」翻译成一个可分类的原因：链上确实有过失败时 +// 用那一级的原因；一次失败都没有（链上压根没有云端档位，或它被跳过）就是未配置。 +func silenceCause(lastErr error) error { + if lastErr == nil { + return ErrNotConfigured + } + return lastErr +} diff --git a/internal/runtime/multimodal/voice/outcome_test.go b/internal/runtime/multimodal/voice/outcome_test.go new file mode 100644 index 0000000..32083cc --- /dev/null +++ b/internal/runtime/multimodal/voice/outcome_test.go @@ -0,0 +1,160 @@ +package voice + +// 本文件锁定**一次链上 STT 的诚实报告**（切片 IM-ASR）：`STTOutcome` 必须把静默档位 +// 掩盖的真实原因带出来，`Transcribed()` 必须把「真转写（文本可能为空）」与「静默降级/ +// 未配置」分开，`CircuitWorthy` 只对「再试也是白试」的类别为真。 +// +// 这些断言直接对应桥接器的熔断与诚实提示：判错了就会出现「每条语音都撞一次付费墙」 +// 或「一条被拒的语音关掉整个转写面」。 + +import ( + "errors" + "fmt" + "testing" +) + +func TestSTTOutcomeReportsRealTranscription(t *testing.T) { + t.Parallel() + chain := NewChain(&stubProvider{name: "cloud", stt: &stubSTT{text: "你好"}}) + + out := chain.STTOutcome(t.Context(), Audio{Format: "ogg", Data: []byte("a")}) + + if !out.Transcribed() { + t.Fatalf("Transcribed() = false, want true（provider=%q cause=%v）", out.Provider, out.Cause) + } + if out.Text != "你好" || out.Provider != "cloud" || out.Cause != nil { + t.Fatalf("outcome = %+v, want 你好/cloud/nil", out) + } +} + +// TestSTTOutcomeEmptyTextFromRealProviderIsNotFailure 钉住一个容易搞反的区分：云端 +// 转写成功但**没识别出文字**（静音/噪声）不是故障 —— 它绝不熔断、也不该说「服务挂了」。 +func TestSTTOutcomeEmptyTextFromRealProviderIsNotFailure(t *testing.T) { + t.Parallel() + chain := NewChain(&stubProvider{name: "cloud", stt: &stubSTT{text: ""}}) + + out := chain.STTOutcome(t.Context(), Audio{Format: "ogg", Data: []byte("a")}) + + if !out.Transcribed() { + t.Fatalf("Transcribed() = false, want true（空文本但服务正常）: %+v", out) + } + if out.Provider != "cloud" || out.Cause != nil { + t.Fatalf("outcome = %+v, want cloud/nil", out) + } + if CircuitWorthy(out.Cause) { + t.Fatal("空文本被判成「再试也是白试」，这会误开熔断") + } +} + +// TestSTTOutcomeRevealsSilentDegradation 是本切片的核心断言：静默档位返回「空文本 + +// nil 错」，因此只看文本会把「没有可用的识别后端」当成功 —— `Cause` 必须把它揭穿。 +func TestSTTOutcomeRevealsSilentDegradation(t *testing.T) { + t.Parallel() + chain := NewChain( + &stubProvider{name: "cloud", stt: &stubSTT{err: ErrAuthFailed}}, + SilentProvider(), + ) + + out := chain.STTOutcome(t.Context(), Audio{Format: "ogg", Data: []byte("a")}) + + if out.Transcribed() { + t.Fatalf("静默降级被判成真转写：%+v", out) + } + if out.Provider != SilentProviderName { + t.Fatalf("provider = %q, want %q", out.Provider, SilentProviderName) + } + if !errors.Is(out.Cause, ErrAuthFailed) { + t.Fatalf("cause = %v, want ErrAuthFailed（被静默档位掩盖的真实原因必须带出来）", out.Cause) + } + if !CircuitWorthy(out.Cause) { + t.Fatal("鉴权失败必须可熔断（否则每条语音都撞一次付费墙）") + } +} + +// TestSTTOutcomeWithoutAnyStepReportsNotConfigured 覆盖「链上压根没有云端档位」： +// 静默档位成功但没有前车之鉴 ⇒ 归为「未配置」，同样可熔断。 +func TestSTTOutcomeWithoutAnyStepReportsNotConfigured(t *testing.T) { + t.Parallel() + out := NewChain(SilentProvider()).STTOutcome(t.Context(), Audio{Format: "ogg"}) + + if out.Transcribed() || !errors.Is(out.Cause, ErrNotConfigured) { + t.Fatalf("outcome = %+v, want 未配置（不可当真转写）", out) + } + if !CircuitWorthy(out.Cause) { + t.Fatal("未配置必须可熔断（否则每条语音都白跑一趟）") + } +} + +func TestSTTOutcomeEmptyChain(t *testing.T) { + t.Parallel() + out := NewChain().STTOutcome(t.Context(), Audio{Format: "ogg"}) + + if out.Transcribed() || !errors.Is(out.Cause, ErrProviderFailed) { + t.Fatalf("outcome = %+v, want ErrProviderFailed", out) + } +} + +// TestSTTOutcomeKeepsProviderOrder 钉住它仍走既有的降级顺序（不因为新增报告面而改变 +// 「先云端、后本地」的语义）。 +func TestSTTOutcomeKeepsProviderOrder(t *testing.T) { + t.Parallel() + chain := NewChain( + &stubProvider{name: "cloud", stt: &stubSTT{err: ErrRateLimited}}, + &stubProvider{name: "local", stt: &stubSTT{text: "本地转写"}}, + SilentProvider(), + ) + + out := chain.STTOutcome(t.Context(), Audio{Format: "ogg"}) + + if !out.Transcribed() || out.Text != "本地转写" || out.Provider != "local" { + t.Fatalf("outcome = %+v, want 落到 local", out) + } +} + +// TestCircuitWorthyClassification 钉住分类口径：只有「账户/配额/权限层面」的失败为真； +// 按内容发生的失败（格式不支持、解码失败）为假 —— 后者若熔断，一条被拒的语音会让 +// 后续语音全部降级。 +func TestCircuitWorthyClassification(t *testing.T) { + t.Parallel() + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"未配置", ErrNotConfigured, true}, + {"鉴权失败", ErrAuthFailed, true}, + {"欠费/额度", ErrQuotaExceeded, true}, + {"限流", ErrRateLimited, true}, + {"包装后的鉴权失败", fmt.Errorf("voice: 上游: %w", ErrAuthFailed), true}, + {"普通上游失败", ErrProviderFailed, false}, + {"格式不支持", fmt.Errorf("voice: %w: unsupported format", ErrProviderFailed), false}, + {"未知错误", errors.New("boom"), false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := CircuitWorthy(tc.err); got != tc.want { + t.Fatalf("CircuitWorthy(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +// TestSTTOutcomeErrorStringsAreClassified 钉住鉴权/限流/欠费的错误串里同时带着 +// 分类哨兵与上游细节（既能让调用方分类，也让日志能说清是哪一次失败）。 +func TestSTTOutcomeErrorStringsAreClassified(t *testing.T) { + t.Parallel() + for name, err := range map[string]error{ + "auth": ErrAuthFailed, + "quota": ErrQuotaExceeded, + "rate": ErrRateLimited, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + if !errors.Is(err, ErrProviderFailed) { + t.Fatalf("分类哨兵 %v 必须以 ErrProviderFailed 为底（既有 errors.Is 判定不得变化）", err) + } + }) + } +} diff --git a/internal/runtime/multimodal/voice/provider.go b/internal/runtime/multimodal/voice/provider.go index cbb3de9..153162f 100644 --- a/internal/runtime/multimodal/voice/provider.go +++ b/internal/runtime/multimodal/voice/provider.go @@ -60,10 +60,15 @@ func EspeakProvider(tts TTS) *StaticProvider { return NewProvider("espeak", nil, tts) } +// SilentProviderName 是末端静默档位（SilentProvider）的稳定标识。调用方靠它**识破 +// 静默降级**：该档位总是「成功」（空文本 + 无错误），因此只看转写文本会把「没有可用 +// 的识别后端」当成一次成功的空转写（切片 IM-ASR）。 +const SilentProviderName = "silent" + // SilentProvider 是末端一级。它总是成功，返回空 // 转写与空音频，因此包含它的链永不报错。 func SilentProvider() *StaticProvider { - return NewProvider("silent", silenceSTT{}, silenceTTS{}) + return NewProvider(SilentProviderName, silenceSTT{}, silenceTTS{}) } // silenceSTT 是静默 STT 一级；它返回空转写。 diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 0bd5056..1a8c098 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -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 } diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index fa9432c..f1ba251 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -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 diff --git a/internal/services/channelbridge/inbound_documents.go b/internal/services/channelbridge/inbound_documents.go index f863c2e..9995a24 100644 --- a/internal/services/channelbridge/inbound_documents.go +++ b/internal/services/channelbridge/inbound_documents.go @@ -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 返回消息里全部文件类媒体（保序）。 diff --git a/internal/services/channelbridge/inbound_voice.go b/internal/services/channelbridge/inbound_voice.go new file mode 100644 index 0000000..878e2a9 --- /dev/null +++ b/internal/services/channelbridge/inbound_voice.go @@ -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 +} diff --git a/internal/services/channelbridge/inbound_voice_test.go b/internal/services/channelbridge/inbound_voice_test.go new file mode 100644 index 0000000..97f6e50 --- /dev/null +++ b/internal/services/channelbridge/inbound_voice_test.go @@ -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) + } +} diff --git a/internal/services/channelbridge/voice_breaker.go b/internal/services/channelbridge/voice_breaker.go new file mode 100644 index 0000000..be5a6e3 --- /dev/null +++ b/internal/services/channelbridge/voice_breaker.go @@ -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 +} diff --git a/plugins/channels/kook/message.go b/plugins/channels/kook/message.go index 902a0da..8e528ea 100644 --- a/plugins/channels/kook/message.go +++ b/plugins/channels/kook/message.go @@ -107,7 +107,7 @@ func (a *Adapter) toUnifiedMessage(data json.RawMessage, frame []byte) (adapter. ChannelID: channelID, SenderID: senderID, SenderName: senderName, - Content: content, + Content: chatTextOf(ct, content), ContentType: ct, Media: media, Mentions: mentionsFrom(ex.Mention), @@ -190,6 +190,19 @@ func messageKind(kind int, content string) (adapter.ContentType, []adapter.Media } } +// chatTextOf 给出这条消息的**聊天文本**：只有文本类消息的 content 字段才是人说的话。 +// +// 对图片/视频/文件/音频消息，KOOK 把**附件地址**放在同一个 content 字段里（见 +// messageKind 的说明），它不是聊天文本。把它当正文会同时坏两件事：① 往模型上下文塞 +// 一条 CDN 地址（还有签名参数）；② 让下游「这条消息已经有文字了」的判据误判 —— +// 语音因此在有平台文字时被跳过转写（切片 IM-ASR）。地址仍然完整保留在 Media[].URL。 +func chatTextOf(ct adapter.ContentType, content string) string { + if ct == adapter.ContentText { + return content + } + return "" +} + // mentionsFrom converts an array of mentioned user ids into adapter.Mention. func mentionsFrom(ids []string) []adapter.Mention { if len(ids) == 0 { diff --git a/plugins/channels/kook/message_test.go b/plugins/channels/kook/message_test.go index 7bf1b36..2f1665b 100644 --- a/plugins/channels/kook/message_test.go +++ b/plugins/channels/kook/message_test.go @@ -151,6 +151,36 @@ func TestToUnified_Image(t *testing.T) { if len(msg.Media) != 1 || msg.Media[0].Kind != adapter.MediaImage || msg.Media[0].URL != "https://img/photo.jpg" { t.Errorf("Media = %+v, want image URL https://img/photo.jpg", msg.Media) } + if msg.Content != "" { + t.Errorf("Content = %q, want 空 —— content 字段在图片消息里是**附件地址**，不是聊天文本", msg.Content) + } +} + +// TestToUnified_AudioKeepsAddressOutOfContent 钉住切片 IM-ASR 修掉的正文污染：音频 +// （type=8）的 content 字段是 CDN 地址。写进正文会同时坏两件事 —— 往模型上下文塞一条 +// 带签名参数的地址，并让「这条消息已经有文字了」的判据误判（语音因此被跳过转写）。 +func TestToUnified_AudioKeepsAddressOutOfContent(t *testing.T) { + t.Parallel() + a := New(Config{BotToken: "x"}) + data := json.RawMessage(`{ + "channel_type":"GROUP","type":8,"target_id":"chan-3","author_id":"u-3", + "content":"https://img.kookapp.cn/audio/a.mp3","msg_id":"m-5","msg_timestamp":1700000000000,"nonce":"", + "extra":{"type":8,"guild_id":"g-3","author":{"id":"u-3","username":"u","nickname":"N"}} + }`) + msg, ok := a.toUnifiedMessage(data, nil) + if !ok { + t.Fatal("expected a deliverable message") + } + if msg.ContentType != adapter.ContentVoice { + t.Errorf("ContentType = %q, want voice", msg.ContentType) + } + if msg.Content != "" { + t.Errorf("Content = %q, want 空（地址只配进 Media[].URL）", msg.Content) + } + if len(msg.Media) != 1 || msg.Media[0].Kind != adapter.MediaVoice || + msg.Media[0].URL != "https://img.kookapp.cn/audio/a.mp3" { + t.Errorf("Media = %+v, want 语音地址完整保留在 Media[].URL", msg.Media) + } } func TestToUnified_VideoMapsToFile(t *testing.T) {