From 5717735da9b2b23c5ce6efb4ca48414a0b783e93 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 26 Oct 2026 18:46:06 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(gateway):=20M44=20=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E5=B8=A7=E4=B8=8E=E6=8F=A1=E6=89=8B=E5=9B=9E=E6=94=BE=E6=8C=82?= =?UTF-8?q?=E4=B8=8A=20voice=EF=BC=88=E7=BA=AF=E5=A2=9E=E9=87=8F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修实根：`message_persisted` / `channel_persisted` 与握手 `history` 回放 此前只在 `toHistoryMessage` 之后直接扇出，于是消息 DTO 的语音面在这三条 面上恒缺席，客户端只能等下一次 REST 拉取才知道那是一条语音。 - `history_voice.go`：把语音补齐抽成存储无关的唯一收口点 `attachVoiceTo` （Server 与 hub 共用），并加 `voiceColumnsOf` 把任意历史视图窄化成读面， 未实现该视图时为真 nil 接口（字段缺席，不 panic、不 500）。 - `broadcast.go`：新增 `persistedWire` —— thinking 抹除 + 可选 voice， `broadcastPersisted` 改用它；两类帧因此天然同口径。 - `ws_history.go`：握手回放同样补一次批量读列（最多 50 个 id 一条 SQL）。 成本实测：一帧 = 1 次 `IN (?)` 批量读列（1 个 id）；回放一页 = 1 次 （去重后的全部 id）；存储未提供该视图时 0 次。`history.Message` 没有 voice 列字段（`messageColumns` 也不含），故无法「零查询直接填」——补它 要改 internal/runtime/history 与 INSERT 写路径，超出本切片范围。 纯增量：voice 是可选键，非语音消息整键缺席（不是 null、不是 {}）； 帧壳、帧类型、REST DTO 形状一律未动。 测试（广播_voice_test.go）：帧序列化输出上断言 ①②③④，并逐字比对 「帧/回放里的 voice」与「REST 读面同一行」；摘掉任一挂载即打红。 --- internal/services/gateway/broadcast.go | 26 +- .../services/gateway/broadcast_voice_test.go | 436 ++++++++++++++++++ internal/services/gateway/history_voice.go | 55 ++- internal/services/gateway/ws_history.go | 9 +- 4 files changed, 507 insertions(+), 19 deletions(-) create mode 100644 internal/services/gateway/broadcast_voice_test.go diff --git a/internal/services/gateway/broadcast.go b/internal/services/gateway/broadcast.go index 07cdbb2..45dae9c 100644 --- a/internal/services/gateway/broadcast.go +++ b/internal/services/gateway/broadcast.go @@ -1,6 +1,7 @@ package gateway import ( + "context" "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" @@ -36,10 +37,10 @@ func (s *Server) BroadcastPersisted(msg history.Message) { // 帧的 data 另带**可选**的 `unreadCount`（契约 v0.76）：该会话此刻的绝对未读数， // 与 GET /sessions 的 `unreadCount` 同一口径。它只在 >0 且真能算出来时出现（见 // unreadCountOf），缺席即老行为。Web 路径与 IM 路径共用本函数 ⇒ 两处口径天然一致， -// 不存在「Web 帧带、IM 帧不带」的分叉。 +// 不存在「Web 帧带、IM 帧不带」的分叉。帧内的 `message` 对象另带**可选**的 `voice` +// （契约 v0.67 §4.4；见 persistedWire）：非语音消息整键缺席，因此老客户端逐字不变。 func (s *Server) broadcastPersisted(msg history.Message, proactive bool) { - wire := toHistoryMessage(msg) - wire.Thinking = "" + wire := s.persistedWire(msg) typ := "channel_persisted" if msg.SessionID == history.MainSessionID || strings.HasPrefix(msg.SessionID, history.MainSessionID+":") { typ = "message_persisted" @@ -58,6 +59,25 @@ func (s *Server) broadcastPersisted(msg history.Message, proactive bool) { s.hub.broadcast(wsEvent{Type: typ, SessionID: msg.SessionID, Data: data}) } +// persistedWire 把一条待广播的消息投影成帧内的 `message` 对象：thinking 抹除（与 WS +// 历史回放一致，design/16 §6）＋ 可选的语音面（契约 v0.67 §4.4）。 +// +// 语音面与 REST 读面共用**同一个**收口点（attachVoiceTo → historyVoiceOf），因此 +// 「同一行消息在帧里与 GET /sessions/{id}/messages 里」是逐字相同的对象，客户端不必 +// 再拉一次 REST 才认出语音气泡。 +// +// 代价（实测，见交付报告）：每帧**一次**批量读列（`IN (?)` 一条 SQL，页内 id 去重）， +// 与本路径既有的 unreadCountOf 查询同量级；存储未提供该视图时**零**查询、字段缺席。 +// 之所以要查库而不是「从 history.Message 直接填」：那个结构体没有 voice 列字段 +// （`messageColumns` 也不含它），补它要改 `internal/runtime/history` 与 INSERT 写路径 +// ——那是本切片的范围之外（见交付报告的跨片接缝）。 +func (s *Server) persistedWire(msg history.Message) historyMessage { + one := []historyMessage{toHistoryMessage(msg)} + one[0].Thinking = "" + attachVoiceTo(context.Background(), s.log, voiceColumnsOf(s.history), one) + return one[0] +} + // wsCapacityAwaken 是能力状态迁移的 WS 事件类型（design/14 §5）。 // 其 data 字段是前端的 AwakenPayload {fromState,toState,tool}。 const wsCapacityAwaken = "capacity_awaken" diff --git a/internal/services/gateway/broadcast_voice_test.go b/internal/services/gateway/broadcast_voice_test.go new file mode 100644 index 0000000..0439276 --- /dev/null +++ b/internal/services/gateway/broadcast_voice_test.go @@ -0,0 +1,436 @@ +package gateway + +// 本文件钉住 M44：语音面在**实时帧**与**握手回放**上的纯增量覆盖（契约 v0.67 §4.4）。 +// +// 五条断言各对应一条取舍： +// ① `message_persisted` 的 `data.message.voice` 出现，且与 `GET /sessions/{id}/messages` +// 同一行**逐字相等** —— 这是本切片唯一的意义：客户端不必再拉一次 REST 才认出语音气泡； +// ② 非语音消息该键**整键缺席**（不是 `null`、不是 `{}`）； +// ③ `channel_persisted`（隔离渠道会话）同样带，且与渠道日志读面同一行逐字相等； +// ④ 握手 `history` 回放同样带（老客户端因此重连即可拿到，不必等重新拉取）； +// ⑤ 成本实测：一帧 = **一次**批量读列（`IN (?)`）、回放一页 = **一次**（id 去重）—— +// 不是逐条查，也不是「每帧多打几枪」。 +// +// 断言一律打在**帧的序列化输出**上（`readFakeFrame` / `ReadMessage` 都是先 marshal 再 +// 回读），因此「键缺席」与「零值」的区别是真的可分辨的 —— 解码进结构体做不到这一点。 + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/db" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" +) + +// countingVoiceStore 在真实存储外面只给 `VoiceColumns` 记账：本切片的核心成本问题是 +// 「一条消息多打几枪、一次带几个 id」，所以把调用次数与 id 数**量出来**，而不是凭印象 +// 说「一次批量」。 +type countingVoiceStore struct { + *history.Store + mu sync.Mutex + calls int + ids []string +} + +func (c *countingVoiceStore) VoiceColumns(ctx context.Context, ids []string) (map[string]string, error) { + c.mu.Lock() + c.calls++ + c.ids = append(c.ids, ids...) + c.mu.Unlock() + return c.Store.VoiceColumns(ctx, ids) +} + +// reset 把计数器归零：每个断言段各自量自己的代价。 +func (c *countingVoiceStore) reset() { + c.mu.Lock() + defer c.mu.Unlock() + c.calls, c.ids = 0, nil +} + +// snapshot 返回 (调用次数, 各次查询的 id 并集)。 +func (c *countingVoiceStore) snapshot() (int, []string) { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls, append([]string(nil), c.ids...) +} + +// voiceFrameEnv 打包实时帧 / 回放测试所需的依赖：真实（内存）历史存储（外套记账层） +// + httptest 网关 + JWT 服务。 +type voiceFrameEnv struct { + srv *Server + ts *httptest.Server + jwt *auth.JWTService + st *history.Store + count *countingVoiceStore +} + +func newVoiceFrameEnv(t *testing.T) *voiceFrameEnv { + t.Helper() + ctx := context.Background() + conn, err := db.Open(ctx, ":memory:", testLogger()) + if err != nil { + t.Fatalf("打开内存库: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + st, err := history.NewStore(ctx, conn, testLogger()) + if err != nil { + t.Fatalf("构造历史存储: %v", err) + } + count := &countingVoiceStore{Store: st} + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{History: count}) + return &voiceFrameEnv{srv: srv, ts: ts, jwt: jwt, st: st, count: count} +} + +// seed 追加一条消息，并把（可选的）语音事实写进它的 voice 列 —— 这与生产里渠道桥 +// 「落库后回填同一列」的两段式同形。 +func (e *voiceFrameEnv) seed(t *testing.T, m history.Message, raw string) history.Message { + t.Helper() + got, err := e.st.AppendFull(context.Background(), m) + if err != nil { + t.Fatalf("落库消息: %v", err) + } + if raw == "" { + return got + } + if err := e.st.UpdateMessageVoice(context.Background(), got.ID, raw); err != nil { + t.Fatalf("写 voice 列: %v", err) + } + return got +} + +// encodeVoiceFact 编码一条语音事实；失败即测试前置条件坏掉。 +func encodeVoiceFact(t *testing.T, ref voiceref.Ref) string { + t.Helper() + raw, err := voiceref.Encode(ref) + if err != nil { + t.Fatalf("编码语音事实: %v", err) + } + return raw +} + +// canonicalJSON 把一段 JSON 归一成「键序由 json.Marshal 的 map 排序决定」的文本。 +// JSON 对象的键**无序**，因此「逐字相等」在对象层面就该按归一后的文本来判：两侧都走 +// 同一套归一，任何一处多一个键 / 少一个键 / 值不同都会让文本不同。 +func canonicalJSON(t *testing.T, raw json.RawMessage) string { + t.Helper() + var v any + if err := json.Unmarshal(raw, &v); err != nil { + t.Fatalf("解析 JSON（raw=%s）: %v", raw, err) + } + b, err := json.Marshal(v) + if err != nil { + t.Fatalf("归一 JSON（raw=%s）: %v", raw, err) + } + return string(b) +} + +// frameMessageVoice 从**已序列化并回读**的帧里取出 `data.message.voice`，返回归一后的 +// JSON 文本与「该键是否在场」。本面的关键取舍就是「缺席」，因此先断言在场性再看内容。 +func frameMessageVoice(t *testing.T, ev wsEvent) (string, bool) { + t.Helper() + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("帧 data = %+v, want 一个 JSON 对象", ev.Data) + } + msg, ok := data["message"].(map[string]any) + if !ok { + t.Fatalf("帧 data.message = %+v, want 一个 JSON 对象", data["message"]) + } + raw, present := msg["voice"] + if !present { + return "", false + } + b, err := json.Marshal(raw) + if err != nil { + t.Fatalf("归一帧里的 voice（%v）: %v", raw, err) + } + return string(b), true +} + +// replayMessageVoice 与 frameMessageVoice 同形，只是取的是握手 `history` 帧里 +// `data.messages[]` 的某一条（按 id 定位）。 +func replayMessageVoice(t *testing.T, ev wsEvent, id string) (string, bool) { + t.Helper() + data, ok := ev.Data.(map[string]any) + if !ok { + t.Fatalf("回放帧 data = %+v, want 一个 JSON 对象", ev.Data) + } + rows, ok := data["messages"].([]any) + if !ok { + t.Fatalf("回放帧 data.messages = %+v, want 一个数组", data["messages"]) + } + for _, row := range rows { + msg, ok := row.(map[string]any) + if !ok { + t.Fatalf("回放消息 = %+v, want 一个 JSON 对象", row) + } + if msg["id"] != id { + continue + } + raw, present := msg["voice"] + if !present { + return "", false + } + b, err := json.Marshal(raw) + if err != nil { + t.Fatalf("归一回放里的 voice（%v）: %v", raw, err) + } + return string(b), true + } + t.Fatalf("回放里没有 id=%q 的消息", id) + return "", false +} + +// restVoiceOf 取 REST 读面里**同一行**消息的 `voice`，归一成与帧侧可比的文本。 +// 键缺席时返回 ("", false)。 +func restVoiceOf(t *testing.T, rows []map[string]json.RawMessage, id string) (string, bool) { + t.Helper() + for _, row := range rows { + var gotID string + if err := json.Unmarshal(row["id"], &gotID); err != nil { + t.Fatalf("解析 id: %v", err) + } + if gotID != id { + continue + } + raw, present := row["voice"] + if !present { + return "", false + } + return canonicalJSON(t, raw), true + } + t.Fatalf("REST 读面里没有 id=%q 的消息", id) + return "", false +} + +// TestBroadcastPersistedCarriesVoiceMatchingRestRead 断言 ①②⑤：语音消息的 +// `message_persisted` 帧带 `voice`、与 REST 读面逐字相等；非语音消息整键缺席； +// 一帧的代价是一次批量读列（一个 id）。 +func TestBroadcastPersistedCarriesVoiceMatchingRestRead(t *testing.T) { + t.Parallel() + env := newVoiceFrameEnv(t) + tok := signAny(t, env.jwt, "acct", auth.RoleOwner) + name := "0123456789abcdef0123456789abcdef.ogg" + voiceMsg := env.seed(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "转写文本", + }, encodeVoiceFact(t, voiceref.Ref{ + Source: voiceref.SourceASR, + Status: voiceref.StatusTranscribed, + Name: name, + MIME: "audio/ogg", + Size: 4242, + DurationMs: 3200, + })) + plainMsg := env.seed(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "普通文本", + }, "") + + c := newOwnerConn(env.srv.hub, 8) + env.srv.hub.register(c) + + env.count.reset() + env.srv.BroadcastPersisted(voiceMsg) + ev := readFakeFrame(t, c) + if ev.Type != "message_persisted" { + t.Fatalf("帧类型 = %q, want message_persisted", ev.Type) + } + frameVoice, present := frameMessageVoice(t, ev) + if !present { + t.Fatalf("语音消息的 message_persisted 帧缺 data.message.voice（帧=%+v）", ev) + } + // ⑤ 成本：一帧 = 一次批量读列，恰好一个 id（没有逐条多打一枪）。**先**取快照， + // 免得下面那次 REST 拉取的记账混进来。 + calls, ids := env.count.snapshot() + + // 与 REST 读面同一行**逐字相等**：这是本切片唯一的意义。 + resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("REST status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + restVoice, restOK := restVoiceOf(t, messagePage(t, body), voiceMsg.ID) + if !restOK { + t.Fatalf("REST 读面同一行缺 voice 键（body=%s）", body) + } + if frameVoice != restVoice { + t.Fatalf("帧里的 voice = %s, REST 读面同一行 = %s（两处各算一份就失去意义）", frameVoice, restVoice) + } + + // 字段值逐项落地（给失败信息可读性；形状正确性由上面的逐字比较兜底）。 + var fields historyVoice + if err := json.Unmarshal([]byte(frameVoice), &fields); err != nil { + t.Fatalf("解析帧里的 voice: %v（raw=%s）", err, frameVoice) + } + if fields.Source != "asr" || fields.Status != "transcribed" || + fields.Name != name || fields.MIME != "audio/ogg" || fields.Size != 4242 || + fields.DurationMs != 3200 || fields.URL != voiceref.MediaPathPrefix+name { + t.Fatalf("帧里的 voice 字段 = %+v", fields) + } + + // ⑤ 成本：一帧 = 一次批量读列，恰好一个 id（没有逐条多打一枪）。 + if calls != 1 || len(ids) != 1 || ids[0] != voiceMsg.ID { + t.Fatalf("一帧的读列代价 = %d 次调用 / ids=%v，want 1 次 / [%s]", calls, ids, voiceMsg.ID) + } + + // ② 非语音消息：该键**整键缺席**（不是 `null`、不是 `{}`）。 + env.srv.BroadcastPersisted(plainMsg) + plainFrame := readFakeFrame(t, c) + if got, present := frameMessageVoice(t, plainFrame); present { + t.Fatalf("非语音消息的帧带上了 voice = %s，want 整键缺席", got) + } +} + +// TestChannelPersistedCarriesVoiceMatchingChannelLogRead 断言 ③⑤：隔离渠道会话 +// （群聊 / 陌生人）走的是同一个构造函数，因此 `channel_persisted` 同样带 `voice`， +// 且与渠道日志读面同一行逐字相等 —— 「Web 帧带、IM 帧不带」的分叉不存在。 +func TestChannelPersistedCarriesVoiceMatchingChannelLogRead(t *testing.T) { + t.Parallel() + env := newVoiceFrameEnv(t) + tok := signAny(t, env.jwt, "acct", auth.RoleOwner) + adapter := "qq-obv11:10001" + name := "ffffffffffffffffffffffffffffffff.mp3" + msg := env.seed(t, history.Message{ + SessionID: "sess-group-1", Role: history.RoleUser, Content: "群里的语音", + Platform: "qq-obv11", AdapterID: adapter, SenderName: "某人", + }, encodeVoiceFact(t, voiceref.Ref{ + Source: voiceref.SourceASR, Status: voiceref.StatusTranscribed, + Name: name, MIME: "audio/mpeg", Size: 99, + })) + + c := newOwnerConn(env.srv.hub, 8) + env.srv.hub.register(c) + + env.count.reset() + env.srv.BroadcastPersisted(msg) + ev := readFakeFrame(t, c) + if ev.Type != "channel_persisted" { + t.Fatalf("帧类型 = %q, want channel_persisted（隔离会话不是 Web 主会话）", ev.Type) + } + frameVoice, present := frameMessageVoice(t, ev) + if !present { + t.Fatalf("隔离会话的 channel_persisted 帧缺 data.message.voice（帧=%+v）", ev) + } + // ⑤ 成本快照**先**取，免得下面那次渠道日志拉取的记账混进来。 + calls, ids := env.count.snapshot() + + resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/admin/channels/messages?adapter="+adapter, nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("渠道日志 status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + restVoice, restOK := restVoiceOf(t, messagePage(t, body), msg.ID) + if !restOK { + t.Fatalf("渠道日志同一行缺 voice 键（body=%s）", body) + } + if frameVoice != restVoice { + t.Fatalf("帧里的 voice = %s, 渠道日志同一行 = %s", frameVoice, restVoice) + } + + if calls != 1 || len(ids) != 1 { + t.Fatalf("一帧的读列代价 = %d 次调用 / ids=%v，want 1 次 / 1 个 id", calls, ids) + } +} + +// TestWSHandshakeReplayCarriesVoice 断言 ④⑤：握手 `history` 回放同样带 `voice` +// （与 REST 读面同一行逐字相等），非语音行之键缺席，且整页只查一次列。 +func TestWSHandshakeReplayCarriesVoice(t *testing.T) { + t.Parallel() + env := newVoiceFrameEnv(t) + tok := signAny(t, env.jwt, "acct", auth.RoleOwner) + name := "abcdefabcdefabcdefabcdefabcdefab.ogg" + voiceMsg := env.seed(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleUser, Content: "回放里的语音", CreatedAt: 1000, + }, encodeVoiceFact(t, voiceref.Ref{ + Source: voiceref.SourceASR, Status: voiceref.StatusTranscribed, + Name: name, MIME: "audio/ogg", Size: 777, DurationMs: 1200, + })) + plainMsg := env.seed(t, history.Message{ + SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "回放里的纯文本", CreatedAt: 1500, + }, "") + + env.count.reset() + conn := dialWS(t, env.ts.URL, tok) + defer conn.Close() + _ = conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, raw, err := conn.ReadMessage() + if err != nil { + t.Fatalf("读取握手帧: %v", err) + } + var ev wsEvent + if err := json.Unmarshal(raw, &ev); err != nil { + t.Fatalf("解析握手帧 %s: %v", raw, err) + } + if ev.Type != "history" { + t.Fatalf("首帧类型 = %q, want history（%s）", ev.Type, raw) + } + replayVoice, present := replayMessageVoice(t, ev, voiceMsg.ID) + if !present { + t.Fatalf("握手回放里语音行缺 voice（帧=%s）", raw) + } + // ⑤ 整页一次批量：恰好 1 次调用、把两行的 id 都带上。快照**先**取，免得下面那次 + // REST 拉取的记账混进来。 + calls, ids := env.count.snapshot() + + // 与 REST 读面同一行逐字相等。 + resp, body := doReq(t, http.MethodGet, env.ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("REST status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + restVoice, restOK := restVoiceOf(t, messagePage(t, body), voiceMsg.ID) + if !restOK { + t.Fatalf("REST 读面同一行缺 voice 键（body=%s）", body) + } + if replayVoice != restVoice { + t.Fatalf("回放里的 voice = %s, REST 读面同一行 = %s", replayVoice, restVoice) + } + if got, present := replayMessageVoice(t, ev, plainMsg.ID); present { + t.Fatalf("回放里非语音行带上了 voice = %s，want 整键缺席", got) + } + + // ⑤ 整页一次批量：恰好 1 次调用，且把两行的 id 都带上了。 + if calls != 1 || len(ids) != 2 { + t.Fatalf("回放一页的读列代价 = %d 次调用 / ids=%v，want 1 次 / 2 个 id", calls, ids) + } +} + +// TestBroadcastPersistedVoiceAbsentWhenStoreCannotAnswer 钉住两条**诚实缺席**路径： +// 未接线历史存储、以及存储没提供语音列读面（测试替身 / 只做基本读写的实现）。 +// 两者都不许 panic、不许把帧吞掉 —— 那条路径上「问不到」的结果就是字段缺席。 +func TestBroadcastPersistedVoiceAbsentWhenStoreCannotAnswer(t *testing.T) { + t.Parallel() + pending := history.Message{ + ID: "row-1", SessionID: history.MainSessionID, Role: history.RoleUser, Content: "hi", CreatedAt: 1000, + } + + t.Run("history not wired", func(t *testing.T) { + srv, _, _ := buildEnv(t, cfgDefault(), Deps{}) + c := newOwnerConn(srv.hub, 8) + srv.hub.register(c) + + srv.BroadcastPersisted(pending) + ev := readFakeFrame(t, c) + if ev.Type != "message_persisted" { + t.Fatalf("帧类型 = %q, want message_persisted（问不到语音列不该吞掉帧）", ev.Type) + } + if got, present := frameMessageVoice(t, ev); present { + t.Fatalf("未接线历史存储的帧带上了 voice = %s，want 缺席", got) + } + }) + + t.Run("store without the voice view", func(t *testing.T) { + srv, _, _ := buildEnv(t, cfgDefault(), Deps{History: &fakeHistory{}}) + c := newOwnerConn(srv.hub, 8) + srv.hub.register(c) + + srv.BroadcastPersisted(pending) + ev := readFakeFrame(t, c) + if got, present := frameMessageVoice(t, ev); present { + t.Fatalf("未实现语音列读面的存储替身带上了 voice = %s，want 缺席", got) + } + }) +} diff --git a/internal/services/gateway/history_voice.go b/internal/services/gateway/history_voice.go index 300813b..b043759 100644 --- a/internal/services/gateway/history_voice.go +++ b/internal/services/gateway/history_voice.go @@ -20,13 +20,21 @@ package gateway // 没有「读不出」这一态，因此缺席是唯一诚实的表达（与文档列回一条 `unreadable` // 不同：文档有那个态，语音没有）。 // -// 覆盖范围（**如实**）：本文件的补齐点挂在会用 DTO 的读侧出口上（会话消息页、渠道 -// 消息日志两个端点、编辑消息的响应与其广播帧）。**尚未覆盖**：`message_persisted` / -// `channel_persisted`（`broadcast.go`）与握手 `history` 回放（`ws_history.go`）—— -// 前者还额外需要「随 INSERT 写入」的写路径（设计 §10.5 的首选口径），见交付报告。 +// 覆盖范围（**如实**）：本文件的补齐点挂在**所有**会用 DTO 的出口上 —— 会话消息页、 +// 渠道消息日志两个端点、编辑消息的响应与其 `message_edited` 帧、`message_persisted` / +// `channel_persisted` 实时帧（`broadcast.go` 的 persistedWire）与握手 `history` 回放 +// （`ws_history.go`）。三条收口点共用本文件这一份投影，因此「同一条消息在 REST 读面与 +// 实时帧 / 回放里的 `voice` 对象」逐字相同，不会两处各算一份。 +// +// ⚠️ 实时帧那条路的**已知边界**（不在本文件能修的范围内）：入站语音的 `voice` 列由 +// 渠道桥在**消息落库之后**回填（channelbridge 的两段式），而 `message_persisted` 那一帧 +// 在**落库那一刻**就发出去了 ⇒ 这一帧读到的列仍是空串、`voice` 依旧缺席；紧随其后的 +// 握手回放 / REST 拉取才带上它。要让「刚落库的那一帧」也带上，须把渠道桥的两段式顺序 +// 倒过来（先写列、再 notify），见交付报告与 Docs §4.4 的同一处如实登记。 import ( "context" + "log/slog" "strings" voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice" @@ -65,24 +73,30 @@ type voiceColumnReader interface { VoiceColumns(ctx context.Context, ids []string) (map[string]string, error) } -// attachVoice 给一页消息补上各自的语音面（就地写进 DTO 切片）。它只查页内**非空** -// 的消息 id，且**一次批量**（页内 id 集合交给存储侧一条 SQL 查完）。 +// attachVoice 给一页消息补上各自的语音面（就地写进 DTO 切片）。 +func (s *Server) attachVoice(ctx context.Context, msgs []historyMessage) { + attachVoiceTo(ctx, s.log, voiceColumnsOf(s.history), msgs) +} + +// attachVoiceTo 是语音面的**唯一**收口点：REST 读面、实时帧与握手回放都调它，因此三处 +// 对同一条消息给出**同一个**投影（`historyVoiceOf`），不存在「REST 带、帧不带」的分叉。 +// +// 它只查页内**非空**的消息 id，且**一次批量**（页内 id 集合交给存储侧一条 `IN (...)` +// 查完）—— 成本与「一页 / 一帧」同量级，不是逐条查。 +// +// reader 为 nil（存储未提供这一面：未接线 / 测试替身）或 log 为 nil 时都只让这一面整体 +// 缺席：这条路径上「问不到」不是错误，帧与响应照常发出。 // // 失败处理与工具回看面同一纪律：查不到就只让这一面缺席并记 WARN —— 消息本身已经 // 读到了，让一个可选列的故障把整页历史变成 500 是反的。 -func (s *Server) attachVoice(ctx context.Context, msgs []historyMessage) { +func attachVoiceTo(ctx context.Context, log *slog.Logger, reader voiceColumnReader, msgs []historyMessage) { ids := pageMessageIDs(msgs) - if len(ids) == 0 { - return - } - reader, ok := s.history.(voiceColumnReader) - if !ok { - // 存储未提供这一面（未接线 / 测试替身）：整体缺席，不是错误。 + if len(ids) == 0 || reader == nil || log == nil { return } cols, err := reader.VoiceColumns(ctx, ids) if err != nil { - s.log.Warn("history voice unavailable", "err", err, "messages", len(ids)) + log.Warn("history voice unavailable", "err", err, "messages", len(ids)) return } for i := range msgs { @@ -94,13 +108,24 @@ func (s *Server) attachVoice(ctx context.Context, msgs []historyMessage) { if err != nil { // 坏列**可见地**缺席：绝不静默当成「不是语音」。日志只带消息 id 与错误， // 绝不带列内容或正文。 - s.log.Warn("history voice column malformed", "err", err, "message", msgs[i].ID) + log.Warn("history voice column malformed", "err", err, "message", msgs[i].ID) continue } msgs[i].Voice = historyVoiceOf(ref) } } +// voiceColumnsOf 把任意历史存储视图窄化成语音列读面；没实现时返回**真 nil 接口** +// （`_, ok` 形式让未实现的存储得到 nil，而不是一个非 nil 的 nil 指针 —— 后者会在 +// 调用点炸成 panic）。Server 与 hub 持有的 history 字段同源，因此两处同一结果。 +func voiceColumnsOf(store any) voiceColumnReader { + reader, ok := store.(voiceColumnReader) + if !ok { + return nil + } + return reader +} + // pageMessageIDs 收集页内非空且**互不相同**的消息 id（保持首次出现顺序）。去重让 // 同一页的「提问 + 回答」或编辑场景里的两份 DTO 共用同一次查询。 func pageMessageIDs(msgs []historyMessage) []string { diff --git a/internal/services/gateway/ws_history.go b/internal/services/gateway/ws_history.go index 80d81bb..8698c24 100644 --- a/internal/services/gateway/ws_history.go +++ b/internal/services/gateway/ws_history.go @@ -25,7 +25,9 @@ const wsReplayLimit = 50 // // 回放是尽力而为：store 为 nil 或读取出错都静默跳过。 // 回放不含 thinking（design/16 §6），因此该帧即使对 owner/admin -// 也不会泄漏思维链。 +// 也不会泄漏思维链。每条消息**可能**带可选 `voice`（契约 v0.67 §4.4）， +// 且与 REST 读面同一份投影 —— 它的缺席语义同样是「这条消息不是语音」 +// （非语音消息整键缺席），因此回放对老客户端仍是同一份形状。 func (h *hub) replayHistory(c *wsConn, sessionID string) { if !ownerAdminClaims(c.claims) { return @@ -44,5 +46,10 @@ func (h *hub) replayHistory(c *wsConn, sessionID string) { wire.Thinking = "" out = append(out, wire) } + // 语音面（契约 v0.67 §4.4）：回放与 REST 读面共用**同一个**收口点与同一份投影， + // 因此刚连上的客户端就能按 id 认出语音气泡，不必等下一次 REST 拉取。一次批量读列 + // （最多 wsReplayLimit 条 id 一条 SQL），不是逐条查；存储没提供该视图 / 读失败时 + // 只让该字段缺席，回放本身照常发出（尽力而为，与上面的 Recent 同一纪律）。 + attachVoiceTo(context.Background(), h.log, voiceColumnsOf(h.history), out) c.enqueueJSON(wsEvent{Type: "history", SessionID: sessionID, Data: map[string]any{"messages": out}}) } From b955a274791b95e11bb23de4e017e1d4eb7aec36 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 26 Oct 2026 18:54:13 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(channelbridge):=20M44=20=E5=85=A5?= =?UTF-8?q?=E7=AB=99=E8=AF=AD=E9=9F=B3=E5=85=88=E5=86=99=20voice=20?= =?UTF-8?q?=E5=88=97=E5=86=8D=E5=B9=BF=E6=92=AD=20=E2=87=92=20=E9=A6=96?= =?UTF-8?q?=E5=B8=A7=E5=B0=B1=E5=B8=A6=20voice?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 安卓 2026-10-26 回执关心的场景：新来一条语音消息时，**第一条**帧里有没有 `voice`。此前没有 —— `persistBatchUsers` 先 `persistToMain`/`persistIsolated` （内部同步 notify 把帧发出去），之后才 `persistMessageVoice` 写列，帧发出那 一刻列还是空串。网关侧的帧投影（上一 commit）读的就是这一列，因此首帧必然 缺 voice，客户端要等重连或下一次 REST 拉取。 修法（A 路的唯一可行形态，理由见下）： - `persistBatchUsers` 的通知改成 `persistVoiceFirstNotify(ref)`： **INSERT → 写 voice 列 → notify**。语句数不变，只把 notify 从 INSERT 内部 挪到最后；非语音消息（ref 零值）时该包装与 `notifyPersisted` 逐字等价。 - `persist.go` 抽出 `mainRow` / `isolatedRow`（字段清单只写一次）与 `persistIsolatedWith`（可注入通知的变体），`persistToMain`/ `persistIsolated` 的对外行为逐字不变。 为什么不走 B（给 `history.Message` 加 voice 随 INSERT 写）：`persistMessageVoice` 是按 id 的 UPDATE，行还不存在时必然 `ErrMessageNotFound` ⇒ 写列**不可能**真正 挪到 INSERT 之前；可行的最小形态就是上面那次换序，无需动 domain 类型、列清单 与 7 条读语句。B 若要做，属于 design/24 §10.5 的独立切片（那时还应删掉这里的 两段式，避免同一事实两处写）。 整链验收（`inbound_voice_gateway_e2e_test.go`，真桥接器 + 真存储 + 真网关 WS）： - 主人私聊语音：第一帧就是 `message_persisted`，带 `voice`，且与 `GET /sessions/{id}/messages` 同一行**逐字相等**；随后的非语音消息帧整键缺席。 - 群聊语音（被点名 ⇒ 走完整回合）：第一帧 `channel_persisted` 带 `voice`， 且与渠道日志读面同一行逐字相等。 - 变异实测（把顺序还原成「先广播、后写列」）：两条用例都 FAIL，输出即报障症状 （帧里 content 是转写文本，却没有 voice 键）。 顺带把网关侧两处描述旧时序的注释改成新事实。 --- .../services/channelbridge/coalesce_batch.go | 21 +- .../channelbridge/inbound_voice_copy.go | 15 + .../inbound_voice_gateway_e2e_test.go | 346 ++++++++++++++++++ internal/services/channelbridge/persist.go | 40 +- internal/services/gateway/broadcast.go | 4 + internal/services/gateway/history_voice.go | 9 +- 6 files changed, 416 insertions(+), 19 deletions(-) create mode 100644 internal/services/channelbridge/inbound_voice_gateway_e2e_test.go diff --git a/internal/services/channelbridge/coalesce_batch.go b/internal/services/channelbridge/coalesce_batch.go index 7dae605..329ee50 100644 --- a/internal/services/channelbridge/coalesce_batch.go +++ b/internal/services/channelbridge/coalesce_batch.go @@ -141,21 +141,32 @@ func dropLeadingSystem(msgs []model.Message) []model.Message { // persistBatchUsers 按**到达顺序**逐条落 user 行（共享 turnID），并把语音事实与用户 // 消息同源落库（design/24 §10.2 第 5 步）。返回刚落库的行，供回合结束后补抽取结论。 +// +// 🔴 顺序（M44，主人 2026-10-26 指令）：通知 = **先写 `voice` 列、再广播**。此前是 +// 「先广播、后写列」，于是这条语音消息的**第一帧**（`message_persisted` / +// `channel_persisted`）里读不到 voice，客户端得靠重连或下一次 REST 拉取才升级 —— +// 而「实时帧一到就出播放器」正是本切片的诉求。换序后帧发出时列已写好。 +// +// 之所以不能把写列真正挪到落库**之前**：`persistMessageVoice` 是按 id 的 UPDATE， +// 行还不存在时它必然 ErrMessageNotFound（`voice` 列也没有随 INSERT 写入的通道， +// 见 design/24 §10.5 的两种口径）。因此可行的最小形态就是「INSERT → UPDATE 列 → +// notify」——语句数不变，只把 notify 从 INSERT 内部挪到最后。 func (b *Bridge) persistBatchUsers(ctx context.Context, parts []batchMessage, target inboundTarget, turnID string) []history.Message { rows := make([]history.Message, 0, len(parts)) for i := range parts { p := &parts[i] msg := p.prepared.item.msg a := p.prepared.item.a + // 非语音消息（p.voiceRef 零值）时这个包装与 notifyPersisted 逐字等价。 + notify := b.persistVoiceFirstNotify(p.voiceRef) var row history.Message if target.private { - row = b.persistToMain(ctx, history.RoleUser, p.prepared.text, "", p.document, p.image, - target.platform, deviceIDFor(msg), adapterName(a), turnID) + row = b.persistMain(ctx, mainRow(history.RoleUser, p.prepared.text, "", p.document, p.image, + target.platform, deviceIDFor(msg), adapterName(a), turnID), notify) } else { - row = b.persistIsolated(ctx, history.RoleUser, p.prepared.text, p.document, p.image, - deviceIDFor(msg), p.prepared.sess, msg.Platform, p.senderName, adapterName(a), turnID) + row = b.persistIsolatedWith(ctx, isolatedRow(history.RoleUser, p.prepared.text, p.document, p.image, + deviceIDFor(msg), p.prepared.sess, msg.Platform, p.senderName, adapterName(a), turnID), notify) } - b.persistMessageVoice(row, p.voiceRef) rows = append(rows, row) } return rows diff --git a/internal/services/channelbridge/inbound_voice_copy.go b/internal/services/channelbridge/inbound_voice_copy.go index f698cae..301d71e 100644 --- a/internal/services/channelbridge/inbound_voice_copy.go +++ b/internal/services/channelbridge/inbound_voice_copy.go @@ -111,6 +111,21 @@ func (b *Bridge) ingestInboundVoiceCopy(ctx context.Context, msg adapter.Unified return turn.Ref } +// persistVoiceFirstNotify 返回一个 notify：**先**把语音事实写进刚落库的那一行，**再**交给 +// 广播（M44，主人 2026-10-26 指令）。顺序就是内容 —— 实时帧上的 `voice` 是**读列**得来的， +// 若先广播后写列，这条语音消息的**第一帧**就永远缺 `voice`，客户端只能等重连或下一次 REST +// 拉取才升级出播放器（安卓 2026-10-26 回执报的正是这个）。 +// +// 只换顺序、不换代价：仍是 INSERT + 一次 UPDATE（与 design/24 §10.5 的第二选项同一手法， +// 与 document 结论的两段式同族）。非语音消息（ref 零值）时 persistMessageVoice 直接返回， +// 因此本包装与 notifyPersisted 逐字等价 —— 普通消息的时序一个字节都没变。 +func (b *Bridge) persistVoiceFirstNotify(ref voiceref.Ref) func(history.Message) { + return func(m history.Message) { + b.persistMessageVoice(m, ref) + b.notifyPersisted(m) + } +} + // persistMessageVoice 把这条语音的来源事实与副本引用写进刚落库的用户消息 `voice` 列 // （design/24 §10.2 第 5 步）。它由 runInboundTurn 在用户消息落库之后调用，因此 // 「引用 + 三态」最终是同一行里的一份事实，客户端不必自己拼。 diff --git a/internal/services/channelbridge/inbound_voice_gateway_e2e_test.go b/internal/services/channelbridge/inbound_voice_gateway_e2e_test.go new file mode 100644 index 0000000..26ba659 --- /dev/null +++ b/internal/services/channelbridge/inbound_voice_gateway_e2e_test.go @@ -0,0 +1,346 @@ +package channelbridge + +// 本文件是 M44 的**跨层整链**验收：真 IM 入站语音 → 真桥接器落库并写 `voice` 列 → 真网关的 +// **实时帧** → 同一个网关的 REST 读面。 +// +// 它钉的是一条**时序**，而不是一个形状：`voice` 在实时帧上是**读列**得来的，所以「先广播、 +// 后写列」会让这条语音消息的**第一帧**缺 `voice`（安卓 2026-10-26 回执：只能等下一次 REST +// 拉取才升级出播放器）。本用例因此断言： +// 1. 入站语音回合发出的**第一条** `message_persisted` 帧里就带 `voice`； +// 2. 该 `voice` 与随后 `GET /sessions/{id}/messages` 同一行**逐字相等**（同一份事实，不是两处各算）； +// 3. 隔离渠道会话那条腿（`channel_persisted`）同样成立，且与渠道日志读面逐字相等。 +// +// 不用「等一会儿再断言」：帧是从真 WS 连接上读下来的，桥接器的持久化钩子接的就是网关的 +// `BroadcastPersisted`（与 cmd/cyrene 组合根的接线逐字同源），因此读到的就是**那一帧**本身。 + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/services/gateway" +) + +// voiceFrameE2E 打包整链环境：真桥接器（假 STT、真音频存储、真历史存储）+ 真网关 +// （同一个历史存储）+ 已接好的持久化钩子。 +type voiceFrameE2E struct { + bridge *Bridge + ts *httptest.Server + tok string +} + +// newVoiceFrameE2E 装配整链环境。历史存储是**同一个句柄**喂给桥接器与网关 —— 被验证的 +// 正是这条边界：桥接器写进那一列的事实，就是实时帧与 REST 读面要投影出来的事实。 +func newVoiceFrameE2E(t *testing.T, sess session.Session) *voiceFrameE2E { + t.Helper() + dataDir := t.TempDir() + uploads := filepath.Join(dataDir, "uploads", "documents") + log := slog.New(slog.NewTextHandler(&safeBuffer{}, nil)) + st := openGatewayDocStore(t, log) + + ag, err := agent.New("测试系统提示词", nil, nil, agent.WithModel(&stubModelAdapter{}, nil)) + if err != nil { + t.Fatalf("构造 agent: %v", err) + } + runner := loop.New(loop.Options{Agent: ag, MaxRounds: 4, Log: log}) + b, err := New(Options{ + Log: log, + Loop: runner, + Sessions: &stubSessions{sess: sess}, + Owner: testOwnerProfile("telegram", "100"), + History: st, + // 真 STT 替身：转写成功 ⇒ 这条消息的 voice 事实是 asr/transcribed。 + VoiceSTT: transcodingSTT("语音内容"), + // 真音频存储（documents.Store 的音频分支）：副本真落盘，`url` 才有意义。 + DocumentStore: documents.StoreConfig{DataDir: dataDir, UploadsDir: uploads}, + }) + if err != nil { + t.Fatalf("构造桥接器: %v", err) + } + b.Add(&fakeAdapter{name: "telegram"}) + + jwt := auth.New(config.AuthConfig{JWTSecret: gatewayDocSecret}) + srv, err := gateway.New( + &config.Config{Auth: config.AuthConfig{AdminUser: "admin", AdminPass: "secret", JWTSecret: gatewayDocSecret}}, + gateway.Deps{Auth: jwt, History: st, DataDir: dataDir}, + ) + if err != nil { + t.Fatalf("构造网关: %v", err) + } + t.Cleanup(srv.Close) + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(ts.Close) + // 🔴 组合根的同一接线（cmd/cyrene/wire_frontend.go：SetOnPersisted(gw.BroadcastPersisted)）。 + // 桥接器「通知」的那一刻，就是网关构造并扇出实时帧的那一刻 —— 断言因此打在真帧上。 + b.SetOnPersisted(srv.BroadcastPersisted) + + tok, err := jwt.Sign("acct", auth.RoleOwner, time.Hour) + if err != nil { + t.Fatalf("签发 token: %v", err) + } + return &voiceFrameE2E{bridge: b, ts: ts, tok: tok} +} + +// dialOwnerWS 连上真网关的 `/ws/chat`（owner token），并吃掉握手回放的 `history` 帧， +// 使后续读取从「本回合的第一帧」开始。 +func (e *voiceFrameE2E) dialOwnerWS(t *testing.T) *websocket.Conn { + t.Helper() + wsURL := "ws" + strings.TrimPrefix(e.ts.URL, "http") + "/ws/chat?token=" + e.tok + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("连接 WS: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + ev := readWSFrame(t, conn) + if ev.Type != "history" { + t.Fatalf("第一个帧 = %q, want history（握手回放）", ev.Type) + } + return conn +} + +// wsFrame 是下行帧的**原始 JSON 视图**：断言「键缺席」与「键存在」的差别只能靠它 +// （解码进结构体看不出两者的区别）。 +type wsFrame struct { + Type string `json:"type"` + SessionID string `json:"sessionId"` + Data json.RawMessage `json:"data"` +} + +// readWSFrame 从连接上读一帧并解到 raw 视图。 +func readWSFrame(t *testing.T, conn *websocket.Conn) wsFrame { + t.Helper() + _ = conn.SetReadDeadline(time.Now().Add(5 * time.Second)) + _, raw, err := conn.ReadMessage() + if err != nil { + t.Fatalf("读帧: %v", err) + } + var ev wsFrame + if err := json.Unmarshal(raw, &ev); err != nil { + t.Fatalf("解码帧 %s: %v", raw, err) + } + return ev +} + +// persistedMessage 从 `message_persisted` / `channel_persisted` 帧的 `data.message` 里取出 +// 原始 JSON 对象（保持键的在场性）。 +func persistedMessage(t *testing.T, ev wsFrame) map[string]json.RawMessage { + t.Helper() + var data struct { + Message map[string]json.RawMessage `json:"message"` + } + if err := json.Unmarshal(ev.Data, &data); err != nil { + t.Fatalf("解码帧 data %s: %v", ev.Data, err) + } + if data.Message == nil { + t.Fatalf("帧 data 里没有 message 对象: %s", ev.Data) + } + return data.Message +} + +// messageRole 取帧内消息的 role。 +func messageRole(t *testing.T, msg map[string]json.RawMessage) string { + t.Helper() + var role string + if err := json.Unmarshal(msg["role"], &role); err != nil { + t.Fatalf("解析帧里的 role: %v", err) + } + return role +} + +// nextPersistedUserFrame 读到**下一条 user 行的持久化帧**（跳过助手回复等其余帧），并返回 +// 中途跳过的帧类型 —— 「语音行的帧最先到」这类顺序断言因此不需要 sleep 或猜测。 +func nextPersistedUserFrame(t *testing.T, conn *websocket.Conn) (wsFrame, []string) { + t.Helper() + var skipped []string + for i := 0; i < 10; i++ { + ev := readWSFrame(t, conn) + switch ev.Type { + case "message_persisted", "channel_persisted": + if messageRole(t, persistedMessage(t, ev)) == history.RoleUser { + return ev, skipped + } + skipped = append(skipped, ev.Type+"("+history.RoleAssistant+")") + default: + skipped = append(skipped, ev.Type) + } + } + t.Fatalf("读了 10 帧仍没等到 user 行的持久化帧（跳过=%v）", skipped) + return wsFrame{}, skipped +} + +// canonicalVoice 把一段 `voice` JSON 归一成「键序由 json.Marshal 的 map 排序决定」的文本。 +// JSON 对象的键无序，因此「逐字相等」在对象层面应按归一化文本判：两侧同一套归一， +// 任何一处多键 / 少键 / 值不同都会让文本不同。 +func canonicalVoice(t *testing.T, raw json.RawMessage) string { + t.Helper() + var v any + if err := json.Unmarshal(raw, &v); err != nil { + t.Fatalf("解析 voice JSON（raw=%s）: %v", raw, err) + } + b, err := json.Marshal(v) + if err != nil { + t.Fatalf("归一 voice JSON（raw=%s）: %v", raw, err) + } + return string(b) +} + +// voiceOfRowByID 在读面的一页里按 id 找到那一行，返回它的 `voice` 原始 JSON。 +func voiceOfRowByID(t *testing.T, rows []map[string]json.RawMessage, id string) json.RawMessage { + t.Helper() + for _, row := range rows { + var gotID string + _ = json.Unmarshal(row["id"], &gotID) + if gotID != id { + continue + } + raw, ok := row["voice"] + if !ok { + t.Fatalf("读面同一行（id=%s）缺 voice 键: %v", id, row) + } + return raw + } + t.Fatalf("读面里没有帧上的那条消息 id=%q（整页=%v）", id, rows) + return nil +} + +// messageIDOf 取帧内消息的 id。 +func messageIDOf(t *testing.T, msg map[string]json.RawMessage) string { + t.Helper() + var id string + if err := json.Unmarshal(msg["id"], &id); err != nil { + t.Fatalf("解析帧里的 id: %v", err) + } + return id +} + +// TestIMInboundVoiceFirstFrameCarriesVoice 是 M44 的整链验收（主人 2026-10-26 指令）： +// 一条**真入站 IM 语音**发出的**第一帧**里就要有 `voice`，且与 REST 读面同一行逐字相等。 +func TestIMInboundVoiceFirstFrameCarriesVoice(t *testing.T) { + t.Parallel() + env := newVoiceFrameE2E(t, session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}) + conn := env.dialOwnerWS(t) + + // 一条可播格式的入站语音：转写成功（假 STT）+ 真副本落盘（真 documents 存储）。 + env.bridge.process(context.Background(), voiceMessage("v.ogg", "audio/ogg", []byte("OGGDATA"))) + + ev, skipped := nextPersistedUserFrame(t, conn) + if len(skipped) != 0 { + t.Fatalf("语音行的帧之前还夹了别的帧 %v，want 它是最先到的那一帧", skipped) + } + if ev.Type != "message_persisted" { + t.Fatalf("主人私聊的帧类型 = %q, want message_persisted", ev.Type) + } + msg := persistedMessage(t, ev) + raw, ok := msg["voice"] + if !ok { + t.Fatalf("🔴 入站语音的**第一帧**缺 data.message.voice（这正是安卓报的缺口）: %s", ev.Data) + } + // 形状逐项落地（取值由下面的逐字比对兜底，这里给可读的失败信息）。 + shape := canonicalVoice(t, raw) + for _, want := range []string{`"source":"asr"`, `"status":"transcribed"`, `"mime":"audio/ogg"`, `"size":7`} { + if !strings.Contains(shape, want) { + t.Fatalf("帧里的 voice = %s, 缺 %s", shape, want) + } + } + if !strings.Contains(shape, `"url":"/api/v1/media/`) { + t.Fatalf("帧里的 voice = %s, 缺服务端拼出的 url", shape) + } + + // 与 REST 读面**同一行**逐字相等：这是本切片的唯一意义（同一份事实，不是两处各算）。 + id := messageIDOf(t, msg) + restVoice := voiceOfRowByID(t, fetchMessages(t, env.ts.URL, history.MainSessionID, env.tok), id) + if got, want := canonicalVoice(t, raw), canonicalVoice(t, restVoice); got != want { + t.Fatalf("帧里的 voice = %s, REST 读面同一行 = %s（两处各算一份就失去意义）", got, want) + } + + // 纯增量不变：紧随其后的一条**非语音**消息，其帧里该键整键缺席（不是 null、不是 {}）。 + env.bridge.process(context.Background(), directMessage()) + plain, _ := nextPersistedUserFrame(t, conn) + plainMsg := persistedMessage(t, plain) + if got, ok := plainMsg["voice"]; ok { + t.Fatalf("非语音消息的帧带上了 voice = %s, want 整键缺席", got) + } +} + +// TestIMInboundIsolatedVoiceFirstFrameCarriesVoice 钉住隔离渠道会话那条腿：群里的语音消息 +// 走 `channel_persisted`，落库点与主人私聊**同一个**，因此首帧同样带 `voice`，且与渠道日志 +// 读面同一行逐字相等。 +func TestIMInboundIsolatedVoiceFirstFrameCarriesVoice(t *testing.T) { + t.Parallel() + env := newVoiceFrameE2E(t, nonOwnerSession()) + conn := env.dialOwnerWS(t) + + // 被点名 ⇒ 值得回复 ⇒ 走完整回合（否则会被最小打扰静默登记，那条路不落语音副本）。 + msg := nonOwnerGroupMsg("", adapter.Mention{UserID: "bot"}) + msg.ContentType = adapter.ContentVoice + msg.Media = []adapter.Media{{ + Kind: adapter.MediaVoice, Name: "v.ogg", MIMEType: "audio/ogg", + Size: int64(len("OGGDATA")), Data: []byte("OGGDATA"), + }} + env.bridge.process(context.Background(), msg) + + ev, skipped := nextPersistedUserFrame(t, conn) + if len(skipped) != 0 { + t.Fatalf("隔离会话语音行的帧之前还夹了别的帧 %v，want 它最先到", skipped) + } + if ev.Type != "channel_persisted" { + t.Fatalf("隔离会话的帧类型 = %q, want channel_persisted", ev.Type) + } + frameMsg := persistedMessage(t, ev) + raw, ok := frameMsg["voice"] + if !ok { + t.Fatalf("🔴 隔离会话的入站语音**第一帧**缺 data.message.voice: %s", ev.Data) + } + // 渠道日志读面（管理面）同一行必须给出同一份事实。 + logRows := fetchChannelLog(t, env.ts.URL, env.tok) + logVoice := voiceOfRowByID(t, logRows, messageIDOf(t, frameMsg)) + if got, want := canonicalVoice(t, raw), canonicalVoice(t, logVoice); got != want { + t.Fatalf("帧里的 voice = %s, 渠道日志同一行 = %s", got, want) + } +} + +// fetchChannelLog 取渠道日志读面的一页（原始 JSON 对象，保持键的在/缺席语义）。 +func fetchChannelLog(t *testing.T, base, token string) []map[string]json.RawMessage { + t.Helper() + req, err := http.NewRequest(http.MethodGet, base+"/api/v1/admin/channels/messages?limit=50", nil) + if err != nil { + t.Fatalf("构造请求: %v", err) + } + req.Header.Set("Authorization", "Bearer "+token) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("请求渠道日志: %v", err) + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("读响应体: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("渠道日志状态 = %d, want 200 (body=%s)", resp.StatusCode, body) + } + var envelope struct { + Data []map[string]json.RawMessage `json:"data"` + } + if err := json.Unmarshal(body, &envelope); err != nil { + t.Fatalf("解码渠道日志: %v (body=%s)", err, body) + } + return envelope.Data +} diff --git a/internal/services/channelbridge/persist.go b/internal/services/channelbridge/persist.go index d49246a..4b3b022 100644 --- a/internal/services/channelbridge/persist.go +++ b/internal/services/channelbridge/persist.go @@ -82,9 +82,16 @@ func (b *Bridge) notifyPersisted(msg history.Message) { // 结构化判据的帧（切片 P0b）—— 因此带判据的帧只可能来自主动推送，**普通消息的帧永远 // 不带它**。 func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, document, image, platform, deviceID, adapterID, turnID string) history.Message { - return b.persistMain(ctx, history.Message{ + return b.persistMain(ctx, mainRow(role, content, mood, document, image, platform, deviceID, adapterID, turnID), b.notifyPersisted) +} + +// mainRow 构造一条写进 Web 主会话的行。它只写一次字段清单：persistToMain 与 +// 入站语音那条「先写列再广播」的路径（coalesce_batch.go 的 persistBatchUsers） +// 共用它，两处的列集合因此不可能分叉。 +func mainRow(role, content, mood, document, image, platform, deviceID, adapterID, turnID string) history.Message { + return history.Message{ SessionID: history.MainSessionID, Role: role, Content: content, Mood: mood, Document: document, Image: image, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, Mode: modeChat, TurnID: turnID, - }, b.notifyPersisted) + } } // persistMain 是「写进主会话」的**唯一**写入口径：persistToMain（普通消息）与 @@ -126,16 +133,29 @@ func (b *Bridge) persistMain(ctx context.Context, msg history.Message, notify fu // 供调用方在回合结束后补回文档抽取结论。未装配历史存储时为空操作， // 因此桥接器在没有存储的情况下行为不变。 func (b *Bridge) persistIsolated(ctx context.Context, role, content, document, image, deviceID string, sess session.Session, platform, senderName, adapterID, turnID string) history.Message { + return b.persistIsolatedWith(ctx, isolatedRow(role, content, document, image, deviceID, sess, platform, senderName, adapterID, turnID), b.notifyPersisted) +} + +// isolatedRow 构造一条写进受限（群聊/陌生人）会话的行。与 mainRow 同一手法：字段清单 +// 只写一次，persistIsolated 与入站语音那条「先写列再广播」的路径共用。 +func isolatedRow(role, content, document, image, deviceID string, sess session.Session, platform, senderName, adapterID, turnID string) history.Message { + return history.Message{ + SessionID: sess.SessionID, Role: role, Content: content, Document: document, Image: image, Platform: platform, + DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, Mode: modeChat, TurnID: turnID, + } +} + +// persistIsolatedWith 是 persistIsolated 的**可注入通知**变体：写入语义逐字相同，只有 +// 「落库成功后怎么通知」由调用方给出（nil = 不通知）。入站语音用它把「先写 `voice` 列、 +// 再广播」这次换序塞进同一个落库点（见 inbound_voice_copy.go 的 persistVoiceFirstNotify）。 +func (b *Bridge) persistIsolatedWith(ctx context.Context, row history.Message, notify func(history.Message)) history.Message { if b.history == nil { return history.Message{} } - full, err := b.history.AppendFull(ctx, history.Message{ - SessionID: sess.SessionID, Role: role, Content: content, Document: document, Image: image, Platform: platform, - DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, Mode: modeChat, TurnID: turnID, - }) + full, err := b.history.AppendFull(ctx, row) if err != nil { - b.log.Warn("persist isolated session", "err", err, "role", role, - "session", sess.SessionID, "platform", platform, "sender_name", senderName, "adapter", adapterID) + b.log.Warn("persist isolated session", "err", err, "role", row.Role, + "session", row.SessionID, "platform", row.Platform, "sender_name", row.SenderName, "adapter", row.AdapterID) return history.Message{} } // 把隔离写入作为 `channel_persisted` 事件广播，让 IM 适配器 @@ -143,6 +163,8 @@ func (b *Bridge) persistIsolated(ctx context.Context, role, content, document, i // 这不是 Web 主会话，因此 Web 聊天没有它的处理器， // 消息本身仍只落在隔离的 sess-xxx 会话（绝不在 // 主/`web` 会话，design/16 §4.6 §3）。 - b.notifyPersisted(full) + if notify != nil { + notify(full) + } return full } diff --git a/internal/services/gateway/broadcast.go b/internal/services/gateway/broadcast.go index 45dae9c..facaec3 100644 --- a/internal/services/gateway/broadcast.go +++ b/internal/services/gateway/broadcast.go @@ -66,6 +66,10 @@ func (s *Server) broadcastPersisted(msg history.Message, proactive bool) { // 「同一行消息在帧里与 GET /sessions/{id}/messages 里」是逐字相同的对象，客户端不必 // 再拉一次 REST 才认出语音气泡。 // +// 帧上的 `voice` 反映的是**本帧发出那一刻**的列值 ⇒ `voice` 列的写入必须先于 notify。 +// 入站语音那条路已由 M44 倒好顺序（channelbridge 的 persistVoiceFirstNotify）， +// 否则这条语音消息的第一帧会读不到 voice。 +// // 代价（实测，见交付报告）：每帧**一次**批量读列（`IN (?)` 一条 SQL，页内 id 去重）， // 与本路径既有的 unreadCountOf 查询同量级；存储未提供该视图时**零**查询、字段缺席。 // 之所以要查库而不是「从 history.Message 直接填」：那个结构体没有 voice 列字段 diff --git a/internal/services/gateway/history_voice.go b/internal/services/gateway/history_voice.go index b043759..78029e5 100644 --- a/internal/services/gateway/history_voice.go +++ b/internal/services/gateway/history_voice.go @@ -26,11 +26,10 @@ package gateway // （`ws_history.go`）。三条收口点共用本文件这一份投影，因此「同一条消息在 REST 读面与 // 实时帧 / 回放里的 `voice` 对象」逐字相同，不会两处各算一份。 // -// ⚠️ 实时帧那条路的**已知边界**（不在本文件能修的范围内）：入站语音的 `voice` 列由 -// 渠道桥在**消息落库之后**回填（channelbridge 的两段式），而 `message_persisted` 那一帧 -// 在**落库那一刻**就发出去了 ⇒ 这一帧读到的列仍是空串、`voice` 依旧缺席；紧随其后的 -// 握手回放 / REST 拉取才带上它。要让「刚落库的那一帧」也带上，须把渠道桥的两段式顺序 -// 倒过来（先写列、再 notify），见交付报告与 Docs §4.4 的同一处如实登记。 +// ⚠️ 实时帧那条路对**时序**有依赖（已由 M44 的跨层修复保证）：`voice` 是**读列**得来的， +// 所以帧必须在 `voice` 列写好之后才发出。渠道桥原来「先 notify、后写列」，导致入站语音的 +// **第一帧**读不到 voice；M44 把它改成「INSERT → 写 voice 列 → notify」（见 channelbridge +// 的 persistBatchUsers / persistVoiceFirstNotify），本文件因此无需再区别对待实时帧与读面。 import ( "context"