feat(gateway): M44 实时帧与握手回放挂上 voice(纯增量)

修实根:`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 读面同一行」;摘掉任一挂载即打红。
This commit is contained in:
AskaEth committed 2026-10-26 18:46:06 +08:00
1 parent af61b8ccfe
commit 5717735da9
4 files changed
+507 -19

No files matched your search

+23 -3
View File
@@ -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"
@@ -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)
}
})
}
+40 -15
View File
@@ -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 {
+8 -1
View File
@@ -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}})
}