Compare commits
3
Commits
0ec9cbe32b
...
fb55769cd7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fb55769cd7 | ||
|
|
b955a27479 | ||
|
|
5717735da9 |
No files matched your search
@@ -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
|
||||
|
||||
@@ -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 在用户消息落库之后调用,因此
|
||||
// 「引用 + 三态」最终是同一行里的一份事实,客户端不必自己拼。
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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,29 @@ 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 才认出语音气泡。
|
||||
//
|
||||
// 帧上的 `voice` 反映的是**本帧发出那一刻**的列值 ⇒ `voice` 列的写入必须先于 notify。
|
||||
// 入站语音那条路已由 M44 倒好顺序(channelbridge 的 persistVoiceFirstNotify),
|
||||
// 否则这条语音消息的第一帧会读不到 voice。
|
||||
//
|
||||
// 代价(实测,见交付报告):每帧**一次**批量读列(`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)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -20,13 +20,20 @@ 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` 对象」逐字相同,不会两处各算一份。
|
||||
//
|
||||
// ⚠️ 实时帧那条路对**时序**有依赖(已由 M44 的跨层修复保证):`voice` 是**读列**得来的,
|
||||
// 所以帧必须在 `voice` 列写好之后才发出。渠道桥原来「先 notify、后写列」,导致入站语音的
|
||||
// **第一帧**读不到 voice;M44 把它改成「INSERT → 写 voice 列 → notify」(见 channelbridge
|
||||
// 的 persistBatchUsers / persistVoiceFirstNotify),本文件因此无需再区别对待实时帧与读面。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"strings"
|
||||
|
||||
voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice"
|
||||
@@ -65,24 +72,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 +107,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 {
|
||||
|
||||
@@ -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}})
|
||||
}
|
||||
Reference in new issue
Block a user