From 3a01b4e2365b239f722b7e7c88aa8604c928b155 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 17:59:28 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20=E8=AF=AD=E9=9F=B3=E5=9B=9E?= =?UTF-8?q?=E5=90=88=E8=BF=9B=E7=BC=96=E6=8E=92+=E8=90=BD=E5=BA=93?= =?UTF-8?q?=E3=80=81G5=20=E9=98=9F=E5=88=97=E6=BB=A1=E5=B8=A6=E5=9B=9E?= =?UTF-8?q?=E8=BA=AB=E4=BB=BD=E3=80=81voice=5Foutput=20=E5=BD=92=E5=B1=9E?= =?UTF-8?q?=EF=BC=88=E5=AE=89=E5=8D=93=20G5/C1/C3=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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，避免与契约批次打架）。 --- .../gateway/chat_documents_bounds_test.go | 160 ++++++++ .../services/gateway/chat_documents_test.go | 137 ------- .../services/gateway/session_orchestrator.go | 20 +- .../gateway/session_pending_full_test.go | 238 ++++++++++++ internal/services/gateway/session_reject.go | 86 +++++ internal/services/gateway/session_steer.go | 3 + internal/services/gateway/session_voice.go | 111 ++++++ internal/services/gateway/voice.go | 140 +++++-- internal/services/gateway/voice_turn_test.go | 352 ++++++++++++++++++ internal/services/gateway/ws_hub_conn.go | 13 + internal/services/gateway/ws_route.go | 33 +- 11 files changed, 1123 insertions(+), 170 deletions(-) create mode 100644 internal/services/gateway/chat_documents_bounds_test.go create mode 100644 internal/services/gateway/session_pending_full_test.go create mode 100644 internal/services/gateway/session_reject.go create mode 100644 internal/services/gateway/session_voice.go create mode 100644 internal/services/gateway/voice_turn_test.go diff --git a/internal/services/gateway/chat_documents_bounds_test.go b/internal/services/gateway/chat_documents_bounds_test.go new file mode 100644 index 0000000..ec367fc --- /dev/null +++ b/internal/services/gateway/chat_documents_bounds_test.go @@ -0,0 +1,160 @@ +package gateway + +// 本文件锁定切片 S3 的 B 路中**上限与边界**那一半（design/24 §4.2 / §6.1 / §9.5）： +// - `stream:true` 与 WS 帧两条转发路径都要把 documents 原样带走（漏抄字段即静默 +// 丢附件）； +// - 历史重建必须带回文档引用（S2 移交的第一处「静默吞掉」）； +// - 「只带附件」的回合不算空回合（第二处「静默吞掉」）； +// - 单回合附件上限的值本身钉在设计定稿上。 +// +// 它从 chat_documents_test.go 拆出，使两个文件都留在「单文件 ≤500 行」的硬约束内； +// 两个文件是**同一批断言的搬家**，语义一字未改（夹具仍留在原文件，同包可见）。 +// +// 断言一律走哨兵错误 + errors.Is / errors.As（机器可读的 code），不按错误文本 +// 子串断言 —— 文案会变，契约不变。 + +import ( + "context" + "net/http" + "testing" + + "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" +) + +// TestChatDocumentsStreamPathCarriesDocuments 钉住 `stream:true` 这条转发路径： +// 载荷必须把 documents 一并带到 hub（漏抄一个字段就是一次静默丢附件），因此 +// 后台回合里同样能看到文档部件。 +func TestChatDocumentsStreamPathCarriesDocuments(t *testing.T) { + t.Parallel() + env := newChatDocumentsEnv(t) + name := env.storeDocumentFile(t, "流式回合的附件", ".txt") + + resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{ + "sessionId": "web", + "text": "", + "documents": []chatDocumentRef{{Name: name}}, + "stream": true, + }, env.token(t)) + if resp.StatusCode != http.StatusOK { + t.Fatalf("stream:true 未被受理：status = %d (body=%s)", resp.StatusCode, body) + } + calls := env.runner.waitCalls(t, 1) + if docs := chatDocumentsOf(calls[0]); len(docs) != 1 || docs[0].DocumentName != name { + t.Fatalf("流式路径丢掉了文档部件：%+v", calls[0]) + } +} + +// TestHubStreamChatResolvesDocuments 钉住 WS 帧（`chatStreamRequest` 与 REST +// 同形）：带了 documents 就按同一条解析路径装配；未接解析器时**明确回一条 +// error 帧**并中止本回合，绝不静默吞掉附件。 +func TestHubStreamChatResolvesDocuments(t *testing.T) { + t.Parallel() + env := newChatDocumentsEnv(t) + name := env.storeDocumentFile(t, "ws 帧的附件", ".txt") + payload := map[string]any{"sessionId": "web", "text": "看这个", "documents": []chatDocumentRef{{Name: name}}} + + hb := newTestHub(env.runner, nil) + hb.documents = env.srv.resolveChatDocuments + hb.streamChat(context.Background(), payload) + calls := env.runner.snapshot() + if len(calls) != 1 || len(chatDocumentsOf(calls[0])) != 1 { + t.Fatalf("WS 帧路径丢掉了文档部件：%+v", calls) + } + + // 未接解析器（裸 hub）：文档必须可见地失败，而不是被丢掉。 + bare := newTestHub(env.runner, nil) + c := newOwnerConn(bare, 8) + bare.register(c) + bare.streamChat(context.Background(), payload) + ev := readFakeFrame(t, c) + if ev.Type != "error" { + t.Fatalf("帧类型 = %q, want error（附件解析失败必须可见）", ev.Type) + } + data, ok := ev.Data.(map[string]any) + if !ok || data["code"] != documentsUnavailableCode { + t.Fatalf("error data = %+v, want code=%q", ev.Data, documentsUnavailableCode) + } + if len(env.runner.snapshot()) != 1 { + t.Fatalf("解析失败的回合仍然起了 loop —— 必须在起回合之前拦住") + } +} + +// TestTranscriptMessageKeepsDocumentReference 是 design/24 §9.5 第一处的回归锁： +// 从历史重建模型消息时必须带回文档引用（否则后续回合里附件凭空消失）。 +func TestTranscriptMessageKeepsDocumentReference(t *testing.T) { + t.Parallel() + const name = "0123456789abcdef0123456789abcdef.txt" + raw, err := history.EncodeDocuments([]history.DocumentRef{{ + Name: name, MIME: documentSniffOKMIME, Size: 12345, Status: history.DocumentStatusExtracted, + }}) + if err != nil { + t.Fatalf("encode document: %v", err) + } + msg := transcriptMessage(history.Message{ID: "m1", Role: history.RoleUser, Content: "读一下", Document: raw}) + docs := chatDocumentsOf([]model.Message{msg}) + if len(docs) != 1 { + t.Fatalf("文档引用没能重建回来：%+v", msg) + } + if docs[0].DocumentName != name || docs[0].DocumentSize != 12345 || + docs[0].DocumentURL != documentFilePrefix+name || docs[0].DocumentMIME != documentSniffOKMIME { + t.Fatalf("重建出的文档部件 = %+v, want name/mime/url/size 与落库引用一致", docs[0]) + } + // 没有附件的消息照旧没有文档部件；坏列不 panic 也不产生部件（打 WARN）。 + if got := chatDocumentsOf([]model.Message{ + transcriptMessage(history.Message{ID: "m2", Role: history.RoleUser, Content: "无附件"}), + }); len(got) != 0 { + t.Fatalf("无附件消息产生了文档部件：%+v", got) + } + if got := chatDocumentsOf([]model.Message{ + transcriptMessage(history.Message{ID: "m3", Role: history.RoleUser, Content: "坏列", Document: "{not json"}), + }); len(got) != 0 { + t.Fatalf("坏列产生了文档部件：%+v", got) + } + // 助手回合的说话者标注不受本次改动影响；附件不会落到非用户回合上。 + assistant := transcriptMessage(history.Message{ + ID: "m4", Role: history.RoleAssistant, Content: "答复", PersonaID: "cyrene", Document: raw, + }) + if assistant.SenderName != "cyrene" { + t.Fatalf("助手消息的 SenderName = %q, want cyrene", assistant.SenderName) + } + if got := chatDocumentsOf([]model.Message{assistant}); len(got) != 0 { + t.Fatalf("文档部件落到了助手回合上：%+v", got) + } +} + +// TestMessagesEmptyTreatsDocumentAsContent 是 design/24 §9.5 第二处的回归锁： +// 「只带附件」的回合必须算非空，否则纯文档消息会被判成空洞回合而整轮丢弃。 +func TestMessagesEmptyTreatsDocumentAsContent(t *testing.T) { + t.Parallel() + o := NewOrchestrator(OrchestratorOptions{ + Run: func(context.Context, string, []model.Message, loop.SessionEnv) (*loop.Result, error) { + return &loop.Result{Text: "ok", StopReason: "stop"}, nil + }, + }) + docOnly := chatTurnMessages("", "", []model.Document{{Name: "0123456789abcdef0123456789abcdef.txt"}}) + if o.messagesEmpty(docOnly) { + t.Fatalf("纯附件回合被判成空回合：%+v", docOnly) + } + if sr := o.Submit(context.Background(), "web", docOnly, loop.SessionEnv{}); sr.Empty || !sr.Accepted { + t.Fatalf("纯附件回合的判定 = %+v, want 被接收且非 Empty", sr) + } + // 反向：真·空回合仍然被判空（本次改动不得把空判定一起放宽）。 + if !o.messagesEmpty([]model.Message{model.TextMessage(model.RoleUser, "")}) { + t.Fatal("空回合未被判空") + } +} + +// TestMaxChatDocumentsMatchesDesign 单独把「单回合附件数上限的值」钉在设计定稿上。 +// +// 为什么必须单独钉一次：TestChatDocumentsTooManyRejected 是**用本常量算输入条数**的 +// （`make(..., maxChatDocuments+1)`、`for i < maxChatDocuments`），所以把常量改大它照样 +// 绿 —— 实测把 5 改成 1000 时整个包仍全绿。也就是说「上限被悄悄改宽」这件事，除了这里 +// 没有任何用例会发现。design/24 §9.4-3 的定稿值是 5。 +func TestMaxChatDocumentsMatchesDesign(t *testing.T) { + t.Parallel() + if maxChatDocuments != 5 { + t.Fatalf("单回合附件上限 = %d, want 5（design/24 §9.4-3 的定稿值）", maxChatDocuments) + } +} diff --git a/internal/services/gateway/chat_documents_test.go b/internal/services/gateway/chat_documents_test.go index f8842b7..c304310 100644 --- a/internal/services/gateway/chat_documents_test.go +++ b/internal/services/gateway/chat_documents_test.go @@ -28,7 +28,6 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" - "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" ) @@ -372,139 +371,3 @@ func TestChatDocumentsNeverFetchURLs(t *testing.T) { t.Fatalf("被拒的请求起了回合") } } - -// TestChatDocumentsStreamPathCarriesDocuments 钉住 `stream:true` 这条转发路径： -// 载荷必须把 documents 一并带到 hub（漏抄一个字段就是一次静默丢附件），因此 -// 后台回合里同样能看到文档部件。 -func TestChatDocumentsStreamPathCarriesDocuments(t *testing.T) { - t.Parallel() - env := newChatDocumentsEnv(t) - name := env.storeDocumentFile(t, "流式回合的附件", ".txt") - - resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{ - "sessionId": "web", - "text": "", - "documents": []chatDocumentRef{{Name: name}}, - "stream": true, - }, env.token(t)) - if resp.StatusCode != http.StatusOK { - t.Fatalf("stream:true 未被受理：status = %d (body=%s)", resp.StatusCode, body) - } - calls := env.runner.waitCalls(t, 1) - if docs := chatDocumentsOf(calls[0]); len(docs) != 1 || docs[0].DocumentName != name { - t.Fatalf("流式路径丢掉了文档部件：%+v", calls[0]) - } -} - -// TestHubStreamChatResolvesDocuments 钉住 WS 帧（`chatStreamRequest` 与 REST -// 同形）：带了 documents 就按同一条解析路径装配；未接解析器时**明确回一条 -// error 帧**并中止本回合，绝不静默吞掉附件。 -func TestHubStreamChatResolvesDocuments(t *testing.T) { - t.Parallel() - env := newChatDocumentsEnv(t) - name := env.storeDocumentFile(t, "ws 帧的附件", ".txt") - payload := map[string]any{"sessionId": "web", "text": "看这个", "documents": []chatDocumentRef{{Name: name}}} - - hb := newTestHub(env.runner, nil) - hb.documents = env.srv.resolveChatDocuments - hb.streamChat(context.Background(), payload) - calls := env.runner.snapshot() - if len(calls) != 1 || len(chatDocumentsOf(calls[0])) != 1 { - t.Fatalf("WS 帧路径丢掉了文档部件：%+v", calls) - } - - // 未接解析器（裸 hub）：文档必须可见地失败，而不是被丢掉。 - bare := newTestHub(env.runner, nil) - c := newOwnerConn(bare, 8) - bare.register(c) - bare.streamChat(context.Background(), payload) - ev := readFakeFrame(t, c) - if ev.Type != "error" { - t.Fatalf("帧类型 = %q, want error（附件解析失败必须可见）", ev.Type) - } - data, ok := ev.Data.(map[string]any) - if !ok || data["code"] != documentsUnavailableCode { - t.Fatalf("error data = %+v, want code=%q", ev.Data, documentsUnavailableCode) - } - if len(env.runner.snapshot()) != 1 { - t.Fatalf("解析失败的回合仍然起了 loop —— 必须在起回合之前拦住") - } -} - -// TestTranscriptMessageKeepsDocumentReference 是 design/24 §9.5 第一处的回归锁： -// 从历史重建模型消息时必须带回文档引用（否则后续回合里附件凭空消失）。 -func TestTranscriptMessageKeepsDocumentReference(t *testing.T) { - t.Parallel() - const name = "0123456789abcdef0123456789abcdef.txt" - raw, err := history.EncodeDocuments([]history.DocumentRef{{ - Name: name, MIME: documentSniffOKMIME, Size: 12345, Status: history.DocumentStatusExtracted, - }}) - if err != nil { - t.Fatalf("encode document: %v", err) - } - msg := transcriptMessage(history.Message{ID: "m1", Role: history.RoleUser, Content: "读一下", Document: raw}) - docs := chatDocumentsOf([]model.Message{msg}) - if len(docs) != 1 { - t.Fatalf("文档引用没能重建回来：%+v", msg) - } - if docs[0].DocumentName != name || docs[0].DocumentSize != 12345 || - docs[0].DocumentURL != documentFilePrefix+name || docs[0].DocumentMIME != documentSniffOKMIME { - t.Fatalf("重建出的文档部件 = %+v, want name/mime/url/size 与落库引用一致", docs[0]) - } - // 没有附件的消息照旧没有文档部件；坏列不 panic 也不产生部件（打 WARN）。 - if got := chatDocumentsOf([]model.Message{ - transcriptMessage(history.Message{ID: "m2", Role: history.RoleUser, Content: "无附件"}), - }); len(got) != 0 { - t.Fatalf("无附件消息产生了文档部件：%+v", got) - } - if got := chatDocumentsOf([]model.Message{ - transcriptMessage(history.Message{ID: "m3", Role: history.RoleUser, Content: "坏列", Document: "{not json"}), - }); len(got) != 0 { - t.Fatalf("坏列产生了文档部件：%+v", got) - } - // 助手回合的说话者标注不受本次改动影响；附件不会落到非用户回合上。 - assistant := transcriptMessage(history.Message{ - ID: "m4", Role: history.RoleAssistant, Content: "答复", PersonaID: "cyrene", Document: raw, - }) - if assistant.SenderName != "cyrene" { - t.Fatalf("助手消息的 SenderName = %q, want cyrene", assistant.SenderName) - } - if got := chatDocumentsOf([]model.Message{assistant}); len(got) != 0 { - t.Fatalf("文档部件落到了助手回合上：%+v", got) - } -} - -// TestMessagesEmptyTreatsDocumentAsContent 是 design/24 §9.5 第二处的回归锁： -// 「只带附件」的回合必须算非空，否则纯文档消息会被判成空洞回合而整轮丢弃。 -func TestMessagesEmptyTreatsDocumentAsContent(t *testing.T) { - t.Parallel() - o := NewOrchestrator(OrchestratorOptions{ - Run: func(context.Context, string, []model.Message, loop.SessionEnv) (*loop.Result, error) { - return &loop.Result{Text: "ok", StopReason: "stop"}, nil - }, - }) - docOnly := chatTurnMessages("", "", []model.Document{{Name: "0123456789abcdef0123456789abcdef.txt"}}) - if o.messagesEmpty(docOnly) { - t.Fatalf("纯附件回合被判成空回合：%+v", docOnly) - } - if sr := o.Submit(context.Background(), "web", docOnly, loop.SessionEnv{}); sr.Empty || !sr.Accepted { - t.Fatalf("纯附件回合的判定 = %+v, want 被接收且非 Empty", sr) - } - // 反向：真·空回合仍然被判空（本次改动不得把空判定一起放宽）。 - if !o.messagesEmpty([]model.Message{model.TextMessage(model.RoleUser, "")}) { - t.Fatal("空回合未被判空") - } -} - -// TestMaxChatDocumentsMatchesDesign 单独把「单回合附件数上限的值」钉在设计定稿上。 -// -// 为什么必须单独钉一次：TestChatDocumentsTooManyRejected 是**用本常量算输入条数**的 -// （`make(..., maxChatDocuments+1)`、`for i < maxChatDocuments`），所以把常量改大它照样 -// 绿 —— 实测把 5 改成 1000 时整个包仍全绿。也就是说「上限被悄悄改宽」这件事，除了这里 -// 没有任何用例会发现。design/24 §9.4-3 的定稿值是 5。 -func TestMaxChatDocumentsMatchesDesign(t *testing.T) { - t.Parallel() - if maxChatDocuments != 5 { - t.Fatalf("单回合附件上限 = %d, want 5（design/24 §9.4-3 的定稿值）", maxChatDocuments) - } -} diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index c4c3b38..560dd1b 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -146,6 +146,9 @@ type sessionState struct { // 结算时把抽取结论回写进那一列（design/24 §6.1）。它按代次配对，因此被中断/ // 取代的陈旧回合读不到本回合的条目。见 session_document_status.go。 turnDocs turnDocuments + // voiceTurn 是本回合的语音收尾回调（协同 C1），按代次配对；纯文本回合记为 + // 零值。见 session_voice.go。 + voiceTurn voiceTurn } // pendingMsg 是一条已排队用户回合。它携带自己的 env，使续跑 @@ -169,6 +172,9 @@ type pendingMsg struct { // 文档抽取结论（design/24 §6.1）：它在入队那一刻就落了库，而那时抽取还没 // 发生。见 session_document_status.go。 persisted []history.Message + // voice 是这条待处理回合的语音收尾回调（协同 C1）：它源自 voice_input 时 + // 非 nil，使续跑回合结算后仍能把音频定向回发起它的那根连接。 + voice VoiceReplyFunc } // Orchestrator 拥有每会话的消息状态机。它对并发安全 @@ -257,8 +263,9 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes // 状态事件在这里解锁之后再发。 a := o.admitLocked(ctx, st, sessionID) if a.full { + rejected := o.rejectedIdentityLocked(ctx, st, sessionID) o.mu.Unlock() - o.emitState(sessionID, EventPendingFull, map[string]any{"sessionId": sessionID, "mode": a.mode, "max": o.maxPending}) + o.emitPendingFull(ctx, sessionID, a.mode, rejected) return SubmitResult{PendingFull: true}, nil } if a.dup { @@ -272,7 +279,7 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes if st.mode != ModeIdle { st.pendingSeq++ seq := st.pendingSeq - st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now(), seq: seq}) + st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now(), seq: seq, voice: voiceReplySinkFrom(ctx)}) depth := len(st.pending) mode := st.mode o.mu.Unlock() @@ -305,6 +312,8 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes // 且落库只写**引用** —— 抽取还发生在后面，结论由回合结算时补进同一列 // （两段式见 session_document_status.go）。 o.rememberTurnDocuments(sessionID, gen, o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID)) + // 语音收尾回调同样按代次配对（协同 C1，见 session_voice.go）。 + o.rememberVoiceTurn(sessionID, gen, voiceReplySinkFrom(ctx)) if sync { return o.runImmediateSync(runCtx, sessionID, gen, turn, env, fence, deviceID, msgMode, a.hint.MessageID) @@ -357,10 +366,12 @@ func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Mes // // 它同时是文档抽取结论回写的落点（design/24 §6.1）：本回合用户消息的 // document 列里此前只有引用，装配期解析出的 status/chars/truncated 在这里补上 -// —— 见 session_document_status.go 的两段式说明。 +// —— 见 session_document_status.go；语音回合的音频回复同样在此收尾（协同 C1， +// 见 session_voice.go）：只有这里知道这个回合真正产出了什么。 func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID, mode, personaID string) error { o.persistAssistantReply(sessionID, res, deviceID, mode, personaID) o.persistDocumentStatus(sessionID, gen, res) + o.dispatchVoiceReply(sessionID, gen, res) o.mu.Lock() st := o.states[sessionID] if st == nil || st.gen != gen { @@ -389,6 +400,9 @@ func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Resu // 续跑的这一轮同样有「回合前已落库的用户消息」，因此把它们的文档状态 // 记到新代次上（design/24 §6.1）。 st.turnDocs = turnDocuments{gen: gen, msgs: next.persisted} + // 续跑的这条若源自语音请求，它的音频回复照旧定向回发起它的那根连接 + // （协同 C1：忙时排队不许把音频丢掉）。 + st.voiceTurn = voiceTurn{gen: gen, sinks: voiceSinksOf(next.voice)} depth := len(st.pending) resume := append([]model.Message(nil), st.tail...) env := next.env diff --git a/internal/services/gateway/session_pending_full_test.go b/internal/services/gateway/session_pending_full_test.go new file mode 100644 index 0000000..db13fbe --- /dev/null +++ b/internal/services/gateway/session_pending_full_test.go @@ -0,0 +1,238 @@ +package gateway + +import ( + "context" + "encoding/json" + "log/slog" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// 本文件是协同项 G5 的回归：pending_full 必须说清「是哪一条被丢了」。 +// +// 修前：pending_full 只带 {sessionId, mode, max}，客户端收到「队列满」时无法判定 +// 该收掉哪一条本地乐观气泡（WS 路径也没有 REST 那 409 chat_busy 的等价物）， +// 那条气泡会永远挂在「发送中…」。 +// +// 修后：帧里带上被拒那条的 clientMsgId（未携带时省略、绝不伪造），该键此前已 +// 落库/已认领时再带上 messageId；发送者还会收到一条定向拒绝帧（type=error、 +// data.code=chat_busy，与 REST 同码）。既有三个字段一个不改。 +// +// 断言一律按字段判定，不按错误文本子串。 + +// pendingFullData 返回记录型汇聚点里最近一次 pending_full 的载荷。 +func pendingFullData(t *testing.T, sink *recordingSink) map[string]any { + t.Helper() + sink.mu.Lock() + defer sink.mu.Unlock() + for i := len(sink.events) - 1; i >= 0; i-- { + if sink.events[i].typ == EventPendingFull { + return sink.events[i].data + } + } + t.Fatal("没有收到 pending_full 事件") + return nil +} + +// fillPendingQueue 把 "s1" 推到「有一个在飞回合 + 待处理队列已满」的状态。 +func fillPendingQueue(t *testing.T, o *Orchestrator, maxPending int) { + t.Helper() + if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("A")}, loop.OwnerEnv()); !sr.Immediate { + t.Fatalf("A = %+v, want Immediate", sr) + } + for i := 0; i < maxPending; i++ { + if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("p")}, loop.OwnerEnv()); !sr.Pending { + t.Fatalf("第 %d 条排队 = %+v, want Pending", i, sr) + } + } +} + +// TestPendingFullCarriesRejectedClientMsgId 钉住主路径：队列满且新消息带 +// clientMsgId ⇒ pending_full 载荷含该键；未落库时**不得**凭空给出 messageId。 +func TestPendingFullCarriesRejectedClientMsgId(t *testing.T) { + t.Parallel() + run := newGatedRunner(8) + o, sink := seedOrchestrator(run, 1) + fillPendingQueue(t, o, 1) + + ctx := withClientMsgHint(context.Background(), clientMsgHint{ClientMsgID: "cm-42"}) + if sr := o.Submit(ctx, "s1", []model.Message{userMsg("overflow")}, loop.OwnerEnv()); !sr.PendingFull { + t.Fatalf("overflow = %+v, want PendingFull", sr) + } + data := pendingFullData(t, sink) + if data["clientMsgId"] != "cm-42" { + t.Fatalf("pending_full.clientMsgId = %v, want cm-42（客户端必须知道是哪条被丢）", data["clientMsgId"]) + } + if _, present := data["messageId"]; present { + t.Fatalf("未落库的拒绝不许伪造 messageId：%+v", data) + } + // 既有字段一个不改、不删。 + for _, key := range []string{"sessionId", "mode", "max"} { + if _, ok := data[key]; !ok { + t.Fatalf("pending_full 丢了既有字段 %q：%+v", key, data) + } + } + if data["sessionId"] != "s1" || data["max"] != 1 { + t.Fatalf("既有字段的值被改动：%+v", data) + } +} + +// TestPendingFullOmitsIdentityWhenAbsent 钉住「省略、不伪造」：老客户端不带 +// clientMsgId 时，两个新增键都必须缺席（而不是回一个空串或编造的 id）。 +func TestPendingFullOmitsIdentityWhenAbsent(t *testing.T) { + t.Parallel() + run := newGatedRunner(8) + o, sink := seedOrchestrator(run, 1) + fillPendingQueue(t, o, 1) + + if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("overflow")}, loop.OwnerEnv()); !sr.PendingFull { + t.Fatalf("overflow = %+v, want PendingFull", sr) + } + data := pendingFullData(t, sink) + if _, present := data["clientMsgId"]; present { + t.Fatalf("未携带 clientMsgId 时该键必须缺席：%+v", data) + } + if _, present := data["messageId"]; present { + t.Fatalf("未落库时 messageId 必须缺席：%+v", data) + } + if data["sessionId"] != "s1" { + t.Fatalf("既有字段必须照旧：%+v", data) + } +} + +// TestPendingFullCarriesMessageIdWhenAlreadyClaimed 钉住「已落库则带身份」：同一个 +// clientMsgId 此前已经认领并落库时，队列满的拒绝要把它已有（且终将存在）的 +// messageId 一并给出，让客户端能把那条气泡直接对齐，而不是重试一次。 +func TestPendingFullCarriesMessageIdWhenAlreadyClaimed(t *testing.T) { + t.Parallel() + run := newGatedRunner(8) + sink := &recordingSink{} + o := NewOrchestrator(OrchestratorOptions{ + Run: run.Run, + Emit: sink.Emit, + Log: slog.Default(), + MaxPending: 1, + MaxTail: 20, + History: &fakeHistory{}, + }) + ctx := withClientMsgHint(context.Background(), clientMsgHint{ClientMsgID: "cm-7"}) + first := o.Submit(ctx, "s1", []model.Message{userMsg("第一次")}, loop.OwnerEnv()) + if !first.Immediate || first.MessageID == "" { + t.Fatalf("第一次提交 = %+v, want Immediate 且带预分配 messageId", first) + } + fillPendingQueueAfterFirst(t, o, 1) + + retry := o.Submit(ctx, "s1", []model.Message{userMsg("重发同一条")}, loop.OwnerEnv()) + if !retry.PendingFull { + t.Fatalf("重发 = %+v, want PendingFull（队列满是先于判重的第一道门）", retry) + } + data := pendingFullData(t, sink) + if data["clientMsgId"] != "cm-7" { + t.Fatalf("clientMsgId = %v, want cm-7", data["clientMsgId"]) + } + if data["messageId"] != first.MessageID { + t.Fatalf("messageId = %v, want %q（该键此前已落库，必须给出已有的那条消息 id）", data["messageId"], first.MessageID) + } +} + +// fillPendingQueueAfterFirst 在已有一个在飞回合的前提下填满待处理队列。 +func fillPendingQueueAfterFirst(t *testing.T, o *Orchestrator, maxPending int) { + t.Helper() + for i := 0; i < maxPending; i++ { + if sr := o.Submit(context.Background(), "s1", []model.Message{userMsg("排队")}, loop.OwnerEnv()); !sr.Pending { + t.Fatalf("第 %d 条排队 = %+v, want Pending", i, sr) + } + } +} + +// newOrchestratedTestHub 构建一个接了编排器的 hub（回合驱动器由调用方给出），使 +// 「必须在编排器之下才成立」的行为（语音回合、队列满拒绝）能被单元级回归。 +func newOrchestratedTestHub(t *testing.T, run TurnRunner, hist HistoryStore) *hub { + t.Helper() + hb := newTestHub(&stubRunner{res: &loop.Result{Text: "ok", StopReason: "stop"}}, nil) + hb.orchestrator = NewOrchestrator(OrchestratorOptions{ + Run: run, + Emit: func(sessionID, typ string, data map[string]any) { + hb.broadcast(wsEvent{Type: typ, SessionID: sessionID, Data: data}) + }, + Log: sourcedLog(nil), + MaxTail: 20, + History: hist, + MaxPending: defaultMaxPending, + }) + return hb +} + +// TestPendingFullNotifiesSenderDirectly 钉住定向拒绝帧：队列满时，发送这条消息的 +// 连接必须收到一条 {type:"error", data:{code:"chat_busy"}}（与 REST 的 409 同码）， +// 且带上被打回那条的 clientMsgId；不带 clientMsgId 时该键缺席。 +func TestPendingFullNotifiesSenderDirectly(t *testing.T) { + t.Parallel() + run := newGatedRunner(8) + hb := newOrchestratedTestHub(t, run.Run, nil) + c := newOwnerConn(hb, 64) + hb.register(c) + + // 用编排器直接填队列：三条 route 各自起 goroutine，提交顺序不确定，而这道 + // 门本身依赖「先忙起来、再满」。 + fillPendingQueue(t, hb.orchestrator, defaultMaxPending) + if got := hb.orchestrator.QueueDepth("s1"); got != defaultMaxPending { + t.Fatalf("队列深度 = %d, want %d", got, defaultMaxPending) + } + + // 第三条经 WS 帧进来 ⇒ 它带着「定向回这条连接」的回调。 + c.route(wsInbound{Type: "chat", Data: json.RawMessage(`{"sessionId":"s1","text":"overflow","clientMsgId":"cm-9"}`)}) + full, rejected := waitFullAndRejection(t, c) + if full["clientMsgId"] != "cm-9" { + t.Fatalf("广播的 pending_full.clientMsgId = %v, want cm-9", full["clientMsgId"]) + } + if rejected["clientMsgId"] != "cm-9" || rejected["code"] != wsChatBusyCode { + t.Fatalf("定向拒绝帧 = %+v, want clientMsgId=cm-9 且 code=%q", rejected, wsChatBusyCode) + } + if rejected["sessionId"] != "s1" { + t.Fatalf("定向拒绝帧必须标明会话：%+v", rejected) + } + if _, present := rejected["messageId"]; present { + t.Fatalf("未落库的拒绝不许伪造 messageId：%+v", rejected) + } + + // 老客户端（不带 clientMsgId）：照旧收到拒绝帧，但两个身份键都必须缺席。 + c.route(wsInbound{Type: "chat", Data: json.RawMessage(`{"sessionId":"s1","text":"again"}`)}) + _, rejected = waitFullAndRejection(t, c) + if rejected["code"] != wsChatBusyCode { + t.Fatalf("定向拒绝帧 = %+v, want code=%q", rejected, wsChatBusyCode) + } + for _, key := range []string{"clientMsgId", "messageId"} { + if _, present := rejected[key]; present { + t.Fatalf("未携带身份时 %q 必须缺席：%+v", key, rejected) + } + } +} + +// waitFullAndRejection 从连接上收集一条 pending_full 广播与一条定向拒绝帧。 +func waitFullAndRejection(t *testing.T, c *wsConn) (full, rejected map[string]any) { + t.Helper() + deadline := time.After(3 * time.Second) + for full == nil || rejected == nil { + select { + case b := <-c.send: + var ev wsEvent + if err := json.Unmarshal(b, &ev); err != nil { + t.Fatalf("解码帧 %s: %v", b, err) + } + data, _ := ev.Data.(map[string]any) + switch { + case ev.Type == EventPendingFull: + full = data + case ev.Type == "error" && data["code"] == wsChatBusyCode: + rejected = data + } + case <-deadline: + t.Fatalf("等待 pending_full / 定向拒绝帧超时（full=%v rejected=%v）", full, rejected) + } + } + return full, rejected +} diff --git a/internal/services/gateway/session_reject.go b/internal/services/gateway/session_reject.go new file mode 100644 index 0000000..4122a07 --- /dev/null +++ b/internal/services/gateway/session_reject.go @@ -0,0 +1,86 @@ +package gateway + +import "context" + +// 本文件归属协同项 G5 的「队列满的拒绝要说清是哪一条」：pending_full 状态帧的 +// 载荷回填，以及发往**发送者连接**的那条定向拒绝帧。 +// +// 背景：pending_full 此前只带 {sessionId, mode, max}，于是客户端收到「队列满」 +// 时无法判定该收掉哪一条本地乐观气泡 —— 那条气泡会永远挂在「发送中…」（REST +// 路径有 409 chat_busy 可以回滚，WS 路径没有等价物）。这里给出确定的键，并让 +// 发送者收到一条能精确处理的定向帧。 +// +// 它与 ws 边界的关系：编排器不认识连接，只认识 ctx 上由传输层给出的「怎么把 +// 拒绝送回发送者」回调（与 deviceID / mode / 幂等键同一种机制）。REST 路径不接 +// 回调 —— 它的 409 chat_busy 承担同一职责。 + +// PendingFullNotifier 把「本次提交因待处理队列已满被拒」定向通知给发送者。 +// 它只作用于发起这条消息的那根连接：广播的 pending_full 是状态信号，各端自行 +// 按 clientMsgId 对齐，其它连接忽略即可。clientMsgID 与 messageID 都可能为 "" +// （该键省略，绝不伪造）。 +type PendingFullNotifier func(sessionID, clientMsgID, messageID string) + +// pendingFullNotifierKey 是携带定向拒绝回调的 context 键。它从 WS 帧边界 +// （ws_route.go 的 submitChat / voice.go 的 runVoiceTurn）引入编排器。 +type pendingFullNotifierKey struct{} + +// withPendingFullNotifier 返回携带定向拒绝回调的 context。nil 回调是空操作： +// 未接线传输层（REST / 单元测试）的提交在 ctx 上不留任何痕迹。 +func withPendingFullNotifier(ctx context.Context, n PendingFullNotifier) context.Context { + if n == nil { + return ctx + } + return context.WithValue(ctx, pendingFullNotifierKey{}, n) +} + +// pendingFullNotifierFrom 返回 ctx 携带的定向拒绝回调，未设置时为 nil。 +func pendingFullNotifierFrom(ctx context.Context) PendingFullNotifier { + n, _ := ctx.Value(pendingFullNotifierKey{}).(PendingFullNotifier) + return n +} + +// emitPendingFull 广播 pending_full 状态帧，并在接线了发送者回调时定向回一条 +// 拒绝帧。调用方必须已从状态锁退出（状态锁内不做广播，与既有纪律一致）。 +func (o *Orchestrator) emitPendingFull(ctx context.Context, sessionID string, mode Mode, rejected clientMsgHint) { + o.emitState(sessionID, EventPendingFull, pendingFullPayload(sessionID, mode, o.maxPending, rejected)) + if n := pendingFullNotifierFrom(ctx); n != nil { + n(sessionID, rejected.ClientMsgID, rejected.MessageID) + } +} + +// rejectedIdentityLocked 为队列满被拒的那条消息取「服务端能给出的身份」（G5）： +// clientMsgId 原样来自 ctx（省略时为空 ⇒ 该字段被省略）；messageId 只在它**此前 +// 已经落库或已认领**时给出。 +// +// 它绝不新铸 id：队列满的拒绝发生在认领之前，一旦在这里认领一个新 id，客户端用 +// 同一个键重试就会被判成「重复到达」而永远进不来（见 admitLocked 的顺序说明）。 +// 因此未落库时只报 clientMsgId。必须在 o.mu 下调用。 +func (o *Orchestrator) rejectedIdentityLocked(ctx context.Context, st *sessionState, sessionID string) clientMsgHint { + id := clientMsgIDFrom(ctx) + if id == "" { + return clientMsgHint{} + } + if msgID, ok := st.clientMsgs[id]; ok { + return clientMsgHint{ClientMsgID: id, MessageID: msgID} + } + if msgID, ok := o.findPersistedClientMsg(ctx, sessionID, id); ok { + o.rememberClientMsgLocked(st, id, msgID) + return clientMsgHint{ClientMsgID: id, MessageID: msgID} + } + return clientMsgHint{ClientMsgID: id} +} + +// pendingFullPayload 组装 pending_full 状态帧的载荷。既有三个键 +// （sessionId / mode / max）**一个不改**，新增键是纯增量： +// - clientMsgId：被拒那条消息的客户端幂等键（未携带时该键不出现）； +// - messageId：该键此前已落库/已认领时的消息 id（未落库时该键不出现）。 +func pendingFullPayload(sessionID string, mode Mode, max int, rejected clientMsgHint) map[string]any { + data := map[string]any{"sessionId": sessionID, "mode": mode, "max": max} + if rejected.ClientMsgID != "" { + data["clientMsgId"] = rejected.ClientMsgID + } + if rejected.MessageID != "" { + data["messageId"] = rejected.MessageID + } + return data +} diff --git a/internal/services/gateway/session_steer.go b/internal/services/gateway/session_steer.go index 5e9991f..a706cab 100644 --- a/internal/services/gateway/session_steer.go +++ b/internal/services/gateway/session_steer.go @@ -220,6 +220,9 @@ func (o *Orchestrator) consumePending(sessionID string, gen uint64, seqs map[uin for _, p := range st.pending { if seqs[p.seq] { removed = append(removed, p.messages...) + // 被注入的若是语音请求，它的音频要由本回合结算时一并送回（协同 C1）； + // 否则那条请求的回合已被本回合接手，它却永远等不到 voice_output。 + o.addVoiceSinkLocked(st, gen, p.voice) continue } kept = append(kept, p) diff --git a/internal/services/gateway/session_voice.go b/internal/services/gateway/session_voice.go new file mode 100644 index 0000000..79e6a37 --- /dev/null +++ b/internal/services/gateway/session_voice.go @@ -0,0 +1,111 @@ +package gateway + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" +) + +// 本文件归属协同项 C1 的「语音回合收尾」：把一次语音回合的结果交回语音层， +// 由它合成音频并**定向**回发起这条语音的连接。 +// +// 语音回合现在与文本回合走同一条路（进编排器 → 落库 → 流式 → 跨端同步）， +// 因此音频回复不能还在提交那一刻同步等待 —— 回合是异步完成的（可能排队、 +// 可能被中断）。编排器唯一知道「这个回合真正产出了什么」的地方是回合结算处 +// （completeTurn，与 S3.5 回写文档结论同一收尾点），所以收尾回调在那里触发。 +// +// 与 session_reject.go 同一纪律：编排器对传输层一无所知，「怎么把帧送回那根 +// 连接」由 WS 边界在 ctx 上给出。 + +// VoiceReplyFunc 是语音回合的收尾回调：回合结算后，编排器把该回合的结果交给 +// 它（res 可能为 nil —— 被取消或出错的回合没有结果）。 +type VoiceReplyFunc func(res *loop.Result) + +// voiceReplyKey 是携带语音收尾回调的 context 键。它从 WS 帧边界 +// （voice.go 的 runVoiceTurn）引入编排器，与 deviceID / 幂等键同一种机制。 +type voiceReplyKey struct{} + +// withVoiceReplySink 返回携带语音收尾回调的 context。nil 回调是空操作：文本 +// 回合（以及未接线传输层的提交）在 ctx 上不留任何痕迹。 +func withVoiceReplySink(ctx context.Context, sink VoiceReplyFunc) context.Context { + if sink == nil { + return ctx + } + return context.WithValue(ctx, voiceReplyKey{}, sink) +} + +// voiceReplySinkFrom 返回 ctx 携带的语音收尾回调，未设置时为 nil。 +func voiceReplySinkFrom(ctx context.Context) VoiceReplyFunc { + sink, _ := ctx.Value(voiceReplyKey{}).(VoiceReplyFunc) + return sink +} + +// voiceTurn 配对一次语音回合与它的收尾回调（按代次配对，与 turnDocuments +// 同理）：被打断或被新回合取代的陈旧回合在结算时不该替别人回音频。 +// +// sinks 是切片，因为一个回合可能同时有**多条**语音请求在等音频：忙时入 pending +// 的语音消息可能被转向注入（design/15 §2.4，见 session_steer.go 的 consumePending） +// 而由同一个回合作答 —— 它的音频必须由这个回合结算时一并送回，否则那条请求会 +// 静默地收不到声音（音频回复不能丢）。 +type voiceTurn struct { + gen uint64 + sinks []VoiceReplyFunc +} + +// voiceSinksOf 把单个回调收成切片（nil 回调 → nil 切片）。 +func voiceSinksOf(sink VoiceReplyFunc) []VoiceReplyFunc { + if sink == nil { + return nil + } + return []VoiceReplyFunc{sink} +} + +// rememberVoiceTurn 记下本回合的语音收尾回调。**每个新回合都必须调用它一次** +// （纯文本回合记 nil）：否则一个陈旧的语音回调会被沿用到后面的文本回合上， +// 让一次文本问答无端合成一段音频。 +func (o *Orchestrator) rememberVoiceTurn(sessionID string, gen uint64, sink VoiceReplyFunc) { + o.mu.Lock() + defer o.mu.Unlock() + if st := o.states[sessionID]; st != nil { + st.voiceTurn = voiceTurn{gen: gen, sinks: voiceSinksOf(sink)} + } +} + +// addVoiceSinkLocked 把一条**被转向注入**的语音请求的收尾回调追加到当前回合上。 +// 必须在 o.mu 下调用；代次不符（那条请求不属于本回合）时什么都不做。 +func (o *Orchestrator) addVoiceSinkLocked(st *sessionState, gen uint64, sink VoiceReplyFunc) { + if st == nil || sink == nil || st.voiceTurn.gen != gen { + return + } + st.voiceTurn.sinks = append(st.voiceTurn.sinks, sink) +} + +// takeVoiceTurn 取走与 gen 配对的所有语音收尾回调并清空记录。第二个返回值为 +// false 表示代次已陈旧（中断/取代）：迟到的结算不该回音频。 +func (o *Orchestrator) takeVoiceTurn(sessionID string, gen uint64) ([]VoiceReplyFunc, bool) { + o.mu.Lock() + defer o.mu.Unlock() + st := o.states[sessionID] + if st == nil || st.voiceTurn.gen != gen { + return nil, false + } + sinks := st.voiceTurn.sinks + st.voiceTurn = voiceTurn{} + return sinks, true +} + +// dispatchVoiceReply 在回合结算时把结果交回语音层（协同 C1）。不是语音回合时 +// 是空操作；被取代的陈旧回合同样什么都不做。 +// +// 每个回调在自己的 goroutine 上跑：TTS 是网络调用，绝不许它拖住编排器的状态迁移 +// （助手回复此刻已经落库，音频是它之后的增强；pending 续跑与 session_idle +// 不该等一段语音合成）。 +func (o *Orchestrator) dispatchVoiceReply(sessionID string, gen uint64, res *loop.Result) { + sinks, ok := o.takeVoiceTurn(sessionID, gen) + if !ok { + return + } + for _, sink := range sinks { + go sink(res) + } +} diff --git a/internal/services/gateway/voice.go b/internal/services/gateway/voice.go index d8ea193..68e5732 100644 --- a/internal/services/gateway/voice.go +++ b/internal/services/gateway/voice.go @@ -5,8 +5,10 @@ import ( "encoding/base64" "encoding/json" "fmt" + "log/slog" "strings" + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "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" @@ -87,6 +89,10 @@ type voiceInputRequest struct { Text string `json:"text"` SessionID string `json:"sessionId"` DeviceID string `json:"deviceId,omitempty"` + // ClientMsgID 是客户端自带的幂等键（协同 G4），与文本 chat 帧**同形**： + // 同一个非空 clientMsgId 只承认第一条，重复到达不再落库、不再起回合 + // （回一条 chat_duplicate）。它可选，省略时行为与加入该字段之前逐字相同。 + ClientMsgID string `json:"clientMsgId,omitempty"` } // SetVoice 注入 voice_input WS 路由所用的语音 STT/TTS 服务。 @@ -99,9 +105,9 @@ func (s *Server) SetVoice(v VoiceService) { } // handleVoiceInput 路由一个 voice_input 帧。未注入语音 -// 服务时它以明确错误应答（绝不崩溃）；否则它解码 -// 请求并异步运行 STT→chat→TTS 回合，使读泵绝不被 -// provider 延迟阻塞。 +// 服务时它以明确错误应答（绝不崩溃）；否则它异步跑这个语音 +// 回合（STT → 编排器 → 结算时的 TTS），使读泵绝不被 +// provider 或模型延迟阻塞。 func (c *wsConn) handleVoiceInput(in wsInbound) { if c.h.voice == nil { c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "voice unavailable"}}) @@ -116,14 +122,24 @@ func (c *wsConn) handleVoiceInput(in wsInbound) { go c.h.runVoiceTurn(c, req) } -// runVoiceTurn 驱动一个语音回合：STT（或提供的转写）→ agent chat -// → TTS，然后向请求连接回复一个 voice_output 帧， -// 携带 base64 音频加上识别出的转写与助手回复。 +// voiceLogSource 是语音路径的日志来源标签，使这条路径上的 WARN 能被追溯到 +// 具体子功能（AGENTS §10 的「日志必须标来源」）。 +const voiceLogSource = "gateway/voice" + +// runVoiceTurn 驱动一个语音回合。转写（STT）仍在这里完成 —— 识别是语音入口 +// 自己的事；而**回合**从此刻起与文本回合走同一条路：进编排器 → 落库 → 流式 → +// 跨端同步（协同 C1）。此前它直调 runner.Run，于是语音问过的事不进历史、不跨端、 +// 也不参与 busy/pending。 +// +// 音频回复不在本函数等待：回合是异步完成的（可能排队、可能被中断），因此 TTS 与 +// 定向回帧由收尾回调（voiceReplySink）在回合**结算**时完成 —— 那是编排器唯一 +// 知道这个回合产出了什么的地方（与 S3.5 回写文档结论同一收尾点）。 func (h *hub) runVoiceTurn(c *wsConn, req voiceInputRequest) { ctx := context.Background() - // 把客户端的 deviceID 串下来，使持久化的语音回合被发送端 - // 标注（design/16 §4.5），与 ws_hub.go:240 一致。 - ctx = withDeviceID(ctx, req.DeviceID) + // 把客户端的 deviceID 与可选幂等键串下来，使持久化的语音回合被发送端标注 + //（design/16 §4.5）并让重复到达的 clientMsgId 被判重（协同 G4）—— 与 + // ws_hub_turn.go 的文本路径同一机制。 + ctx = withClientMsgHint(withDeviceID(ctx, req.DeviceID), clientMsgHint{ClientMsgID: req.ClientMsgID}) if req.SessionID == "" { req.SessionID = "web" } @@ -135,13 +151,13 @@ func (h *hub) runVoiceTurn(c *wsConn, req voiceInputRequest) { } audio, err := decodeVoiceAudio(req.Audio, req.Format) if err != nil { - h.log.Warn("voice audio decode", "err", err) + h.voiceLog().Warn("voice audio decode", "err", err, "session", req.SessionID) h.voiceError(c, "voice audio could not be decoded") return } stt, err := h.voice.STT(ctx, audio) if err != nil { - h.log.Warn("voice stt", "err", err) + h.voiceLog().Warn("voice stt", "err", err, "session", req.SessionID) h.voiceError(c, "voice transcription failed") return } @@ -155,38 +171,110 @@ func (h *hub) runVoiceTurn(c *wsConn, req voiceInputRequest) { h.voiceError(c, "chat service is not configured") return } - // Web 主会话是主人的私密对话，因此它绑定 - // 全访问 env：记忆读+写，行为不变。角色来自该 - // 会话的绑定（design/22 §4.5）。 + // Web 主会话是主人的私密对话，因此它绑定全访问 env：记忆读+写，行为不变。 + // 角色来自该会话的绑定（design/22 §4.5）。 env := h.envResolver.owner(ctx, req.SessionID) - res, err := h.runner.Run(loop.WithSessionEnv(ctx, env), req.SessionID, []model.Message{model.TextMessage(model.RoleUser, transcript)}) + msgs := []model.Message{model.TextMessage(model.RoleUser, transcript)} + // 两条「怎么回到这条连接」的通路都由本连接给出：收尾回调（音频回复，C1）与 + // 定向拒绝回调（队列满，G5）。编排器只把它们当成本回合的传输线索。 + ctx = withVoiceReplySink(ctx, h.voiceReplySink(c, req.SessionID, req.DeviceID, transcript)) + ctx = withPendingFullNotifier(ctx, c.notifyPendingFull) + if h.orchestrator != nil { + h.orchestrator.Submit(ctx, req.SessionID, msgs, env) + return + } + h.runVoiceTurnDirect(ctx, c, req.SessionID, req.DeviceID, transcript, env) +} + +// runVoiceTurnDirect 是未接编排器时的旧直跑路径（未接线的构建 / 裸 hub 单元 +// 测试）：转写之后的回合直接跑 loop，然后立即合成并回音频。它与编排器路径共用 +// 同一个帧构造（emitVoiceOutput），因此两条路径的 voice_output 契约一致。 +func (h *hub) runVoiceTurnDirect(ctx context.Context, c *wsConn, sessionID, deviceID, transcript string, env loop.SessionEnv) { + res, err := h.runner.Run(loop.WithSessionEnv(ctx, env), sessionID, []model.Message{model.TextMessage(model.RoleUser, transcript)}) if err != nil { - h.log.Warn("voice chat run", "err", err, "session", req.SessionID) + h.voiceLog().Warn("voice chat run", "err", err, "session", sessionID, "device_id", deviceID) h.voiceError(c, "voice chat failed") return } - reply := res.Text + h.synthesizeVoiceReply(ctx, c, sessionID, deviceID, transcript, voiceReplyText(res)) +} + +// voiceReplySink 返回本次语音请求的收尾回调：回合结算后取助手正文 → TTS → +// **定向**回发起它的那根连接。它是 C1 的核心接线点 —— 回合异步完成，只有结算处 +// 知道助手正文。 +// +// 连接已断（客户端走了）时记一条带来源的 WARN，绝不静默丢：那正是「静默成功但 +// 没声音」那类缺陷的温床（协同 C2 的教训）。 +func (h *hub) voiceReplySink(c *wsConn, sessionID, deviceID, transcript string) VoiceReplyFunc { + return func(res *loop.Result) { + if c.closed() { + h.voiceLog().Warn("voice reply dropped", "reason", "connection closed", "session", sessionID, "device_id", deviceID) + return + } + // 回合的 ctx 已随回合消亡，因此合成用全新的后台 context：音频是「回复已 + // 落库」之后的增强，不该被一次被取消的回合连带取消。 + h.synthesizeVoiceReply(context.Background(), c, sessionID, deviceID, transcript, voiceReplyText(res)) + } +} + +// synthesizeVoiceReply 把助手正文合成音频并定向回发起这条语音的连接 +// （协同 C1/C3）。三条纪律： +// - TTS 失败或降级为空音频时**照旧回帧**并带既有 provider 降级信号，让客户端 +// 能提示「没生成声音」（绝不是只回一条错误帧就完事）； +// - 没有助手正文（被取消 / 空回复）时不打一次无谓的 TTS，直接回降级帧 —— +// 客户端总能看到这条语音回合的终点，而不是永远等一个不会来的 voice_output； +// - 日志只带会话与设备 id，**绝不记正文/音频**。 +func (h *hub) synthesizeVoiceReply(ctx context.Context, c *wsConn, sessionID, deviceID, transcript, reply string) { if strings.TrimSpace(reply) == "" { - reply = res.StopReason + h.voiceLog().Warn("voice reply empty", "reason", "turn produced no text", "session", sessionID, "device_id", deviceID) + emitVoiceOutput(c, sessionID, deviceID, transcript, "", voice.Audio{}, providerSignal("", true)) + return } audi, rung, err := ttsWithProvider(ctx, h.voice, reply) if err != nil { - h.log.Warn("voice tts", "err", err) - h.voiceError(c, "voice synthesis failed") + h.voiceLog().Warn("voice tts", "err", err, "session", sessionID, "device_id", deviceID) + emitVoiceOutput(c, sessionID, deviceID, transcript, reply, voice.Audio{}, providerSignal("", true)) return } + emitVoiceOutput(c, sessionID, deviceID, transcript, reply, audi, providerSignal(rung, len(audi.Data) == 0)) +} + +// emitVoiceOutput 定向回一条 voice_output 帧（只有发起这条语音的连接收得到）。 +// 载荷在既有四键（audio/format/text/reply）之上**纯增量**地带上 sessionId 与 +// deviceId（协同 C3）：多端同时说话时客户端能区分归属。provider 只在没有真实 +// 音频时出现（既有降级信号），它不是业务错误。 +func emitVoiceOutput(c *wsConn, sessionID, deviceID, transcript, reply string, audi voice.Audio, provider string) { data := map[string]any{ - "audio": base64.StdEncoding.EncodeToString(audi.Data), - "format": audi.Format, - "text": transcript, - "reply": reply, + "audio": base64.StdEncoding.EncodeToString(audi.Data), + "format": audi.Format, + "text": transcript, + "reply": reply, + "sessionId": sessionID, } - if sig := providerSignal(rung, len(audi.Data) == 0); sig != "" { - data["provider"] = sig + if deviceID != "" { + data["deviceId"] = deviceID + } + if provider != "" { + data["provider"] = provider } c.enqueueJSON(wsEvent{Type: wsVoiceOutput, Data: data}) } +// voiceReplyText 返回一个回合的助手正文（没有结果时为空串）。空正文**不拿 +// stopReason 顶替**：把 "stop"/"idle" 这类内部状态合成成助手声音，客户端会把它 +// 当回复显示；空正文一律走既有的 provider 降级信号，如实说「没生成声音」。 +func voiceReplyText(res *loop.Result) string { + if res == nil { + return "" + } + return res.Text +} + +// voiceLog 返回带语音来源标签的 logger（AGENTS §10：日志必须标来源）。 +func (h *hub) voiceLog() *slog.Logger { + return logger.WithSource(h.log, voiceLogSource) +} + // voiceError 为失败的语音回合回复一个带类型的错误帧。 func (h *hub) voiceError(c *wsConn, message string) { c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": message}}) diff --git a/internal/services/gateway/voice_turn_test.go b/internal/services/gateway/voice_turn_test.go new file mode 100644 index 0000000..61f9313 --- /dev/null +++ b/internal/services/gateway/voice_turn_test.go @@ -0,0 +1,352 @@ +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 + } + } +} diff --git a/internal/services/gateway/ws_hub_conn.go b/internal/services/gateway/ws_hub_conn.go index dc9d66d..0ef96a1 100644 --- a/internal/services/gateway/ws_hub_conn.go +++ b/internal/services/gateway/ws_hub_conn.go @@ -132,6 +132,19 @@ func (c *wsConn) shutdown() { }) } +// closed 报告连接是否已被关闭（readPump 退出时 shutdown 关闭 done）。异步收尾 +// 路径（如语音回合结算后的定向回帧）用它判断「这条回帧还送得到吗」—— send 通道 +// 不会在关闭时被关上，入队后只会静默丢失，所以必须显式问一次，才能给出一条带 +// 来源的 WARN 而不是无声无息。nil 字段（零值 wsConn）视为未关闭。 +func (c *wsConn) closed() bool { + select { + case <-c.done: + return true + default: + return false + } +} + // enqueue 把序列化后的帧入队，慢客户端缓冲满时丢弃它， // 使生产者绝不阻塞。 func (c *wsConn) enqueue(b []byte) { diff --git a/internal/services/gateway/ws_route.go b/internal/services/gateway/ws_route.go index 9e8d5b5..f894d6a 100644 --- a/internal/services/gateway/ws_route.go +++ b/internal/services/gateway/ws_route.go @@ -55,8 +55,9 @@ func (c *wsConn) route(in wsInbound) { c.handleInterrupt(in.Data) case "voice_input": // 语音经注入的 STT→chat→TTS 管线处理。没有 voice service 时回一个 - // 明确的错误，而不是把帧丢掉。音频回复作为 voice_output 事件发在 - // 发起请求的那根连接上。 + // 明确的错误，而不是把帧丢掉。转写之后的**回合**与文本同一条路（协同 + // C1：编排器 → 落库 → 流式 → 跨端同步），音频回复在回合结算时作为 + // voice_output 事件定向回发起请求的那根连接。 c.handleVoiceInput(in) case "subagent_ask": c.handleSubagentAsk(in) @@ -152,6 +153,28 @@ func claimRole(claims *auth.Claims) string { return string(claims.Role) } +// wsChatBusyCode 是待处理队列满时定向回发送者的机器可读错误码。它与 REST +// 同步/流式路径的 409 `chat_busy` **同码**：客户端用同一个 code 处理两条入口的 +// 「队列满」，不必为 WS 再造一套判定（协同 G5）。 +const wsChatBusyCode = "chat_busy" + +// notifyPendingFull 定向回一条拒绝帧，告诉发送这条消息的连接「会话正忙、待处理 +// 消息已满、本条未入队」，并带上能对齐本地乐观气泡的键（协同 G5）：clientMsgId +// 原样带回（未携带该键时不伪造），messageId 只在该键此前已落库/已认领时出现。 +// +// 它只到发送者、不广播：广播的 pending_full 是状态信号，其余连接按 clientMsgId +// 自行忽略即可。 +func (c *wsConn) notifyPendingFull(sessionID, clientMsgID, messageID string) { + data := map[string]any{"code": wsChatBusyCode, "message": "会话正忙，待处理消息已满，本条未入队", "sessionId": sessionID} + if clientMsgID != "" { + data["clientMsgId"] = clientMsgID + } + if messageID != "" { + data["messageId"] = messageID + } + c.enqueueJSON(wsEvent{Type: "error", Data: data}) +} + // submitChat 把一条 chat 上行帧直接交给会话编排器（design/15 §2.1/§2.2）， // 而不是经 bus 转发：会话消息不得依赖可丢弃的通道（见 route 的 chat 分支）。 // 解码、角色解析与用户消息落库都在该 goroutine 上完成，因此读泵绝不等待 @@ -168,6 +191,8 @@ func (c *wsConn) submitChat(payload any) { return } // 回合自身的取消由编排器持有（Interrupt/InterruptAll 取消它）， - // 因此这里的基 ctx 只需是存活到回合结束的背景上下文。 - go c.h.streamChat(context.Background(), payload) + // 因此这里的基 ctx 只需是存活到回合结束的背景上下文。队列满的拒绝要定向回 + // **这条**连接（协同 G5）：编排器只认 ctx 上给出的回调（REST 走 409，不接）。 + ctx := withPendingFullNotifier(context.Background(), c.notifyPendingFull) + go c.h.streamChat(ctx, payload) }