AskaEth
3a01b4e236
- G5：pending_full 增 clientMsgId/messageId（仅在可给出时出现、绝不新铸 id —— 队列满的拒绝发生在幂等键认领之前，代铸会让客户端重试被判重复而永远进不来）；另定向回发送者 {type:error,data:{code:chat_busy,...}}（与 REST 409 chat_busy 同码），广播帧照旧带同样的键供其它端对齐。
- C1：语音回合改走与文本同一条路径（编排器 Submit → 落库 → 流式 → 跨端同步），deviceId 走既有 withDeviceID、platform 走存储默认 web；未接编排器时保留旧直连路径。音频在回合结算处（persistAssistantReply → persistDocumentStatus → dispatchVoiceReply）合成并定向回原连接，回调各自 goroutine（TTS 延迟不拖住 pending_drain/session_idle）；续跑 FIFO 与转向注入两条路径都带走回调 ⇒ 忙时排队/被注入的语音也不丢音频。
- C3：voice_output 增 sessionId/deviceId；三处主动加码（按最优方向）：① voice_input 增可选 clientMsgId（语音也有幂等）② 不再拿 stopReason 顶替空正文当语音内容（空正文走 provider 降级信号）③ TTS 失败照旧回帧 + provider:silent（不静默）。
- 拆 chat_documents_test.go（510 行越界）→ 373 + 160，只搬不改（拆前后同 -run 过滤 PASS 集合逐条一致）。
- 主会话独立验收：隔离 worktree（HEAD + 本批次 11 文件）build/vet 绿、gateway -race 绿、全树 59 包（唯一红是既有负载 flake orchestration/TestSpawnTimeoutFromToolDefault）、linux+darwin 交叉编译绿；变异独立复现（pending_full 摘掉 clientMsgId ⇒ 3 条红）。
- 待办（已登记）：Docs/api-client.md 与 api-reference.md 的帧/字段登记随 v0.59 一并落笔（本批次刻意未动 Docs，避免与契约批次打架）。
353 lines
14 KiB
Go
353 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 排空连接上的帧直到出现给定类型（语音回合的收尾帧会排在若干状态
|
||
// 信号广播之后），超时即失败。
|
||
func waitFrame(t *testing.T, c *wsConn, typ string) wsEvent {
|
||
t.Helper()
|
||
deadline := time.After(3 * 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
|
||
}
|
||
}
|
||
}
|