diff --git a/internal/runtime/voice/voice.go b/internal/runtime/voice/voice.go index 2bbad0f..ec65f97 100644 --- a/internal/runtime/voice/voice.go +++ b/internal/runtime/voice/voice.go @@ -32,6 +32,9 @@ const ( // SourcePlatform 表示渠道平台自带转写（例如微信 iLink 的 `voice_item.text`）： // 平台把语音转成文字后送达，我们**不调 ASR**（不重复付费）。 SourcePlatform Source = "platform" + // SourceTTS 表示正文是我们自己合成的语音（出站方向，design/25 §7.1）：成功时 + // Name 指向那份可播副本，降级时 Status=failed 且正文照发文本、只带降级原因。 + SourceTTS Source = "tts" ) // Status 是转写结果的**三态闭集**（design/24 §10.4）。客户端按它分支，**绝不按正文 @@ -39,7 +42,8 @@ const ( type Status string const ( - // StatusTranscribed 表示正文即转写文本（真转写，或平台自带转写）。 + // StatusTranscribed 表示正文即转写文本（真转写，或平台自带转写）；出站方向表示 + // **合成成功**（正文即这段语音的文字内容，Name 指向可播副本）。 StatusTranscribed Status = "transcribed" // StatusEmpty 表示转写成功但没识别出文字（静音 / 太短 / 全是噪声）——是服务正常 // 的空结果，**不是**故障。 @@ -85,6 +89,10 @@ type Ref struct { // DurationMs 是时长，**只有渠道如实给出或转码产物可确证时**才 >0；读不出来就是 0 // （编码时该键缺席）。**绝不估算**（design/24 §10.4）。 DurationMs int64 `json:"durationMs,omitempty"` + // FallbackReason 只在**出站合成降级**时非空（Source=tts、Status=failed）：说明为什么 + // 这条消息没能变成语音（能力未接线 / 预算超时 / 合成为空）。入站转写不使用它 —— + // 入站的失败原因由正文那行诚实提示承载。 + FallbackReason string `json:"fallbackReason,omitempty"` } // IsVoice 报告这条事实是否代表「这条消息来自一条语音消息」（= 来源与状态都在闭集内）。 @@ -110,6 +118,11 @@ func (r Ref) Validate() error { if r.DurationMs < 0 { return fmt.Errorf("%w: negative durationMs %d", ErrBadRef, r.DurationMs) } + // 降级原因只属于失败态：成功还带原因会让人以为「其实降级了」，客户端也就无从按 + // status 单点分支（design/24 §10.4 的「客户端绝不按文案分支」同一条纪律）。 + if r.FallbackReason != "" && r.Status != StatusFailed { + return fmt.Errorf("%w: fallbackReason with status %q", ErrBadRef, r.Status) + } switch { case r.Name == "" && (r.MIME != "" || r.Size != 0): return fmt.Errorf("%w: mime/size without name", ErrBadRef) @@ -124,7 +137,7 @@ func (r Ref) Validate() error { // IsZero 报告这条事实是零值（= 这条消息不是语音消息）。 func (r Ref) IsZero() bool { return r.Source == "" && r.Status == "" && r.Name == "" && r.MIME == "" && - r.Size == 0 && r.DurationMs == 0 + r.Size == 0 && r.DurationMs == 0 && r.FallbackReason == "" } // Encode 把一条语音事实编码成落库用的 JSON 单对象。零值返回 ""（⇒ 列里空串 ⇒ DTO 里 @@ -167,7 +180,9 @@ func Decode(raw string) (Ref, error) { } // validSource / validStatus 是闭集判定，只此一处（编码、解码与渲染共用）。 -func validSource(s Source) bool { return s == SourceASR || s == SourcePlatform } +func validSource(s Source) bool { + return s == SourceASR || s == SourcePlatform || s == SourceTTS +} func validStatus(s Status) bool { return s == StatusTranscribed || s == StatusEmpty || s == StatusFailed diff --git a/internal/runtime/voice/voice_test.go b/internal/runtime/voice/voice_test.go index 645076c..c0f205b 100644 --- a/internal/runtime/voice/voice_test.go +++ b/internal/runtime/voice/voice_test.go @@ -31,6 +31,14 @@ func TestEncodeDecodeRoundTrip(t *testing.T) { Source: SourceASR, Status: StatusEmpty, Name: "0123456789abcdef0123456789abcdef.mp3", MIME: "audio/mpeg", Size: 4321, }}, + {"出站合成成功 + 可播副本", Ref{ + Source: SourceTTS, Status: StatusTranscribed, + Name: "fedcba9876543210fedcba9876543210.ogg", MIME: "audio/ogg", + Size: 8192, DurationMs: 1500, + }}, + {"出站合成降级（正文照发文本）", Ref{ + Source: SourceTTS, Status: StatusFailed, FallbackReason: "budget_exceeded", + }}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -119,6 +127,11 @@ func TestEncodeRejectsBrokenRef(t *testing.T) { {"有 name 无 mime", Ref{Source: SourceASR, Status: StatusTranscribed, Name: "x.ogg", Size: 12}}, {"有 name 无 size", Ref{Source: SourceASR, Status: StatusTranscribed, Name: "x.ogg", MIME: "audio/ogg"}}, {"时长为负", Ref{Source: SourceASR, Status: StatusTranscribed, DurationMs: -1}}, + {"降级原因挂在成功态", Ref{ + Source: SourceTTS, Status: StatusTranscribed, + Name: "x.ogg", MIME: "audio/ogg", Size: 12, FallbackReason: "budget_exceeded", + }}, + {"只有降级原因", Ref{FallbackReason: "budget_exceeded"}}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -149,6 +162,7 @@ func TestDecodeRejectsGarbage(t *testing.T) { {"闭集外的来源", `{"source":"cloud","status":"transcribed"}`}, {"闭集外的状态", `{"source":"asr","status":"done"}`}, {"副本字段不同生共死", `{"source":"asr","status":"transcribed","size":9}`}, + {"降级原因挂在成功态", `{"source":"tts","status":"transcribed","fallbackReason":"x"}`}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -164,6 +178,33 @@ func TestDecodeRejectsGarbage(t *testing.T) { } } +// TestOutboundTTSJSONShape 把**出站**两种落库形状钉死（design/25 §7.1）： +// 成功 = source/status/name/mime/size（+ 可选 durationMs）；降级 = source/status + +// fallbackReason，**没有 name/url**（正文照发文本，客户端据此只挂一个小标）。 +func TestOutboundTTSJSONShape(t *testing.T) { + t.Parallel() + raw, err := Encode(Ref{ + Source: SourceTTS, Status: StatusTranscribed, + Name: "fedcba9876543210fedcba9876543210.ogg", MIME: "audio/ogg", Size: 8192, + }) + if err != nil { + t.Fatalf("Encode(成功): %v", err) + } + want := `{"source":"tts","status":"transcribed","name":"fedcba9876543210fedcba9876543210.ogg",` + + `"mime":"audio/ogg","size":8192}` + if raw != want { + t.Fatalf("出站成功 JSON = %s\n want %s", raw, want) + } + + raw, err = Encode(Ref{Source: SourceTTS, Status: StatusFailed, FallbackReason: "budget_exceeded"}) + if err != nil { + t.Fatalf("Encode(降级): %v", err) + } + if want := `{"source":"tts","status":"failed","fallbackReason":"budget_exceeded"}`; raw != want { + t.Fatalf("出站降级 JSON = %s\n want %s", raw, want) + } +} + // TestRefIsVoiceAndHasCopy 钉住两个判据的边界： // IsVoice = 来源与状态都在闭集内（= 这条消息来自一条语音）； // HasCopy = 有 name（name/mime/size 同生共死，因此 name 是代表）。 @@ -178,6 +219,8 @@ func TestRefIsVoiceAndHasCopy(t *testing.T) { {"零值（不是语音消息）", Ref{}, false, false}, {"平台转写无副本", Ref{Source: SourcePlatform, Status: StatusTranscribed}, true, false}, {"有副本", Ref{Source: SourceASR, Status: StatusFailed, Name: "n.wav", MIME: "audio/wav", Size: 4}, true, true}, + {"出站合成成功", Ref{Source: SourceTTS, Status: StatusTranscribed, Name: "n.ogg", MIME: "audio/ogg", Size: 4}, true, true}, + {"出站降级（算语音消息，但无可播副本）", Ref{Source: SourceTTS, Status: StatusFailed, FallbackReason: "no_capability"}, true, false}, {"只有来源（半截，不算语音）", Ref{Source: SourceASR}, false, false}, } for _, tc := range cases { diff --git a/internal/services/channelbridge/bubbles.go b/internal/services/channelbridge/bubbles.go index d9c8d4a..2ecef8d 100644 --- a/internal/services/channelbridge/bubbles.go +++ b/internal/services/channelbridge/bubbles.go @@ -16,7 +16,8 @@ import ( type bubbleSender func(ctx context.Context, msg adapter.UnifiedMessage) error // sendIMBubbles 是 sendReply（回复路径）与 SendProactive（主动推送） -// 共用的唯一 IM 出站出口。它把内容变成 IM 渠道应收到的纯聊天气泡： +// 共用的唯一 IM 出站出口（带音频的回复走 sendIMBubblesVoice，它只是多挂一条 Media）。 +// 它把内容变成 IM 渠道应收到的聊天气泡： // 角色扮演/动作/旁白标记（在 Web 上渲染为暗色斜体）会被剥离， // 每个

段落作为独立 // 气泡/消息投递（design/17 §2），多气泡连发时相邻发送间隔 0.5-2s， @@ -25,10 +26,19 @@ type bubbleSender func(ctx context.Context, msg adapter.UnifiedMessage) error // // target 携带平台/渠道路由字段，并被复制到每个 // 气泡上（每个气泡只有 Content/ContentType/Timestamp 不同，入站载荷字段一律 -// 不带 —— 见 outboundBubble）；sender 是 +// 不带 —— 见 outboundRoute）；sender 是 // 适配器的出站调用。被剥成空的内容（纯动作行） // 直接丢弃不发送；单个气泡失败时会带上其下标包装返回。 func (b *Bridge) sendIMBubbles(ctx context.Context, target adapter.UnifiedMessage, sender bubbleSender, content string) error { + return b.sendIMBubblesVoice(ctx, target, sender, content, nil) +} + +// sendIMBubblesVoice 是 sendIMBubbles 的**带音频**变体：voice 非 nil 时，那一份音频只 +// 挂在**第一个**气泡上，其余气泡仍是纯文本。这条「只挂第一个」不是优化而是正确性 —— +// 同一条合成语音若跟着每个气泡各发一次，对方会收到 N 份同样的音频。 +// +// voice 为 nil 时行为与 sendIMBubbles 逐字相同（因此主动推送等纯文本出口不受影响）。 +func (b *Bridge) sendIMBubblesVoice(ctx context.Context, target adapter.UnifiedMessage, sender bubbleSender, content string, voice *adapter.Media) error { // splitBubbles 对每个气泡应用 stripNarration（因此 / 与 // 星号漂移的角色扮演形式都会为 IM 剥离），并按

切分。它是 // 唯一的清洗入口，与旧的回复路径共用。 @@ -42,6 +52,9 @@ func (b *Bridge) sendIMBubbles(ctx context.Context, target adapter.UnifiedMessag time.Sleep(b.bubbleGapFn()) } out := outboundBubble(target, bubble) + if i == 0 && voice != nil { + out = outboundVoiceBubble(target, bubble, *voice) + } // 每个气泡各自发送一次，间隔才会落在消息之间。 if err := sender(ctx, out); err != nil { return fmt.Errorf("channelbridge: send bubble #%d of %d: %w", i+1, len(bubbles), err) @@ -50,7 +63,9 @@ func (b *Bridge) sendIMBubbles(ctx context.Context, target adapter.UnifiedMessag return nil } -// outboundBubble 依据一条入站消息的**路由字段**构造一个纯文本出站气泡。 +// outboundRoute 依据一条入站消息的**路由字段**构造一个出站信封：只有路由字段， +// 没有内容。outboundBubble（纯文本）与 outboundVoiceBubble（文本 + 音频）都从它出发， +// 因此「哪些字段允许出站」只有一份白名单。 // // 为什么不能直接 `out := target`：target 是**入站**消息，整条复制会把入站事件的 // 载荷（Media 附件、原始平台载荷、提及列表、回复引用、入站消息 id、发送者昵称） @@ -60,8 +75,9 @@ func (b *Bridge) sendIMBubbles(ctx context.Context, target adapter.UnifiedMessag // 之后**整条返回错误，主人连文本都收不到。 // // 因此这里按白名单只带路由字段（适配器据此解析投递对端），新加进 UnifiedMessage -// 的字段默认不外发：出站是纯文本气泡，本派发器只接收文本。 -func outboundBubble(target adapter.UnifiedMessage, content string) adapter.UnifiedMessage { +// 的字段默认不外发。⚠️ 出站的 `Media` 只能由出站语音自己填（那份副本是我们刚合成的， +// 见 outbound_voice.go），**绝不**从 target 继承。 +func outboundRoute(target adapter.UnifiedMessage) adapter.UnifiedMessage { return adapter.UnifiedMessage{ // 路由字段：平台 / 渠道 / 群 / 线程 / 发送者 id。 Platform: target.Platform, @@ -80,9 +96,29 @@ func outboundBubble(target adapter.UnifiedMessage, content string) adapter.Unifi IsGroup: target.Sender.IsGroup, GroupID: target.Sender.GroupID, }, - // 本气泡的内容：出站永远是纯文本。 - Content: content, - ContentType: adapter.ContentText, - Timestamp: time.Now().UTC(), } } + +// outboundBubble 构造一个**纯文本**出站气泡：路由字段 + 文本内容。 +func outboundBubble(target adapter.UnifiedMessage, content string) adapter.UnifiedMessage { + out := outboundRoute(target) + out.Content = content + out.ContentType = adapter.ContentText + out.Timestamp = time.Now().UTC() + return out +} + +// outboundVoiceBubble 构造**带音频**的出站气泡：与纯文本气泡同一套路由字段与同一份 +// 正文，只是把 ContentType 标成 voice 并挂上那一条 Media。 +// +// 正文**照发**（不是「有音频就不发文字」）：适配器把它当「文字 item + 语音 item」一起 +// 投递（weixin_ilink 的 sendToOne），对方既看得到文字也点得开音频；而任何一侧失败都 +// 只影响那一侧（音频发不出去时文字仍在）。 +func outboundVoiceBubble(target adapter.UnifiedMessage, content string, md adapter.Media) adapter.UnifiedMessage { + out := outboundRoute(target) + out.Content = content + out.ContentType = adapter.ContentVoice + out.Media = []adapter.Media{md} + out.Timestamp = time.Now().UTC() + return out +} diff --git a/internal/services/channelbridge/coalesce_batch.go b/internal/services/channelbridge/coalesce_batch.go index 329ee50..b03c453 100644 --- a/internal/services/channelbridge/coalesce_batch.go +++ b/internal/services/channelbridge/coalesce_batch.go @@ -161,7 +161,7 @@ func (b *Bridge) persistBatchUsers(ctx context.Context, parts []batchMessage, ta notify := b.persistVoiceFirstNotify(p.voiceRef) var row history.Message if target.private { - row = b.persistMain(ctx, mainRow(history.RoleUser, p.prepared.text, "", p.document, p.image, + row = b.persistMainWith(ctx, mainRow(history.RoleUser, p.prepared.text, "", p.document, p.image, target.platform, deviceIDFor(msg), adapterName(a), turnID), notify) } else { row = b.persistIsolatedWith(ctx, isolatedRow(history.RoleUser, p.prepared.text, p.document, p.image, diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index ab34298..13e0e65 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -193,15 +193,31 @@ func (b *Bridge) deliverReply(ctx context.Context, msg adapter.UnifiedMessage, a if res != nil { mood = res.Mood } + // 出站语音（design/25 §7.1）：先把结算结论折成 voice 列事实，再让落库那条路走 + // 「INSERT → 写 voice 列 → notify」（M44 首帧纪律）—— 与入站语音共用同一份 + // persistVoiceFirstNotify，因此首帧上就带着 `voice`，客户端不必等重连。 + ref, hasVoice := b.outboundVoiceRef(resultVoice(res)) + notify := b.notifyPersisted + if hasVoice { + notify = b.persistVoiceFirstNotify(ref) + } // 持久化助手回复（主人私聊 → 主会话 + message_persisted // 广播；受限 → 隔离日志 + channel_persisted 广播），随后 // 送回渠道。 if target.private { - b.persistToMain(ctx, history.RoleAssistant, reply, mood, "", "", target.platform, deviceIDFor(msg), adapterName(a), turnID) + b.persistMainWith(ctx, mainRow(history.RoleAssistant, reply, mood, "", "", target.platform, deviceIDFor(msg), adapterName(a), turnID), notify) } else { - b.persistIsolated(ctx, history.RoleAssistant, reply, "", "", "", sess, msg.Platform, "", adapterName(a), turnID) + b.persistIsolatedWith(ctx, isolatedRow(history.RoleAssistant, reply, "", "", "", sess, msg.Platform, "", adapterName(a), turnID), notify) } - if err := b.sendReply(ctx, msg, a, reply); err != nil { + // 音频只在合成成功、且盘上那份副本读得回来时才挂上；读不回来就诚实退回纯文本 + // （正文照发，WARN 已由 outboundVoiceMedia 记下）。 + var voiceMedia *adapter.Media + if hasVoice { + if md, ok := b.outboundVoiceMedia(msg, ref); ok { + voiceMedia = &md + } + } + if err := b.sendReplyVoice(ctx, msg, a, reply, voiceMedia); err != nil { if errors.Is(err, adapter.ErrUnsupported) { b.log.Debug("channel does not support outbound reply, skipped", "platform", msg.Platform, "session", target.sessionID) diff --git a/internal/services/channelbridge/inbound_voice_copy.go b/internal/services/channelbridge/inbound_voice_copy.go index 301d71e..59adc02 100644 --- a/internal/services/channelbridge/inbound_voice_copy.go +++ b/internal/services/channelbridge/inbound_voice_copy.go @@ -116,6 +116,9 @@ func (b *Bridge) ingestInboundVoiceCopy(ctx context.Context, msg adapter.Unified // 若先广播后写列，这条语音消息的**第一帧**就永远缺 `voice`，客户端只能等重连或下一次 REST // 拉取才升级出播放器（安卓 2026-10-26 回执报的正是这个）。 // +// 入站与出站**共用**它：两边都是「结论已经在手里、列还没写」的同一形状（入站是转写结论， +// 出站是合成结论），首帧纪律因此只有一处实现。 +// // 只换顺序、不换代价：仍是 INSERT + 一次 UPDATE（与 design/24 §10.5 的第二选项同一手法， // 与 document 结论的两段式同族）。非语音消息（ref 零值）时 persistMessageVoice 直接返回， // 因此本包装与 notifyPersisted 逐字等价 —— 普通消息的时序一个字节都没变。 @@ -126,9 +129,9 @@ func (b *Bridge) persistVoiceFirstNotify(ref voiceref.Ref) func(history.Message) } } -// persistMessageVoice 把这条语音的来源事实与副本引用写进刚落库的用户消息 `voice` 列 -// （design/24 §10.2 第 5 步）。它由 runInboundTurn 在用户消息落库之后调用，因此 -// 「引用 + 三态」最终是同一行里的一份事实，客户端不必自己拼。 +// persistMessageVoice 把这条语音的来源事实与副本引用写进刚落库的消息 `voice` 列 +// （design/24 §10.2 第 5 步；出站方向同列同形）。它由落库那条路在消息写进去之后调用， +// 因此「引用 + 三态」最终是同一行里的一份事实，客户端不必自己拼。 // // 三个提前返回都是**正常**情形，不刷日志：这条消息不是语音（零值）、没有落库 // （msg.ID 为空 ⇒ 存储未接或写库失败）、编码后为空串。 @@ -152,7 +155,7 @@ func (b *Bridge) persistMessageVoice(msg history.Message, ref voiceref.Ref) { if err != nil { // 编码失败只可能是坏数据（闭集破坏 / 副本字段不同生共死）：**绝不**写半截， // 但也绝不静默 —— 消息本身已经落库，少的是这条语音标记。 - b.voiceLogger().Error("inbound voice ref not persisted: encode failed", + b.voiceLogger().Error("voice ref not persisted: encode failed", "message", msg.ID, "source", ref.Source, "status", ref.Status, "error", err) return } @@ -163,13 +166,13 @@ func (b *Bridge) persistMessageVoice(msg history.Message, ref voiceref.Ref) { if !ok { // 历史存储没有 voice 列写入视图（迁移 0020 只加列，不替调用方接写路径）： // 如实报告「没写成」，绝不假装写了。 - b.voiceLogger().Warn("inbound voice ref not persisted", + b.voiceLogger().Warn("voice ref not persisted", "reason", "history store has no voice column path", "message", msg.ID, "source", ref.Source, "status", ref.Status) return } if err := writer.UpdateMessageVoice(context.Background(), msg.ID, raw); err != nil { - b.voiceLogger().Warn("inbound voice ref update", + b.voiceLogger().Warn("voice ref update", "err", err, "message", msg.ID, "source", ref.Source, "status", ref.Status) } } diff --git a/internal/services/channelbridge/outbound.go b/internal/services/channelbridge/outbound.go index 921a13d..b49e9e5 100644 --- a/internal/services/channelbridge/outbound.go +++ b/internal/services/channelbridge/outbound.go @@ -17,13 +17,20 @@ import ( // 使适配器解析出消息来自的同一对端，并经 // 共享的 IM 气泡派发器出口（剥离角色扮演/旁白 + 按

切分 + 0.5-2s 间隔）。 func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, a adapter.PlatformAdapter, text string) error { + return b.sendReplyVoice(ctx, in, a, text, nil) +} + +// sendReplyVoice 是 sendReply 的**带音频**变体：voice 非 nil 时随第一个气泡一起送出 +// （出站语音，design/25 §7.1）。正文照发文本 —— 音频是这条回复的另一种呈现，不是 +// 文本的替代品，因此音频发不出去时对方仍读得到字。 +func (b *Bridge) sendReplyVoice(ctx context.Context, in adapter.UnifiedMessage, a adapter.PlatformAdapter, text string, voice *adapter.Media) error { if a == nil { a = b.adapter(in.Platform) } if a == nil { return fmt.Errorf("channelbridge: no adapter for platform %q", in.Platform) } - return b.sendIMBubbles(ctx, in, func(ctx context.Context, msg adapter.UnifiedMessage) error { + return b.sendIMBubblesVoice(ctx, in, func(ctx context.Context, msg adapter.UnifiedMessage) error { return a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{msg}}) - }, text) + }, text, voice) } diff --git a/internal/services/channelbridge/outbound_voice.go b/internal/services/channelbridge/outbound_voice.go new file mode 100644 index 0000000..4692fd8 --- /dev/null +++ b/internal/services/channelbridge/outbound_voice.go @@ -0,0 +1,134 @@ +package channelbridge + +// 本文件承载**出站语音**（design/25 §7.1；切片 M58）：把循环结算出的 +// `loop.OutboundVoice`（M57 的 speak 工具结论）折成两件东西 —— +// +// 1. 一条 `voice` 列事实（`voiceref.Ref`，source=tts）：成功即 +// `{source:"tts",status:"transcribed",name,mime,size,durationMs}`，降级即 +// `{source:"tts",status:"failed",fallbackReason}`。它由 deliverReply 交给与入站 +// **同一条**「先写列、再广播」的落库口径（persistVoiceFirstNotify），因此出站语音 +// 的**第一帧**就带着 `voice`（M44 首帧纪律，入站已有先例）。 +// 2. 一份出站 `adapter.Media`：把盘上那份副本读回字节挂到第一个气泡上，渠道才真的 +// 发得出去音频（bubbles.go 的 outboundVoiceBubble）。 +// +// 三条口径： +// - **绝不写半截**：副本三字段（name/mime/size）不齐就不写 `transcribed`，宁可降级成 +// `failed` + 机器可读原因 —— 一个没有 mime 的 name 会让客户端拼出读不懂的对象。 +// - **拿不到字节就不挂音频**：附件存储未接线、或副本读不出来时，正文照发文本、只留一条 +// WARN。这与入站「没有字节就不落副本」是同一条诚实纪律的反方向。 +// - **绝不静默**：每一类失败都有一条带来源（`services/channelbridge/voice`）的日志。 +// +// ⚠️ 出站 Media 的 URL 刻意留空：`/api/v1/media/` 走 Bearer 鉴权（gateway 的 +// writeAuth），适配器没有、也不该有 owner 令牌，因此它只能拿**字节**，不能自取 URL。 + +import ( + "os" + "path/filepath" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" +) + +// outboundVoiceFallbackCopyIncomplete 是本模块自己产生的降级码（speak 工具不会给出它）： +// 合成报告了成功，但副本事实不完整（缺 mime/size，或 name 不是合法的内容寻址名）。 +// 这是上游缺陷、不该发生，因此记 ERROR；但写进列里的原因仍要机器可读 —— 客户端据此 +// 与真实降级一样只挂小标，而不是去点一个播不出来的播放器。 +const outboundVoiceFallbackCopyIncomplete = "copy_incomplete" + +// resultVoice 取出这一回合的出站语音结论；回合报错（res 为 nil）时没有结论。 +func resultVoice(res *loop.Result) *loop.OutboundVoice { + if res == nil { + return nil + } + return res.Voice +} + +// outboundVoiceRef 把出站语音结论折成一条 `voice` 列事实。第二个返回值报告「这一回合 +// 要不要写 voice 列」—— false 表示这次回复就是纯文本，落库路径与引入本文件之前逐字相同。 +// +// 三种结局（与 loop.OutboundVoice 的注释一一对应）： +// - 未请求（`v == nil` 或 `!v.Requested`）⇒ 不写列，返回 false； +// - 合成成功且有完整副本 ⇒ `transcribed` + 副本事实； +// - 请求了但没有音频 ⇒ `failed` + 降级原因（来自 speak 工具，或本模块的 copy_incomplete）。 +// +// 「Requested 却既无副本又无原因」是上游契约被破坏的状态：不写列（写一个没有原因的 +// failed 等于把「为什么」丢掉），但必须 WARN —— 静默吞掉会让这条语音彻底看不见。 +func (b *Bridge) outboundVoiceRef(v *loop.OutboundVoice) (voiceref.Ref, bool) { + if v == nil || !v.Requested { + return voiceref.Ref{}, false + } + log := b.voiceLogger() + if v.Name != "" { + if v.Mime == "" || v.Size <= 0 || !voiceref.ValidMediaName(v.Name) { + log.Error("outbound voice copy is incomplete, degraded to text", + "name", v.Name, "mime", v.Mime, "size", v.Size, + "duration_ms", v.DurationMs) + return voiceref.Ref{ + Source: voiceref.SourceTTS, Status: voiceref.StatusFailed, + FallbackReason: outboundVoiceFallbackCopyIncomplete, + }, true + } + ref := voiceref.Ref{ + Source: voiceref.SourceTTS, Status: voiceref.StatusTranscribed, + Name: v.Name, MIME: v.Mime, Size: v.Size, + } + // 时长只取上游如实给出且非负的值：Ref.Validate 拒绝负数，绝不为了「有个数」 + // 而估算（与入站同一口径）。 + if v.DurationMs > 0 { + ref.DurationMs = v.DurationMs + } + return ref, true + } + if v.FallbackReason != "" { + return voiceref.Ref{ + Source: voiceref.SourceTTS, Status: voiceref.StatusFailed, + FallbackReason: v.FallbackReason, + }, true + } + log.Warn("outbound voice requested without copy or fallback reason, treated as text", + "cost_ms", v.CostMs, "budget_ms", v.BudgetMs) + return voiceref.Ref{}, false +} + +// outboundVoiceMedia 把盘上那份副本读回字节，折成一个出站媒体项（design/25 §7.1）。 +// 第二个返回值报告「挂不挂得上」：false 时调用方照发纯文本，绝不发一个空音频。 +// +// 目录与上限来自组合根注入的文档存储配置（与入站附件同一份 `documentsCfg`）：副本就是 +// `documents.SaveAudio` 落下的内容寻址文件，因此 `Dir()` + 存储名就是它的路径 —— 不需要 +// （也没有）一条按名读回的存储 API。名字再校验一次是纵深防御：`filepath.Join` 前面绝不 +// 允许出现未经验证的渠道/上游字符串。 +func (b *Bridge) outboundVoiceMedia(msg adapter.UnifiedMessage, ref voiceref.Ref) (adapter.Media, bool) { + if !ref.HasCopy() { + return adapter.Media{}, false + } + log := b.voiceLogger() + if !voiceref.ValidMediaName(ref.Name) { + log.Error("outbound voice media skipped: illegal copy name", "name", ref.Name) + return adapter.Media{}, false + } + store := b.inboundDocumentStore(msg) + if store == nil || store.Dir() == "" { + log.Warn("outbound voice media skipped: document store not wired", + "platform", msg.Platform, "channel", msg.ChannelID, "name", ref.Name) + return adapter.Media{}, false + } + data, err := os.ReadFile(filepath.Join(store.Dir(), ref.Name)) + if err != nil { + log.Warn("outbound voice media skipped: copy unreadable", + "platform", msg.Platform, "channel", msg.ChannelID, "name", ref.Name, "error", err) + return adapter.Media{}, false + } + if int64(len(data)) != ref.Size { + // 列里的 size 与实际字节不一致：以实际字节为准（发出去的就是这些），但要说出来。 + log.Warn("outbound voice copy size differs from column", + "name", ref.Name, "column_size", ref.Size, "actual_size", len(data)) + } + return adapter.Media{ + Kind: adapter.MediaVoice, + Name: ref.Name, + Size: int64(len(data)), + MIMEType: ref.MIME, + Data: data, + }, true +} diff --git a/internal/services/channelbridge/outbound_voice_gateway_e2e_test.go b/internal/services/channelbridge/outbound_voice_gateway_e2e_test.go new file mode 100644 index 0000000..87bdeb7 --- /dev/null +++ b/internal/services/channelbridge/outbound_voice_gateway_e2e_test.go @@ -0,0 +1,438 @@ +package channelbridge + +// 本文件是 M58 的**跨层整链**验收：出站语音结论（`loop.Result.Voice`，M57 的 speak 工具 +// 结算）→ 桥接器落库并写 `chat_messages.voice` 列 → 真网关的**实时帧** + REST 读面； +// 以及同一回合的出站 `adapter.Media` 挂载。 +// +// 手法与入站整链（inbound_voice_gateway_e2e_test.go）同源：**同一个**真 `history.Store` +// 喂给桥接器与网关，断言因此打在真 WS 帧与真 REST 读面上，而不是两个替身各自的想象。 +// 唯一的差别是驱动方式：出站结论来自循环，所以这里用 `stubChatter` 直接给出带 `Voice` +// 的 `Result` —— 真循环 + 假模型跑不出 `Result.Voice`（假模型不会去调 speak 工具）， +// 而本切片要钉的正是「结论已经在手里之后」那一段。 +// +// 钉住四件事： +// 1. 合成成功 ⇒ 助手回复的**第一帧**就带 `voice`（source=tts/status=transcribed + 副本四键）； +// 2. 合成降级 ⇒ `voice.status=failed` + `fallbackReason` 在帧上**与**历史回看（REST）上都可见， +// 且正文照发原文（降级不是「少说一句话」）； +// 3. 音频只挂在**第一个**气泡上（多段回复里重复挂会让对方收到 N 份同样的语音）； +// 4. 出站 Media 的字节就是盘上那份副本（读不回来时宁可不挂，见 outbound_voice.go）。 + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "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" + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway" +) + +const ( + // outboundVoiceText 是所有出站用例的回复正文（单段，便于把「正文原样」写成字面量）。 + outboundVoiceText = "晚安啦，做个好梦。" + // outboundVoiceStoredText 是它经回合可见回复规整（turnreply.NormalizeParagraphs）后 + // **真正落库、真正上帧**的文本：单段被包进一对 `

