Cyrene_v2/internal/services/gateway/voice_turn_test.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 == "语音答复":
			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
		}
	}
}