Cyrene_v2/internal/services/channelbridge/inbound_voice_gateway_e2e_test.go

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
}