AskaEth
5de708a560
裁定（已发安卓）：**能修**，走**写入侧规整** —— 不指望模型输出、不动客户端。 - 新增 internal/runtime/turnreply/paragraph.go 的 NormalizeParagraphs： 把未被 <p> 包裹的段落补上 <p>（按空行分段；<action>/<narration> 段也包进 <p>， 与正确形态 <p><action>…</action>…</p> 一致） - **幂等**守门断言：已完全包裹的文本逐字不变（防二次加工） - 落点覆盖 Web 与 IM 两条落库路径（gateway/session_turn_reply.go、channelbridge/turnreply.go、 proactive.go）⇒ 两边共用同一份规整文本 ⇒ **历史一次性变好、两个客户端零改动**、渲染口径不分叉 （测试与变异详见切片报告）
358 lines
14 KiB
Go
358 lines
14 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"encoding/base64"
|
||
"encoding/json"
|
||
"errors"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/gorilla/websocket"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice"
|
||
)
|
||
|
||
// 本文件是协同项 C1（语音回合接进编排器与持久化）与 C3（voice_output 带
|
||
// sessionId/deviceId）的回归。
|
||
//
|
||
// 修前：runVoiceTurn 直调 runner.Run ⇒ 语音问过的事不进历史、不跨端同步、不参与
|
||
// busy/pending，音频却只能同步等在那个回合上。
|
||
//
|
||
// 修后：语音回合与文本回合走同一条路（进编排器 → 落库 → 流式 → 跨端同步），音频
|
||
// 由回合**结算**时的收尾回调合成并定向回原连接；TTS 失败/降级照旧回帧并带既有
|
||
// provider 降级信号，空正文不打无谓的 TTS。客户端既有绕行（transcribe → WS chat
|
||
// → speech）不受影响，因为它们走的是另一条入口。
|
||
//
|
||
// 断言一律按帧字段与已落库记录判定；日志/错误不按文本子串断言。
|
||
|
||
// waitFrame 排空连接上的帧直到出现给定类型（语音回合的收尾帧会排在若干状态
|
||
// 信号广播之后），超时即失败。
|
||
//
|
||
// 兜底期限取 15s：**断言的是「帧确实到达」，不是「3 秒内到达」** —— 语音回合
|
||
// 要经编排器 → 假 runner（带门控）→ TTS 假实现 → 定向回连接，在整树并行
|
||
// （多包同时跑）或宿主负载高时 3s 会被误判成超时（实测：单跑 -count=20 全绿、
|
||
// 只在整树并行下偶红）。放宽兜底不放宽断言：帧没到照样 Fatal。
|
||
func waitFrame(t *testing.T, c *wsConn, typ string) wsEvent {
|
||
t.Helper()
|
||
deadline := time.After(15 * time.Second)
|
||
for {
|
||
select {
|
||
case b := <-c.send:
|
||
var ev wsEvent
|
||
if err := json.Unmarshal(b, &ev); err != nil {
|
||
t.Fatalf("解码帧 %s: %v", b, err)
|
||
}
|
||
if ev.Type == typ {
|
||
return ev
|
||
}
|
||
case <-deadline:
|
||
t.Fatalf("等待 %q 帧超时", typ)
|
||
return wsEvent{}
|
||
}
|
||
}
|
||
}
|
||
|
||
// frameData 返回一帧的载荷 map（形状不对直接判失败）。
|
||
func frameData(t *testing.T, ev wsEvent) map[string]any {
|
||
t.Helper()
|
||
data, ok := ev.Data.(map[string]any)
|
||
if !ok {
|
||
t.Fatalf("帧载荷 = %T, want map[string]any（frame=%+v）", ev.Data, ev)
|
||
}
|
||
return data
|
||
}
|
||
|
||
// voiceInputFrame 构造一条 voice_input 上行帧。text 非空时它覆盖 STT（测试因此
|
||
// 不依赖假 STT 的内容）；为空时帧里带一段假音频，走真实的解码 + STT 分支。
|
||
func voiceInputFrame(text, sessionID, deviceID string) wsInbound {
|
||
data := map[string]any{
|
||
"text": text,
|
||
"format": "wav",
|
||
"sessionId": sessionID,
|
||
"deviceId": deviceID,
|
||
}
|
||
if text == "" {
|
||
data["audio"] = base64.StdEncoding.EncodeToString([]byte("fake-audio"))
|
||
}
|
||
raw, err := json.Marshal(data)
|
||
if err != nil {
|
||
panic(err)
|
||
}
|
||
return wsInbound{Type: "voice_input", Data: raw}
|
||
}
|
||
|
||
// TestVoiceTurnRepliesWithAudioAndIdentity 钉住 C1 的音频收尾与 C3 的归属字段：
|
||
// 一条语音回合（走编排器）结算后，定向回一条 voice_output，带助手正文合成的
|
||
// 音频、识别出的转写，以及 sessionId/deviceId；有真实音频时**不带** provider
|
||
// 降级信号。
|
||
func TestVoiceTurnRepliesWithAudioAndIdentity(t *testing.T) {
|
||
t.Parallel()
|
||
hb := newOrchestratedTestHub(t, fixedResult("我在听"), nil)
|
||
hb.voice = &signalVoice{text: "你好昔涟", audio: voice.Audio{Format: "wav", Data: []byte("audio-bytes")}, rung: "cloud"}
|
||
c := newOwnerConn(hb, 64)
|
||
hb.register(c)
|
||
|
||
c.route(voiceInputFrame("", "web", "android:pixel:7a"))
|
||
data := frameData(t, waitFrame(t, c, wsVoiceOutput))
|
||
if data["text"] != "你好昔涟" {
|
||
t.Fatalf("voice_output.text = %v, want 识别出的转写", data["text"])
|
||
}
|
||
if data["reply"] != "我在听" {
|
||
t.Fatalf("voice_output.reply = %v, want 助手正文", data["reply"])
|
||
}
|
||
if want := base64.StdEncoding.EncodeToString([]byte("audio-bytes")); data["audio"] != want {
|
||
t.Fatalf("voice_output.audio = %v, want 合成音频的 base64", data["audio"])
|
||
}
|
||
if data["sessionId"] != "web" || data["deviceId"] != "android:pixel:7a" {
|
||
t.Fatalf("voice_output 缺归属字段：%+v（协同 C3）", data)
|
||
}
|
||
if _, present := data["provider"]; present {
|
||
t.Fatalf("有真实音频时不许带降级信号：%+v", data)
|
||
}
|
||
}
|
||
|
||
// TestVoiceTurnRepliesFrameWhenTTSError 钉住「降级不静默」：TTS 不可用时**照旧
|
||
// 回帧**（而不是只回一条错误帧让客户端等不到终点），并带既有 provider 降级信号，
|
||
// 客户端据此提示「没生成声音」。
|
||
func TestVoiceTurnRepliesFrameWhenTTSError(t *testing.T) {
|
||
t.Parallel()
|
||
hb := newOrchestratedTestHub(t, fixedResult("我在听"), nil)
|
||
hb.voice = &signalVoice{text: "你好", ttsErr: errors.New("tts down"), rung: "cloud"}
|
||
c := newOwnerConn(hb, 64)
|
||
hb.register(c)
|
||
|
||
c.route(voiceInputFrame("", "web", "android:pixel:7a"))
|
||
data := frameData(t, waitFrame(t, c, wsVoiceOutput))
|
||
if data["provider"] != "silent" {
|
||
t.Fatalf("TTS 失败时必须带降级信号：%+v", data)
|
||
}
|
||
if data["reply"] != "我在听" || data["text"] != "你好" {
|
||
t.Fatalf("TTS 失败也要把转写与正文带回：%+v", data)
|
||
}
|
||
if data["audio"] != "" {
|
||
t.Fatalf("TTS 失败不该给出音频：%v", data["audio"])
|
||
}
|
||
if data["sessionId"] != "web" || data["deviceId"] != "android:pixel:7a" {
|
||
t.Fatalf("voice_output 缺归属字段：%+v（协同 C3）", data)
|
||
}
|
||
}
|
||
|
||
// TestVoiceOutputCarriesIdentityWithoutOrchestrator 钉住未接编排器时的旧直跑
|
||
// 路径（未接线的构建 / 裸 hub）同样带 C3 的两个归属字段：两条路径的
|
||
// voice_output 契约必须一致。
|
||
func TestVoiceOutputCarriesIdentityWithoutOrchestrator(t *testing.T) {
|
||
t.Parallel()
|
||
hb := newTestHub(&stubRunner{res: &loop.Result{Text: "你好", StopReason: "stop"}}, nil)
|
||
hb.voice = &signalVoice{text: "喂", audio: voice.Audio{Format: "wav", Data: []byte("a")}, rung: "cloud"}
|
||
c := newOwnerConn(hb, 16)
|
||
hb.register(c)
|
||
|
||
c.route(voiceInputFrame("喂", "s9", "android:x"))
|
||
data := frameData(t, waitFrame(t, c, wsVoiceOutput))
|
||
if data["sessionId"] != "s9" || data["deviceId"] != "android:x" {
|
||
t.Fatalf("直跑路径的 voice_output 缺归属字段：%+v（协同 C3）", data)
|
||
}
|
||
}
|
||
|
||
// TestVoiceTurnQueuesWhenSessionBusy 钉住 C1 的忙时语义：语音回合与文本回合共用
|
||
// 编排器，因此会话忙时它照常入有界待处理队列（不插队、不丢），前一条结算后按
|
||
// **严格 FIFO** 续跑；两条都落库。
|
||
func TestVoiceTurnQueuesWhenSessionBusy(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(4)
|
||
hist := &fakeHistory{}
|
||
hb := newOrchestratedTestHub(t, run.Run, hist)
|
||
hb.voice = &signalVoice{audio: voice.Audio{Format: "wav", Data: []byte("a")}, rung: "cloud"}
|
||
c := newOwnerConn(hb, 64)
|
||
hb.register(c)
|
||
|
||
c.route(voiceInputFrame("第一次", "web", "dev"))
|
||
if !waitBusy(hb.orchestrator, "web") {
|
||
t.Fatal("会话一直没有忙碌起来")
|
||
}
|
||
// 第二条在会话忙时到达：按编排器规则排队，而不是新起一个回合或丢掉。
|
||
c.route(voiceInputFrame("第二次", "web", "dev"))
|
||
if !waitQueueDepth(hb.orchestrator, "web", 1) {
|
||
t.Fatalf("队列深度 = %d, want 1", hb.orchestrator.QueueDepth("web"))
|
||
}
|
||
|
||
// 放行第一条：只有它能结算（第二条还堵在围栏之后），因此这一帧确定属于
|
||
// 先到的那条 —— FIFO 不是靠时序运气，而是靠门控。
|
||
run.tickets <- struct{}{}
|
||
first := frameData(t, waitFrame(t, c, wsVoiceOutput))
|
||
if first["text"] != "第一次" {
|
||
t.Fatalf("先到的语音回合 = %v, want 第一次（不得插队）", first["text"])
|
||
}
|
||
run.tickets <- struct{}{}
|
||
second := frameData(t, waitFrame(t, c, wsVoiceOutput))
|
||
if second["text"] != "第二次" {
|
||
t.Fatalf("续跑的语音回合 = %v, want 第二次", second["text"])
|
||
}
|
||
if !waitIdle(hb.orchestrator, "web") {
|
||
t.Fatalf("会话一直没有回到空闲：busy=%v depth=%d", hb.orchestrator.Busy("web"), hb.orchestrator.QueueDepth("web"))
|
||
}
|
||
|
||
users := userContents(hist.records())
|
||
if len(users) != 2 || users[0] != "第一次" || users[1] != "第二次" {
|
||
t.Fatalf("落库的用户消息 = %v, want [第一次 第二次]（语音问过的事必须进历史、且一条不丢）", users)
|
||
}
|
||
}
|
||
|
||
// TestVoiceSinkFollowsSteeredVoiceMessage 钉住转向注入这条路径（design/15 §2.4）：
|
||
// 忙时入 pending 的语音消息若在回合内边界被注入**同一个回合**（注入＝已消费），
|
||
// 它的音频必须由那个回合结算时一并送回 —— 那条请求的回合已被接手，若不跟着走，
|
||
// 它会永远等不到 voice_output（音频回复不能丢）。
|
||
//
|
||
// 本测试直接驱动认领那一步（consumePending），因为真正的触发点在 loop 的边界回调上，
|
||
// 而这里要钉的是网关侧的接线：注入时挂上收尾回调、结算时把它派发出去。
|
||
func TestVoiceSinkFollowsSteeredVoiceMessage(t *testing.T) {
|
||
t.Parallel()
|
||
run := newGatedRunner(4)
|
||
hb := newOrchestratedTestHub(t, run.Run, nil)
|
||
hb.voice = &signalVoice{audio: voice.Audio{Format: "wav", Data: []byte("a")}, rung: "cloud"}
|
||
c := newOwnerConn(hb, 64)
|
||
hb.register(c)
|
||
|
||
// 一个普通（非语音）回合在飞，语音请求在它忙时排队。
|
||
if sr := hb.orchestrator.Submit(context.Background(), "web", []model.Message{userMsg("文本问题")}, loop.OwnerEnv()); !sr.Immediate {
|
||
t.Fatalf("文本回合 = %+v, want Immediate", sr)
|
||
}
|
||
if !waitBusy(hb.orchestrator, "web") {
|
||
t.Fatal("会话一直没有忙碌起来")
|
||
}
|
||
c.route(voiceInputFrame("语音问题", "web", "dev"))
|
||
if !waitQueueDepth(hb.orchestrator, "web", 1) {
|
||
t.Fatalf("队列深度 = %d, want 1", hb.orchestrator.QueueDepth("web"))
|
||
}
|
||
|
||
// 模拟 loop 在回合内边界认领那条 pending 的语音消息（注入＝已消费）。
|
||
o := hb.orchestrator
|
||
o.mu.Lock()
|
||
st := o.states["web"]
|
||
gen, seq := st.gen, st.pending[0].seq
|
||
o.mu.Unlock()
|
||
if !o.consumePending("web", gen, map[uint64]bool{seq: true}) {
|
||
t.Fatal("认领失败：pending 里没有那条语音消息")
|
||
}
|
||
|
||
// 放行那个回合：答案就在它的回复里，因此音频必须回给语音连接。
|
||
run.tickets <- struct{}{}
|
||
data := frameData(t, waitFrame(t, c, wsVoiceOutput))
|
||
if data["text"] != "语音问题" {
|
||
t.Fatalf("注入后回的音频属于 %v, want 语音问题", data["text"])
|
||
}
|
||
}
|
||
|
||
// userContents 返回已落库记录里的用户消息正文（按顺序）。
|
||
func userContents(msgs []history.Message) []string {
|
||
out := make([]string, 0, len(msgs))
|
||
for _, m := range msgs {
|
||
if m.Role == history.RoleUser {
|
||
out = append(out, m.Content)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// fixedResult 返回一个总是产出同一助手正文的回合驱动器。
|
||
func fixedResult(text string) TurnRunner {
|
||
return func(context.Context, string, []model.Message, loop.SessionEnv) (*loop.Result, error) {
|
||
return &loop.Result{Text: text, Steps: 1, StopReason: "stop"}, nil
|
||
}
|
||
}
|
||
|
||
// TestVoiceTurnPersistsIntoHistoryEndToEnd 是 C1 的端到端回归（HTTP/WS 层 +
|
||
// 真实 history 存储）：一条 voice_input 之后，用户消息（转写）与助手回复都真的
|
||
// 落库，用户消息带发送端的 deviceId 与平台标签，且历史接口查得到 —— 这正是
|
||
// 「语音问过的事在 Web 上必须看得到」。
|
||
func TestVoiceTurnPersistsIntoHistoryEndToEnd(t *testing.T) {
|
||
t.Parallel()
|
||
dataDir := t.TempDir()
|
||
jwt := auth.New(config.AuthConfig{JWTSecret: testSecret})
|
||
cfg := &config.Config{Auth: config.AuthConfig{AdminUser: "admin", AdminPass: "secret", JWTSecret: testSecret}}
|
||
st := newGatewayHistoryStore(t)
|
||
srv, err := New(cfg, Deps{Auth: jwt, DataDir: dataDir, Loop: &stubRunner{res: &loop.Result{Text: "语音答复", StopReason: "stop"}}, History: st})
|
||
if err != nil {
|
||
t.Fatalf("构造网关失败: %v", err)
|
||
}
|
||
srv.SetVoice(&signalVoice{text: "语音提问", audio: voice.Audio{Format: "wav", Data: []byte("a")}, rung: "cloud"})
|
||
ts := httptest.NewServer(srv.Handler())
|
||
t.Cleanup(ts.Close)
|
||
t.Cleanup(srv.Close)
|
||
tok, err := jwt.Sign("acct", auth.RoleOwner, time.Hour)
|
||
if err != nil {
|
||
t.Fatalf("签发 token 失败: %v", err)
|
||
}
|
||
|
||
conn := dialWS(t, ts.URL, tok)
|
||
t.Cleanup(func() { _ = conn.Close() })
|
||
if err := conn.WriteJSON(map[string]any{"type": "voice_input", "data": map[string]any{
|
||
"audio": base64.StdEncoding.EncodeToString([]byte("raw-audio")),
|
||
"format": "wav",
|
||
"sessionId": history.MainSessionID,
|
||
"deviceId": "android:pixel:7a",
|
||
}}); err != nil {
|
||
t.Fatalf("发送 voice_input 失败: %v", err)
|
||
}
|
||
if ev := readWSType(t, conn, wsVoiceOutput); ev.Type != wsVoiceOutput {
|
||
t.Fatalf("帧类型 = %q", ev.Type)
|
||
}
|
||
|
||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok)
|
||
if resp.StatusCode != http.StatusOK {
|
||
t.Fatalf("历史接口 status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||
}
|
||
var envelope struct {
|
||
Data []struct {
|
||
Role string `json:"role"`
|
||
Content string `json:"content"`
|
||
Platform string `json:"platform"`
|
||
DeviceID string `json:"deviceId"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(body, &envelope); err != nil {
|
||
t.Fatalf("解析历史响应失败: %v (body=%s)", err, body)
|
||
}
|
||
var userOK, assistantOK bool
|
||
for _, m := range envelope.Data {
|
||
switch {
|
||
case m.Role == history.RoleUser && m.Content == "语音提问":
|
||
userOK = true
|
||
if m.DeviceID != "android:pixel:7a" {
|
||
t.Fatalf("用户消息 deviceId = %q, want android:pixel:7a（语音特有身份必须落库）", m.DeviceID)
|
||
}
|
||
if m.Platform != history.PlatformWeb {
|
||
t.Fatalf("用户消息 platform = %q, want %q", m.Platform, history.PlatformWeb)
|
||
}
|
||
case m.Role == history.RoleAssistant && m.Content == storedParagraph("语音答复"):
|
||
assistantOK = true
|
||
}
|
||
}
|
||
if !userOK || !assistantOK {
|
||
t.Fatalf("历史里缺语音回合：user=%v assistant=%v (msgs=%+v)", userOK, assistantOK, envelope.Data)
|
||
}
|
||
}
|
||
|
||
// readWSType 从真连接上读帧直到出现给定类型。
|
||
func readWSType(t *testing.T, conn *websocket.Conn, typ string) wsEvent {
|
||
t.Helper()
|
||
_ = conn.SetReadDeadline(time.Now().Add(5 * time.Second))
|
||
for {
|
||
var ev wsEvent
|
||
if err := conn.ReadJSON(&ev); err != nil {
|
||
t.Fatalf("读 WS 帧失败（等 %q）: %v", typ, err)
|
||
}
|
||
if ev.Type == typ {
|
||
return ev
|
||
}
|
||
}
|
||
}
|