AskaEth
ede00d9928
- 迁移 0020_chat_message_voice：按 design/24 §10 与契约 §7.2 已登记的 voice 形状建列/索引
- 语音回合进编排并落库；voice 引用与统计如实回填（转码/转写没成功就如实标未成功，不假装）
- 复用既有 internal/capability/transcode：非 wav（微信 AMR 等）先转码再送 ASR；失败可分类且不丢消息
⚠️ 部分完成（如实登记）：生产端到端未成立（本片只到落库与转码接入），且**未经验真机**。
397 lines
15 KiB
Go
397 lines
15 KiB
Go
package channelbridge
|
||
|
||
// 本文件锁定**入站语音副本的落盘与 `voice` 列回填**（design/24 §10.2 第 4/5 步；
|
||
// design/25 §7.1 第 ④ 步；切片 CODE-VOICE）。钉住四件事：
|
||
//
|
||
// 1. 可播格式（ogg/wav/mp3…）**原样**落副本，一个字节不改、**不碰转码**；
|
||
// 2. 不可播格式（QQ 的 AMR）在落盘**之前**转码（复用 internal/capability/transcode），
|
||
// 落的是可播副本，存储名带的是**目标扩展名**（不是渠道原文件名）；
|
||
// 3. 转码不可用（未接线 / 没这条路 / 越限 / 解码失败…）⇒ **不落副本、不打假 url**，
|
||
// 只留一条带**分类**的 WARN —— 而**消息本身一个字都不丢**（正文仍是转写文本）。
|
||
// 4. `voice` 列的 JSON 与 design/24 §10.4 的落库形状逐字一致：非语音消息空串、
|
||
// 无副本时 name/mime/size 三个键**一起缺席**（同生共死）。
|
||
//
|
||
// 为什么这些断言必须存在：副本这一路最容易「静默坏掉」——落了一份播不了的字节、
|
||
// 或者给了一个指向不存在文件的 url，客户端只会看到一个转不完的圈，而日志里什么都没有。
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"log/slog"
|
||
"strings"
|
||
"sync"
|
||
"testing"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/transcode"
|
||
"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/documents"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
mmvoice "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice"
|
||
voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice"
|
||
)
|
||
|
||
// voiceHistory 是记录型历史存储（在 documentHistory 之上）再加**本片新增的 `voice` 列
|
||
// 写入视图**（design/24 §10.9-4）。真存储上线后由 `history.Store` 实现同一签名。
|
||
type voiceHistory struct {
|
||
*documentHistory
|
||
voiceMu sync.Mutex
|
||
voices map[string]string
|
||
}
|
||
|
||
func (h *voiceHistory) UpdateMessageVoice(_ context.Context, id, voice string) error {
|
||
h.voiceMu.Lock()
|
||
defer h.voiceMu.Unlock()
|
||
if h.voices == nil {
|
||
h.voices = make(map[string]string)
|
||
}
|
||
h.voices[id] = voice
|
||
return nil
|
||
}
|
||
|
||
// voiceColumnOf 返回 id 所指消息当前的 `voice` 列（被回写过就是回写值）。
|
||
func (h *voiceHistory) voiceColumnOf(id string) string {
|
||
h.voiceMu.Lock()
|
||
updated, ok := h.voices[id]
|
||
h.voiceMu.Unlock()
|
||
if ok {
|
||
return updated
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// lastUser 返回最近落库的那条用户消息（一个回合只有一条用户消息）。
|
||
func (h *voiceHistory) lastUser(t *testing.T) history.Message {
|
||
t.Helper()
|
||
var out history.Message
|
||
found := false
|
||
for _, m := range h.records() {
|
||
if m.Role == history.RoleUser {
|
||
out, found = m, true
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatal("本回合没有落库任何用户消息")
|
||
}
|
||
return out
|
||
}
|
||
|
||
// fakeAudioStore 是音频落盘面的替身：它记录每次保存的**提议名与字节**（好让「转码产物
|
||
// 才是落盘字节」这类纪律可断言），并按真存储的口径给出内容寻址名与下发 MIME。
|
||
type fakeAudioStore struct {
|
||
mu sync.Mutex
|
||
calls []audioSaveCall
|
||
err error
|
||
// incomplete 为 true 时返回一份**残缺**的引用（有 name 没 mime），用来钉住
|
||
// 「同生共死被破坏时不给副本」。
|
||
incomplete bool
|
||
}
|
||
|
||
type audioSaveCall struct {
|
||
name string
|
||
data []byte
|
||
}
|
||
|
||
func (s *fakeAudioStore) SaveAudio(_ context.Context, origName string, data []byte) (documents.Ref, error) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
s.calls = append(s.calls, audioSaveCall{name: origName, data: append([]byte(nil), data...)})
|
||
if s.err != nil {
|
||
return documents.Ref{}, s.err
|
||
}
|
||
if s.incomplete {
|
||
return documents.Ref{Name: "aabbccddeeff00112233445566778899.ogg"}, nil
|
||
}
|
||
ext := documents.ExtOf(origName)
|
||
return documents.Ref{
|
||
Name: contentAddressedName + ext,
|
||
MIME: "audio/" + strings.TrimPrefix(ext, "."),
|
||
Size: int64(len(data)),
|
||
}, nil
|
||
}
|
||
|
||
func (s *fakeAudioStore) callCount() int {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return len(s.calls)
|
||
}
|
||
|
||
func (s *fakeAudioStore) lastCall(t *testing.T) audioSaveCall {
|
||
t.Helper()
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if len(s.calls) == 0 {
|
||
t.Fatal("音频存储从未被调用")
|
||
}
|
||
return s.calls[len(s.calls)-1]
|
||
}
|
||
|
||
// contentAddressedName 是一份合法的存储主名（32 位小写十六进制），与真存储的口径一致。
|
||
const contentAddressedName = "0123456789abcdef0123456789abcdef"
|
||
|
||
// fakeVoiceTranscoder 是转码替身：按 routes 回答能力门，按 out/err 回答转换结果。
|
||
type fakeVoiceTranscoder struct {
|
||
routes map[string]bool
|
||
out transcode.Result
|
||
err error
|
||
mu sync.Mutex
|
||
calls int
|
||
}
|
||
|
||
func newFakeVoiceTranscoder() *fakeVoiceTranscoder {
|
||
return &fakeVoiceTranscoder{routes: map[string]bool{"amr>wav": true}}
|
||
}
|
||
|
||
func (f *fakeVoiceTranscoder) Transcode(_ context.Context, req transcode.Request) (transcode.Result, error) {
|
||
f.mu.Lock()
|
||
f.calls++
|
||
f.mu.Unlock()
|
||
if f.err != nil {
|
||
return transcode.Result{}, f.err
|
||
}
|
||
res := f.out
|
||
if res.Format == "" {
|
||
res.Format = req.Target
|
||
}
|
||
return res, nil
|
||
}
|
||
|
||
func (f *fakeVoiceTranscoder) Supports(src, dst transcode.Format) bool {
|
||
return f.routes[transcode.FormatName(src)+">"+transcode.FormatName(dst)]
|
||
}
|
||
|
||
func (f *fakeVoiceTranscoder) Formats() []transcode.Format {
|
||
return []transcode.Format{transcode.FormatAMR, transcode.FormatWAV}
|
||
}
|
||
|
||
func (f *fakeVoiceTranscoder) callCount() int {
|
||
f.mu.Lock()
|
||
defer f.mu.Unlock()
|
||
return f.calls
|
||
}
|
||
|
||
// voiceCopyEnv 是本片用例的装配：真流水线（`process`）+ 假 STT + 可注入的转码器与音频
|
||
// 存储 + 记录型历史（含 voice 列视图）。
|
||
type voiceCopyEnv struct {
|
||
bridge *Bridge
|
||
hist *voiceHistory
|
||
store *fakeAudioStore
|
||
tr *fakeVoiceTranscoder
|
||
logs *safeBuffer
|
||
}
|
||
|
||
func newVoiceCopyEnv(t *testing.T, stt VoiceSTT, tr *fakeVoiceTranscoder, store *fakeAudioStore) *voiceCopyEnv {
|
||
t.Helper()
|
||
logs := &safeBuffer{}
|
||
log := slog.New(slog.NewTextHandler(logs, nil))
|
||
ag, err := agent.New("测试系统提示词", nil, nil, agent.WithModel(&stubModelAdapter{}, nil))
|
||
if err != nil {
|
||
t.Fatalf("构造 agent: %v", err)
|
||
}
|
||
runner := loop.New(loop.Options{Agent: ag, MaxRounds: 4, Log: log})
|
||
hist := &voiceHistory{documentHistory: &documentHistory{captureHistory: &captureHistory{}}}
|
||
opts := Options{
|
||
Log: log, Loop: runner,
|
||
Sessions: &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}},
|
||
Owner: testOwnerProfile("telegram", "100"),
|
||
History: hist, VoiceSTT: stt,
|
||
}
|
||
// 传 nil 与传具体值必须区别对待：`VoiceTranscoder` 是接口，「没接线」正是本片要
|
||
// 覆盖的一条路径（转码不可用 ⇒ 不落副本），装了 nil 指针会让那条判据失效。
|
||
if tr != nil {
|
||
opts.VoiceTranscoder = tr
|
||
}
|
||
if store != nil {
|
||
opts.VoiceAudioStore = store
|
||
}
|
||
b, err := New(opts)
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
b.Add(&fakeAdapter{name: "telegram"})
|
||
return &voiceCopyEnv{bridge: b, hist: hist, store: store, tr: tr, logs: logs}
|
||
}
|
||
|
||
// voiceRefOf 返回本回合用户消息的 `voice` 事实（列空 ⇒ 零值，即「不是语音消息」）。
|
||
func (e *voiceCopyEnv) voiceRefOf(t *testing.T) voiceref.Ref {
|
||
t.Helper()
|
||
msg := e.hist.lastUser(t)
|
||
raw := e.hist.voiceColumnOf(msg.ID)
|
||
ref, err := voiceref.Decode(raw)
|
||
if err != nil {
|
||
t.Fatalf("解码 voice 列 %q: %v", raw, err)
|
||
}
|
||
return ref
|
||
}
|
||
|
||
// contentOf 返回本回合落库的用户消息正文。
|
||
func (e *voiceCopyEnv) contentOf(t *testing.T) string {
|
||
t.Helper()
|
||
return e.hist.lastUser(t).Content
|
||
}
|
||
|
||
// userContentIn 返回记录型历史里最后一条用户消息的正文（没有则判失败）。
|
||
func userContentIn(t *testing.T, hist *captureHistory) string {
|
||
t.Helper()
|
||
content := ""
|
||
found := false
|
||
for _, m := range hist.records() {
|
||
if m.Role == history.RoleUser {
|
||
content, found = m.Content, true
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatal("没有落库任何用户消息")
|
||
}
|
||
return content
|
||
}
|
||
|
||
// transcodingSTT 是一个「转写成功」的假 STT，供不需要关心转写细节的用例使用。
|
||
func transcodingSTT(text string) *fakeVoiceSTT {
|
||
return &fakeVoiceSTT{reply: mmvoice.STTOutcome{Text: text, Provider: "cloud"}}
|
||
}
|
||
|
||
// TestInboundVoiceCopyStoredAsIsForPlayableFormat 覆盖设计口径①（design/25 §7.1）：
|
||
// 源格式可播 ⇒ **原样落副本**，一个字节不改，且**一次转码都不发生**（可播还转一圈等于
|
||
// 白烧 CPU、把音质往下拽）。同时钉住落库 JSON 的逐字形状。
|
||
func TestInboundVoiceCopyStoredAsIsForPlayableFormat(t *testing.T) {
|
||
t.Parallel()
|
||
store := &fakeAudioStore{}
|
||
tr := newFakeVoiceTranscoder()
|
||
env := newVoiceCopyEnv(t, transcodingSTT("你好"), tr, store)
|
||
|
||
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("OGGDATA")))
|
||
|
||
if store.callCount() != 1 {
|
||
t.Fatalf("音频存储调用次数 = %d, want 1", store.callCount())
|
||
}
|
||
call := store.lastCall(t)
|
||
if call.name != "voice.ogg" {
|
||
t.Fatalf("存储提议名 = %q, want voice.ogg（扩展名决定按哪种格式裁决）", call.name)
|
||
}
|
||
if string(call.data) != "OGGDATA" {
|
||
t.Fatalf("落盘字节 = %q, want 原样（可播格式不转码）", call.data)
|
||
}
|
||
if tr.callCount() != 0 {
|
||
t.Fatalf("转码调用次数 = %d, want 0（可播格式绝不转码）", tr.callCount())
|
||
}
|
||
ref := env.voiceRefOf(t)
|
||
if ref.Source != voiceref.SourceASR || ref.Status != voiceref.StatusTranscribed {
|
||
t.Fatalf("voice 事实 = %+v, want asr/transcribed", ref)
|
||
}
|
||
if ref.Name != contentAddressedName+".ogg" || ref.MIME != "audio/ogg" || ref.Size != 7 {
|
||
t.Fatalf("副本事实 = %+v, want 服务端存储名 + audio/ogg + 7 字节", ref)
|
||
}
|
||
if log := env.logs.String(); !strings.Contains(log, "inbound voice copy stored") ||
|
||
!strings.Contains(log, "source=services/channelbridge/voice") {
|
||
t.Fatalf("落副本日志缺少子功能来源：%q", log)
|
||
}
|
||
}
|
||
|
||
// TestInboundVoiceCopyTranscodesAMRBeforeStoring 覆盖设计口径②（design/24 §10.14-2 的
|
||
// 核实结论：QQ 的 AMR 浏览器播不了 ⇒ 必须转码）：落的是**转码产物**，存储名带的是**目标**
|
||
// 扩展名（`.wav`），且时长取转码产物可确证的值。
|
||
func TestInboundVoiceCopyTranscodesAMRBeforeStoring(t *testing.T) {
|
||
t.Parallel()
|
||
store := &fakeAudioStore{}
|
||
tr := newFakeVoiceTranscoder()
|
||
tr.out = transcode.Result{Data: []byte("RIFFwavdata"), Format: transcode.FormatWAV, DurationMs: 3200}
|
||
env := newVoiceCopyEnv(t, transcodingSTT("语音内容"), tr, store)
|
||
|
||
env.bridge.process(context.Background(), voiceMessage("v.amr", "audio/amr", []byte("#!AMR\nraw")))
|
||
|
||
if tr.callCount() != 1 {
|
||
t.Fatalf("转码调用次数 = %d, want 1（AMR 不可播）", tr.callCount())
|
||
}
|
||
call := store.lastCall(t)
|
||
if call.name != "voice.wav" {
|
||
t.Fatalf("存储提议名 = %q, want voice.wav（转码产物的扩展名，不是渠道原文件名）", call.name)
|
||
}
|
||
if string(call.data) != "RIFFwavdata" {
|
||
t.Fatalf("落盘字节 = %q, want 转码产物", call.data)
|
||
}
|
||
ref := env.voiceRefOf(t)
|
||
if ref.Name != contentAddressedName+".wav" || ref.DurationMs != 3200 {
|
||
t.Fatalf("副本事实 = %+v, want .wav + durationMs=3200（转码产物可确证）", ref)
|
||
}
|
||
if log := env.logs.String(); !strings.Contains(log, "transcoded=true") {
|
||
t.Fatalf("日志没有标出转码事实：%q", log)
|
||
}
|
||
}
|
||
|
||
// TestInboundVoiceCopyNotStoredWhenTranscodeUnavailable 覆盖设计口径③（design/25 §5.4）：
|
||
// **不可播且转码不可用** ⇒ 不落副本、不打假 url，只留一条带**分类**的 WARN；而**消息本身
|
||
// 一个字都不丢**（正文仍是转写文本）—— 这是本片「转码失败要可分类且不丢消息」的验收。
|
||
func TestInboundVoiceCopyNotStoredWhenTranscodeUnavailable(t *testing.T) {
|
||
t.Parallel()
|
||
store := &fakeAudioStore{}
|
||
// 组合根没把转码能力交进来（VoiceTranscoder 为 nil）——今天生产装配就是这个状态。
|
||
env := newVoiceCopyEnv(t, transcodingSTT("语音内容"), nil, store)
|
||
|
||
env.bridge.process(context.Background(), voiceMessage("v.amr", "audio/amr", []byte("#!AMR\nraw")))
|
||
|
||
if store.callCount() != 0 {
|
||
t.Fatalf("音频存储调用次数 = %d, want 0（不可播就不落，绝不给播不了的 url）", store.callCount())
|
||
}
|
||
ref := env.voiceRefOf(t)
|
||
if !ref.IsVoice() || ref.HasCopy() {
|
||
t.Fatalf("voice 事实 = %+v, want 「是语音但无副本」", ref)
|
||
}
|
||
if ref.Source != voiceref.SourceASR || ref.Status != voiceref.StatusTranscribed {
|
||
t.Fatalf("voice 事实 = %+v, want asr/transcribed（正文确实是转写文本）", ref)
|
||
}
|
||
if got := env.contentOf(t); got != "语音内容" {
|
||
t.Fatalf("落库正文 = %q, want 转写文本（消息绝不因为转码不可用而丢）", got)
|
||
}
|
||
log := env.logs.String()
|
||
if !strings.Contains(log, "category=transcode_not_wired") || !strings.Contains(log, "copy not stored") {
|
||
t.Fatalf("缺少带分类的诚实日志：%q", log)
|
||
}
|
||
}
|
||
|
||
// TestInboundVoiceCopyNotStoredWithoutBytesButMarkStillPersisted 覆盖微信那条既有裁决
|
||
// （design/24 §10.14-1）：平台自带转写 ⇒ 连字节都不取 ⇒ **没有可回放副本**，但 🎤 标记
|
||
// （`source:"platform"`）必须照落 —— 客户端据此显示「这是一条语音」，只是不显示播放控件。
|
||
func TestInboundVoiceCopyNotStoredWithoutBytesButMarkStillPersisted(t *testing.T) {
|
||
t.Parallel()
|
||
store := &fakeAudioStore{}
|
||
env := newVoiceCopyEnv(t, &fakeVoiceSTT{}, newFakeVoiceTranscoder(), store)
|
||
msg := voiceMessage("v.silk", "audio/silk", nil)
|
||
msg.Content = "平台自带转写" // 微信 iLink 的 voice_item.text
|
||
|
||
env.bridge.process(context.Background(), msg)
|
||
|
||
if store.callCount() != 0 {
|
||
t.Fatalf("音频存储调用次数 = %d, want 0（没有字节就没有副本）", store.callCount())
|
||
}
|
||
ref := env.voiceRefOf(t)
|
||
if ref.Source != voiceref.SourcePlatform || ref.Status != voiceref.StatusTranscribed {
|
||
t.Fatalf("voice 事实 = %+v, want platform/transcribed", ref)
|
||
}
|
||
if ref.HasCopy() {
|
||
t.Fatalf("voice 事实 = %+v, want 无副本（name/mime/size 全缺席）", ref)
|
||
}
|
||
}
|
||
|
||
// TestInboundVoiceCopyNotStoredWhenSaveFails 钉住「存储侧失败」的诚实处理：正文与标记
|
||
// 都在，只是没有副本 —— 绝不因为落副本失败就把消息当成失败。
|
||
func TestInboundVoiceCopyNotStoredWhenSaveFails(t *testing.T) {
|
||
t.Parallel()
|
||
store := &fakeAudioStore{err: errors.New("磁盘满了")}
|
||
env := newVoiceCopyEnv(t, transcodingSTT("语音内容"), newFakeVoiceTranscoder(), store)
|
||
|
||
env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("OGGDATA")))
|
||
|
||
ref := env.voiceRefOf(t)
|
||
if !ref.IsVoice() || ref.HasCopy() {
|
||
t.Fatalf("voice 事实 = %+v, want 「是语音但无副本」", ref)
|
||
}
|
||
if got := env.contentOf(t); got != "语音内容" {
|
||
t.Fatalf("落库正文 = %q, want 转写文本", got)
|
||
}
|
||
if log := env.logs.String(); !strings.Contains(log, "inbound voice copy not stored") {
|
||
t.Fatalf("缺少落副本失败的 WARN：%q", log)
|
||
}
|
||
}
|