`。降级用例按这个字面量比对， + // 因此「合成失败有没有动正文」是一个逐字问题，而不是「大概还在」。 + outboundVoiceStoredText = "

晚安啦，做个好梦。

" + // outboundVoiceMultiText 是两段回复：出站音频只该跟着第一段走。 + outboundVoiceMultiText = "第一段话，先说给你听。\n\n第二段话，说给明天听。" + // outboundVoiceAudioMIME 是 `.ogg` 副本的下发类型（voice/media.go 的成对裁决产物）。 + outboundVoiceAudioMIME = "audio/ogg" +) + +// outboundVoiceE2E 打包出站语音整链环境：真桥接器（可注入 `Result.Voice` 的循环替身 + +// 真历史存储 + 真音频落盘配置）+ 真网关（同一个历史存储）+ 组合根的持久化钩子。 +type outboundVoiceE2E struct { + bridge *Bridge + chat *stubChatter + platform *fakeAdapter + // uploads 是桥接器会去读副本的那个目录（`documents.Store.Dir()`）。 + uploads string + ts *httptest.Server + tok string +} + +// newOutboundVoiceE2E 装配整链环境。历史存储是**同一个句柄**喂给桥接器与网关 —— +// 被验证的正是这条边界：桥接器写进 `voice` 列的事实，就是实时帧与 REST 读面投影出来的事实。 +func newOutboundVoiceE2E(t *testing.T, sess session.Session) *outboundVoiceE2E { + t.Helper() + dataDir := t.TempDir() + uploads := filepath.Join(dataDir, "uploads", "documents") + log := slog.New(slog.NewTextHandler(&safeBuffer{}, nil)) + st := openGatewayDocStore(t, log) + + chat := &stubChatter{res: &loop.Result{Text: outboundVoiceText}} + ad := &fakeAdapter{name: "telegram"} + b, err := New(Options{ + Log: log, + Loop: chat, + Sessions: &stubSessions{sess: sess}, + Owner: testOwnerProfile("telegram", "100"), + History: st, + // 真音频落盘配置：副本由 documents.Store.SaveAudio 落下，出站时按同一目录读回。 + DocumentStore: documents.StoreConfig{DataDir: dataDir, UploadsDir: uploads}, + }) + if err != nil { + t.Fatalf("构造桥接器: %v", err) + } + // 气泡间隔在生产里是 0.5-2s：多段用例不该为它付真实时间。 + b.bubbleGapFn = func() time.Duration { return 0 } + b.Add(ad) + + jwt := auth.New(config.AuthConfig{JWTSecret: gatewayDocSecret}) + srv, err := gateway.New( + &config.Config{Auth: config.AuthConfig{AdminUser: "admin", AdminPass: "secret", JWTSecret: gatewayDocSecret}}, + gateway.Deps{Auth: jwt, History: st, DataDir: dataDir}, + ) + if err != nil { + t.Fatalf("构造网关: %v", err) + } + t.Cleanup(srv.Close) + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(ts.Close) + // 🔴 组合根的同一接线（cmd/cyrene/wire_frontend.go：SetOnPersisted(gw.BroadcastPersisted)）。 + b.SetOnPersisted(srv.BroadcastPersisted) + + tok, err := jwt.Sign("acct", auth.RoleOwner, time.Hour) + if err != nil { + t.Fatalf("签发 token: %v", err) + } + return &outboundVoiceE2E{bridge: b, chat: chat, platform: ad, uploads: uploads, ts: ts, tok: tok} +} + +// dialOwnerWS 连上真网关的 `/ws/chat`（owner token），并吃掉握手回放的 `history` 帧， +// 使后续读取从「本回合的第一帧」开始（与入站整链同一手法）。 +func (e *outboundVoiceE2E) dialOwnerWS(t *testing.T) *websocket.Conn { + t.Helper() + wsURL := "ws" + strings.TrimPrefix(e.ts.URL, "http") + "/ws/chat?token=" + e.tok + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("连接 WS: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + ev := readWSFrame(t, conn) + if ev.Type != "history" { + t.Fatalf("第一个帧 = %q, want history（握手回放）", ev.Type) + } + return conn +} + +// storeOutboundCopy 用**真**音频落盘分支（`documents.Store.SaveAudio`）把一份合成产物 +// 落到桥接器会去读的那个目录里，返回它签发的引用。名字/类型/大小都按生产口径算出来 —— +// 绝不手拼一个「看起来像」的文件名，否则本文件测的就不是那条读回路径。 +func storeOutboundCopy(t *testing.T, uploads string, data []byte) documents.Ref { + t.Helper() + store := documents.NewStore(documents.StoreConfig{UploadsDir: uploads}) + ref, err := store.SaveAudio(context.Background(), "voice.ogg", data) + if err != nil { + t.Fatalf("落盘出站语音副本: %v", err) + } + if ref.Name == "" || ref.MIME != outboundVoiceAudioMIME || ref.Size != int64(len(data)) { + t.Fatalf("落盘引用 = %+v, want 名字/%s/%d 字节", ref, outboundVoiceAudioMIME, len(data)) + } + return ref +} + +// nextPersistedFrameWithRole 读到**下一条指定 role 的持久化帧**（跳过其余帧）。出站用例 +// 要的是**助手**那一行（本回合的用户行帧先到），所以按 role 过滤而不是按先后猜 —— +// 也因此不需要 sleep 或重试。 +func nextPersistedFrameWithRole(t *testing.T, conn *websocket.Conn, role string) wsFrame { + t.Helper() + for i := 0; i < 10; i++ { + ev := readWSFrame(t, conn) + switch ev.Type { + case "message_persisted", "channel_persisted": + if messageRole(t, persistedMessage(t, ev)) == role { + return ev + } + } + } + t.Fatalf("读了 10 帧仍没等到 %s 行的持久化帧", role) + return wsFrame{} +} + +// frameContent 取帧内消息的正文。 +func frameContent(t *testing.T, msg map[string]json.RawMessage) string { + t.Helper() + var content string + if err := json.Unmarshal(msg["content"], &content); err != nil { + t.Fatalf("解析帧里的 content: %v", err) + } + return content +} + +// TestIMOutboundVoiceFirstFrameCarriesVoice 钉住出站方向的首帧纪律（M44 的同一条口径）： +// 合成成功的回复，其**第一帧**里就要有 `voice`，且与 REST 读面同一行逐字相等。 +func TestIMOutboundVoiceFirstFrameCarriesVoice(t *testing.T) { + t.Parallel() + env := newOutboundVoiceE2E(t, session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}) + audio := []byte("OggS-outbound-voice-payload") + ref := storeOutboundCopy(t, env.uploads, audio) + env.chat.res = &loop.Result{ + Text: outboundVoiceText, + Voice: &loop.OutboundVoice{ + Requested: true, Name: ref.Name, Mime: ref.MIME, Size: ref.Size, DurationMs: 1500, + }, + } + conn := env.dialOwnerWS(t) + env.bridge.process(context.Background(), directMessage()) + + ev := nextPersistedFrameWithRole(t, conn, history.RoleAssistant) + if ev.Type != "message_persisted" { + t.Fatalf("主人私聊的帧类型 = %q, want message_persisted", ev.Type) + } + msg := persistedMessage(t, ev) + raw, ok := msg["voice"] + if !ok { + t.Fatalf("🔴 出站语音的**第一帧**缺 data.message.voice: %s", ev.Data) + } + shape := canonicalVoice(t, raw) + for _, want := range []string{ + `"source":"tts"`, `"status":"transcribed"`, + `"name":"` + ref.Name + `"`, `"mime":"` + outboundVoiceAudioMIME + `"`, + fmt.Sprintf(`"size":%d`, ref.Size), `"durationMs":1500`, + `"url":"/api/v1/media/` + ref.Name + `"`, + } { + if !strings.Contains(shape, want) { + t.Fatalf("帧里的 voice = %s, 缺 %s", shape, want) + } + } + // 与 REST 读面**同一行**逐字相等：同一份事实，不是两处各算。 + id := messageIDOf(t, msg) + restVoice := voiceOfRowByID(t, fetchMessages(t, env.ts.URL, history.MainSessionID, env.tok), id) + if got, want := canonicalVoice(t, raw), canonicalVoice(t, restVoice); got != want { + t.Fatalf("帧里的 voice = %s, REST 读面同一行 = %s", got, want) + } + // 正文一字未改：语音是这条回复的另一种呈现，不是它的替代品。 + if got := frameContent(t, msg); got != outboundVoiceStoredText { + t.Fatalf("帧里的正文 = %q, want %q", got, outboundVoiceStoredText) + } +} + +// TestIMOutboundVoiceFallbackIsVisibleAsFailed 钉住降级的可见性（切片 M58 的第 3 条）： +// `voice.status=failed` + `fallbackReason` 在**实时帧**与**历史回看**上都看得见，正文照发 +// 原文，且渠道侧只收到纯文本气泡（绝不发一个空音频）。 +func TestIMOutboundVoiceFallbackIsVisibleAsFailed(t *testing.T) { + t.Parallel() + env := newOutboundVoiceE2E(t, session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}) + env.chat.res = &loop.Result{ + Text: outboundVoiceText, + Voice: &loop.OutboundVoice{Requested: true, FallbackReason: "tts_timeout"}, + } + conn := env.dialOwnerWS(t) + env.bridge.process(context.Background(), directMessage()) + + ev := nextPersistedFrameWithRole(t, conn, history.RoleAssistant) + msg := persistedMessage(t, ev) + raw, ok := msg["voice"] + if !ok { + t.Fatalf("🔴 合成降级的**第一帧**缺 data.message.voice（客户端看不到「本来想用语音」）: %s", ev.Data) + } + shape := canonicalVoice(t, raw) + for _, want := range []string{ + `"source":"tts"`, `"status":"failed"`, `"fallbackReason":"tts_timeout"`, + } { + if !strings.Contains(shape, want) { + t.Fatalf("帧里的 voice = %s, 缺 %s", shape, want) + } + } + // 没有副本 ⇒ 副本四键**一起缺席**（不是空串、不是 0）：客户端据此不画播放器。 + for _, absent := range []string{`"name"`, `"mime"`, `"size"`, `"url"`} { + if strings.Contains(shape, absent) { + t.Fatalf("降级的 voice = %s, 不该有 %s", shape, absent) + } + } + // 历史回看（REST 读面）与帧上是同一份事实 —— 隔天再打开也看得到这条降级。 + restVoice := voiceOfRowByID(t, fetchMessages(t, env.ts.URL, history.MainSessionID, env.tok), messageIDOf(t, msg)) + if got, want := canonicalVoice(t, raw), canonicalVoice(t, restVoice); got != want { + t.Fatalf("帧里的 voice = %s, REST 读面同一行 = %s", got, want) + } + // 正文原样：降级只影响「有没有音频」，绝不影响说什么。 + if got := frameContent(t, msg); got != outboundVoiceStoredText { + t.Fatalf("降级回复的正文 = %q, want %q（正文不该被降级改写）", got, outboundVoiceStoredText) + } + if len(env.platform.sends) != 1 { + t.Fatalf("渠道侧气泡数 = %d, want 1", len(env.platform.sends)) + } + sent := env.platform.sends[0] + if sent.ContentType != adapter.ContentText { + t.Fatalf("降级气泡的 ContentType = %q, want %q", sent.ContentType, adapter.ContentText) + } + if len(sent.Media) != 0 { + t.Fatalf("降级气泡带上了 Media = %+v, want 空（绝不发空音频）", sent.Media) + } +} + +// TestIMOutboundVoiceMountsMediaOnFirstBubbleOnly 钉住出站 Media：音频只挂**第一个**气泡， +// 字节就是盘上那份副本，第二个气泡仍是纯文本（同一条合成语音跟着每段各发一次，对方会收到 +// N 份同样的音频）。 +func TestIMOutboundVoiceMountsMediaOnFirstBubbleOnly(t *testing.T) { + t.Parallel() + env := newOutboundVoiceE2E(t, session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}) + audio := []byte("OggS-multi-paragraph-voice") + ref := storeOutboundCopy(t, env.uploads, audio) + env.chat.res = &loop.Result{ + Text: outboundVoiceMultiText, + Voice: &loop.OutboundVoice{ + Requested: true, Name: ref.Name, Mime: ref.MIME, Size: ref.Size, + }, + } + env.bridge.process(context.Background(), directMessage()) + + if len(env.platform.sends) != 2 { + t.Fatalf("两段回复的气泡数 = %d, want 2（%+v）", len(env.platform.sends), env.platform.sends) + } + first := env.platform.sends[0] + if !strings.Contains(first.Content, "第一段话") { + t.Fatalf("第一个气泡的正文 = %q, want 第一段", first.Content) + } + if first.ContentType != adapter.ContentVoice { + t.Fatalf("第一个气泡的 ContentType = %q, want %q", first.ContentType, adapter.ContentVoice) + } + if len(first.Media) != 1 { + t.Fatalf("第一个气泡的 Media = %+v, want 恰好一条", first.Media) + } + md := first.Media[0] + if md.Kind != adapter.MediaVoice { + t.Fatalf("出站媒体 Kind = %q, want %q", md.Kind, adapter.MediaVoice) + } + if md.Name != ref.Name || md.MIMEType != outboundVoiceAudioMIME { + t.Fatalf("出站媒体 = %+v, want name=%s mime=%s", md, ref.Name, outboundVoiceAudioMIME) + } + if md.Size != int64(len(audio)) { + t.Fatalf("出站媒体 Size = %d, want %d", md.Size, len(audio)) + } + if !bytes.Equal(md.Data, audio) { + t.Fatalf("出站媒体字节 = %q, want 盘上那份副本 %q", md.Data, audio) + } + // URL 刻意留空：`/api/v1/media/` 走 Bearer 鉴权，适配器没有（也不该有）owner 令牌。 + if md.URL != "" { + t.Fatalf("出站媒体 URL = %q, want 空（适配器只拿字节）", md.URL) + } + + second := env.platform.sends[1] + if !strings.Contains(second.Content, "第二段话") { + t.Fatalf("第二个气泡的正文 = %q, want 第二段", second.Content) + } + if second.ContentType != adapter.ContentText || len(second.Media) != 0 { + t.Fatalf("第二个气泡 = %+v, want 纯文本且无 Media", second) + } +} + +// TestOutboundVoiceRefShapes 是 outbound_voice.go 的纯函数面：三种结局与「上游契约被破坏」 +// 那一种都必须各归其位 —— 一个没有 mime 的 name 绝不写成 transcribed。 +func TestOutboundVoiceRefShapes(t *testing.T) { + t.Parallel() + b, _, _, _ := newTestBridge(t) + const copyName = "0123456789abcdef0123456789abcdef.ogg" + + cases := []struct { + name string + voice *loop.OutboundVoice + wantOK bool + wantRef string + wantSize int64 + }{ + {name: "nil 结论", voice: nil, wantOK: false}, + {name: "未请求", voice: &loop.OutboundVoice{Requested: false}, wantOK: false}, + { + name: "合成成功且副本完整", + voice: &loop.OutboundVoice{Requested: true, Name: copyName, Mime: outboundVoiceAudioMIME, Size: 8192, DurationMs: 1200}, + wantOK: true, wantRef: `{"source":"tts","status":"transcribed","name":"` + copyName + `","mime":"audio/ogg","size":8192,"durationMs":1200}`, wantSize: 8192, + }, + { + name: "缺 mime ⇒ 降级 copy_incomplete", + voice: &loop.OutboundVoice{Requested: true, Name: copyName, Size: 8192}, + wantOK: true, wantRef: `{"source":"tts","status":"failed","fallbackReason":"copy_incomplete"}`, wantSize: 0, + }, + { + name: "名字不是内容寻址名 ⇒ 降级 copy_incomplete", + voice: &loop.OutboundVoice{Requested: true, Name: "../../etc/passwd", Mime: outboundVoiceAudioMIME, Size: 10}, + wantOK: true, wantRef: `{"source":"tts","status":"failed","fallbackReason":"copy_incomplete"}`, wantSize: 0, + }, + { + name: "请求了但没有音频 ⇒ 降级原因来自上游", + voice: &loop.OutboundVoice{Requested: true, FallbackReason: "budget_exceeded"}, + wantOK: true, wantRef: `{"source":"tts","status":"failed","fallbackReason":"budget_exceeded"}`, wantSize: 0, + }, + {name: "请求了但既无副本又无原因", voice: &loop.OutboundVoice{Requested: true, CostMs: 90}, wantOK: false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + ref, ok := b.outboundVoiceRef(tc.voice) + if ok != tc.wantOK { + t.Fatalf("写列 = %v, want %v（ref=%+v）", ok, tc.wantOK, ref) + } + if !ok { + return + } + raw, err := voiceref.Encode(ref) + if err != nil { + t.Fatalf("编码 voice ref: %v", err) + } + if raw != tc.wantRef { + t.Fatalf("voice ref = %s, want %s", raw, tc.wantRef) + } + if ref.Size != tc.wantSize { + t.Fatalf("ref.Size = %d, want %d", ref.Size, tc.wantSize) + } + }) + } +} + +// voiceRefWithCopy 造一条「盘上应当有一份副本」的 voice 事实。字段按生产口径配齐 +// （name/mime/size 同生共死，否则 Validate 会拒绝）—— 名字交给调用方给，本函数不猜。 +func voiceRefWithCopy(name string, size int64) voiceref.Ref { + return voiceref.Ref{ + Source: voiceref.SourceTTS, Status: voiceref.StatusTranscribed, + Name: name, MIME: outboundVoiceAudioMIME, Size: size, + } +} + +// TestOutboundVoiceMediaSkipsUnreadableCopy 钉住「拿不到字节就不挂音频」：附件存储未接线、 +// 或副本读不回来时，出站媒体一律不挂（正文照发文本，WARN 由 outbound_voice.go 记下）。 +func TestOutboundVoiceMediaSkipsUnreadableCopy(t *testing.T) { + t.Parallel() + const copyName = "0123456789abcdef0123456789abcdef.ogg" + ref := voiceRefWithCopy(copyName, 1) + + // ① 附件存储未接线（UploadsDir 与 DataDir 都空）⇒ inboundDocumentStore 返回 nil。 + bare, _, _, _ := newTestBridge(t) + if md, ok := bare.outboundVoiceMedia(directMessage(), ref); ok { + t.Fatalf("存储未接线却挂上了出站媒体 = %+v", md) + } + + // ② 目录配好了，但盘上没有这份副本 ⇒ 读不回来。 + empty := newOutboundVoiceE2E(t, session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}) + if md, ok := empty.bridge.outboundVoiceMedia(directMessage(), ref); ok { + t.Fatalf("副本不存在却挂上了出站媒体 = %+v", md) + } + + // ③ 副本真在盘上 ⇒ 挂得上，且字节逐字相等。 + env := newOutboundVoiceE2E(t, session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}) + audio := []byte("OggS-readback") + stored := storeOutboundCopy(t, env.uploads, audio) + md, ok := env.bridge.outboundVoiceMedia(directMessage(), voiceRefWithCopy(stored.Name, stored.Size)) + if !ok { + t.Fatalf("副本在盘上却没挂上出站媒体（name=%s）", stored.Name) + } + if !bytes.Equal(md.Data, audio) { + t.Fatalf("读回的字节 = %q, want %q", md.Data, audio) + } +} diff --git a/internal/services/channelbridge/persist.go b/internal/services/channelbridge/persist.go index 4b3b022..232fe28 100644 --- a/internal/services/channelbridge/persist.go +++ b/internal/services/channelbridge/persist.go @@ -77,12 +77,12 @@ func (b *Bridge) notifyPersisted(msg history.Message) { // （inbound_documents_persist.go 的两段式）。未装配历史存储时为空操作， // 因此桥接器在没有存储的情况下行为不变。 // -// 写入本身走 persistMain（与主动推送同一个写入口径），通知走普通通道 notifyPersisted。 -// **主动推送不经过本函数**：它直接调 persistMain，并额外走 notifyProactive 追加一条带 +// 写入本身走 persistMainWith（与主动推送同一个写入口径），通知走普通通道 notifyPersisted。 +// **主动推送不经过本函数**：它直接调 persistMainWith，并额外走 notifyProactive 追加一条带 // 结构化判据的帧（切片 P0b）—— 因此带判据的帧只可能来自主动推送，**普通消息的帧永远 // 不带它**。 func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, document, image, platform, deviceID, adapterID, turnID string) history.Message { - return b.persistMain(ctx, mainRow(role, content, mood, document, image, platform, deviceID, adapterID, turnID), b.notifyPersisted) + return b.persistMainWith(ctx, mainRow(role, content, mood, document, image, platform, deviceID, adapterID, turnID), b.notifyPersisted) } // mainRow 构造一条写进 Web 主会话的行。它只写一次字段清单：persistToMain 与 @@ -94,13 +94,16 @@ func mainRow(role, content, mood, document, image, platform, deviceID, adapterID } } -// persistMain 是「写进主会话」的**唯一**写入口径：persistToMain（普通消息）与 -// 主动推送（proactive.go）共用它，因此两条路的存储语义逐字一致 —— 未接线历史存储 -// 时空操作并返回零值、写库失败 WARN 并返回零值（绝不谎称已落库）、成功则返回刚落库的 -// 完整消息。**差别只在 notify**：那是「这条要不要通知实时通道」的接缝，由调用方给出 -// —— 普通消息传 notifyPersisted（`message_persisted` / `channel_persisted`），主动推送 -// 传 notifyProactive（带结构化判据的同一类帧，切片 P0b）。nil 表示本次写入不通知。 -func (b *Bridge) persistMain(ctx context.Context, msg history.Message, notify func(history.Message)) history.Message { +// persistMainWith 是「写进主会话」的**唯一**写入口径（与 persistIsolatedWith 同一手法）： +// persistToMain（普通消息）、主动推送（proactive.go）与入站/出站语音那条「先写 `voice` 列、 +// 再广播」的路径共用它，因此几条路的存储语义逐字一致 —— 未接线历史存储时空操作并返回零值、 +// 写库失败 WARN 并返回零值（绝不谎称已落库）、成功则返回刚落库的完整消息。 +// +// **差别只在 notify**：那是「这条要不要通知实时通道、以及先写列还是先广播」的接缝，由调用方 +// 给出 —— 普通消息传 notifyPersisted（`message_persisted` / `channel_persisted`），主动推送 +// 传 notifyProactive（带结构化判据的同一类帧，切片 P0b），语音传 persistVoiceFirstNotify +// （先写 `voice` 列再广播，M44 首帧纪律）。nil 表示本次写入不通知。 +func (b *Bridge) persistMainWith(ctx context.Context, msg history.Message, notify func(history.Message)) history.Message { if b.history == nil { return history.Message{} } diff --git a/internal/services/channelbridge/proactive.go b/internal/services/channelbridge/proactive.go index f19912f..688d729 100644 --- a/internal/services/channelbridge/proactive.go +++ b/internal/services/channelbridge/proactive.go @@ -83,7 +83,7 @@ func (b *Bridge) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) // 把主动消息镜像进主对话上下文（design/16 // §4.4 + §4.6.1）。否则模型看不到自己主动说了什么， // 主人后续回复就会显得不连贯（该消息「丢失」了历史）。 - // 它用的是普通回复那条写入口径（persistMain，相同的列、相同的时间戳语义）， + // 它用的是普通回复那条写入口径（persistMainWith，相同的列、相同的时间戳语义）， // 作用域恒为 MainSessionID，落库的同样是**规整后**的那一份（Web 与 IM 看到的必须是 // 同一份），因此落库行为与本切片之前逐字相同；变的只有通知的第二步（见下）。 // turn_id 传 ""（零值）：主动推送是**她单方面开口**，没有对应的入站回合，也就没有 @@ -97,13 +97,13 @@ func (b *Bridge) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) // id），前端按 id 去重 ⇒ 仍是一个气泡、仍只提醒一次。 // // ② 只在**确实落了库**时发：`message_persisted` 的载荷要求权威 id，客户端拿不到 id - // 会整帧丢弃 —— 发一条没有 id 的帧等于假装提醒过。落库失败本身已由 persistMain + // 会整帧丢弃 —— 发一条没有 id 的帧等于假装提醒过。落库失败本身已由 persistMainWith // 记 WARN（可见，不静默）。 row := history.Message{ SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: out.Content, Platform: target.platform, DeviceID: proactiveDeviceID(target), AdapterID: target.platform, Mode: modeChat, } - if full := b.persistMain(ctx, row, b.notifyPersisted); full.ID != "" { + if full := b.persistMainWith(ctx, row, b.notifyPersisted); full.ID != "" { b.notifyProactive(full) } return nil diff --git a/internal/services/channelbridge/proactive_notify.go b/internal/services/channelbridge/proactive_notify.go index b552331..9041f7a 100644 --- a/internal/services/channelbridge/proactive_notify.go +++ b/internal/services/channelbridge/proactive_notify.go @@ -6,7 +6,7 @@ import ( // 本文件承载 IM 主动推送那条**实时通知**接缝（切片 P0b「IM 路由的主动消息也要在非前台 // 补通知」）。它与 persist.go 的 notifyPersisted 并列：普通消息走那一条，主动推送走这 -// 一条。两条都写主会话，因此「写」是同一份代码（persistMain），差别只在通知。 +// 一条。两条都写主会话，因此「写」是同一份代码（persistMainWith），差别只在通知。 // // 为什么必须分开而不是去猜：主人裁定「她说话该不该提醒我，与走哪条路由无关」，因此 // 主动推送必须自己声明「我是主动推送」。判据是**结构化**的（本次写入走的是哪条通知 diff --git a/internal/services/gateway/history_voice.go b/internal/services/gateway/history_voice.go index 78029e5..04a9eda 100644 --- a/internal/services/gateway/history_voice.go +++ b/internal/services/gateway/history_voice.go @@ -9,8 +9,12 @@ package gateway // // 四条硬口径： // -// - **🎤 判据 = `voice` 键存在**（契约明写）：非语音消息整键缺席（`omitempty` 的 -// 指针 nil），**不是** `{}` 也不是 `null`。客户端绝不从正文里认 `[语音]` 字样。 +// - **`voice` 键存在 = 这条消息有语音事实**：非语音消息整键缺席（`omitempty` 的指针 +// nil），**不是** `{}` 也不是 `null`。客户端绝不从正文里认 `[语音]` 字样。 +// ⚠️ 出站合成降级（source=tts、status=failed + fallbackReason）同样**有键**，但 +// 没有可播副本（name/mime/size/url 四键一起缺席）⇒ 客户端按 `status` 分支：只挂一个 +// 「本来想用语音」的小标、**不**画播放器，也不打「来自语音」的 🎤（正文是文本， +// 不是转写文本）。契约文档对 🎤 判据的措辞由塔统一收口。 // - **副本四键同生共死**：`name`/`mime`/`size`/`url` 只在**真的有副本**时一起出现 // （`Ref.Validate` 在解码时已保证，本文件再用 `HasCopy` 复核一次），绝不出现 // 「`url` 指着一个不存在的 `name`」。 @@ -29,7 +33,8 @@ package gateway // ⚠️ 实时帧那条路对**时序**有依赖（已由 M44 的跨层修复保证）：`voice` 是**读列**得来的， // 所以帧必须在 `voice` 列写好之后才发出。渠道桥原来「先 notify、后写列」，导致入站语音的 // **第一帧**读不到 voice；M44 把它改成「INSERT → 写 voice 列 → notify」（见 channelbridge -// 的 persistBatchUsers / persistVoiceFirstNotify），本文件因此无需再区别对待实时帧与读面。 +// 的 persistBatchUsers / persistVoiceFirstNotify），出站语音（outbound_voice.go）走的是 +// 同一条顺序纪律。本文件因此无需再区别对待实时帧与读面。 import ( "context" @@ -45,10 +50,12 @@ import ( // `source` / `status` 恒在（属于闭集，见 `voice.Source` / `voice.Status`）；其余 // 全部 `omitempty` —— 给不出那一项时它**缺席**，绝不用 `0` 或空串编一个值。 type historyVoice struct { - // Source ∈ asr（服务端识别，含云端）| platform（渠道平台自带转写）。 + // Source ∈ asr（服务端识别，含云端）| platform（渠道平台自带转写）| + // tts（我们自己合成的出站语音）。 Source string `json:"source"` - // Status ∈ transcribed（正文即转写文本）| empty（转写成功但没识别出文字）| - // failed（没拿到转写文本，正文是一行中文诚实提示）。闭集三态。 + // Status ∈ transcribed（正文即转写文本 / 出站合成成功）| empty（转写成功但没识别出 + // 文字）| failed（没拿到转写文本，正文是一行中文诚实提示；出站方向 = 合成降级）。 + // 闭集三态。 Status string `json:"status"` // Name 是音频副本的存储名；无副本时**整键缺席**（与 mime/size/url 同生共死）。 Name string `json:"name,omitempty"` @@ -58,6 +65,10 @@ type historyVoice struct { Size int64 `json:"size,omitempty"` // DurationMs 只有渠道如实给出、或转码产物可确证时才有；读不出来就缺席，绝不估算。 DurationMs int64 `json:"durationMs,omitempty"` + // FallbackReason 只在**出站合成降级**时出现（source=tts、status=failed）：说明这条 + // 消息为什么没能变成语音（能力未接线 / 预算超时 / 合成为空）。正文此时照发文本， + // 客户端据 status=failed 只挂一个小标，**绝不**把它当成可播气泡。 + FallbackReason string `json:"fallbackReason,omitempty"` // URL 是服务端相对路径 `/api/v1/media/`（唯一派生字段，由 MediaPath 一处拼出）。 URL string `json:"url,omitempty"` } @@ -148,6 +159,11 @@ func historyVoiceOf(ref voiceref.Ref) *historyVoice { return nil } out := &historyVoice{Source: string(ref.Source), Status: string(ref.Status)} + // 降级原因只在失败态下发（`Ref.Validate` 已保证它不会挂在其它状态上）：客户端据此 + // 显示「这条本来想用语音」的小标，正文仍是原文本。 + if ref.Status == voiceref.StatusFailed { + out.FallbackReason = ref.FallbackReason + } if ref.HasCopy() { // 四键同生共死：有副本时 name/mime/size/url 一起在，没有时一起缺席。 out.Name, out.MIME, out.Size = ref.Name, ref.MIME, ref.Size