sundynix-agentix/sundynix-gateway/internal/voice/tts_frame.go

package voice

import (
	"bytes"
	"compress/gzip"
	"encoding/binary"
	"fmt"
	"io"
)

// 火山「双向流式 TTS V3」二进制帧编解码。与 ASR(frame.go)是**不同族**协议：
// 这一族带「事件号 + 会话 ID + gzip payload」，是全双工事件协议（StartConnection/StartSession/
// TaskRequest/FinishSession…）。字节级布局从官方参考实现核实（generate_header / parse_response）：
//
//	byte0 = (version<<4)|headerSize          version=1, headerSize=1(=4字节)
//	byte1 = (msgType<<4)|flags               flags 含 ttsFlagWithEvent 表示带事件号
//	byte2 = (serialization<<4)|compression   请求一律 JSON+GZIP；音频响应为 raw
//	byte3 = 0x00                             保留
//	[event  4B 大端]        flags 带 withEvent 时存在
//	[会话级事件]：sessionIdLen 4B 大端 + sessionId
//	payloadLen 4B 大端 + payload(gzip)
//
// 连接级事件(StartConnection=1 / FinishConnection=2)无会话 ID；会话级(100/200/102…)带会话 ID。

// 消息类型（byte1 高 4bit）。
const (
	ttsClientFull  byte = 0b0001 // 客户端完整请求（JSON）
	ttsClientAudio byte = 0b0010 // 客户端纯音频（TTS 上行不用；保留）
	ttsServerFull  byte = 0b1001 // 服务端完整响应
	ttsServerAck   byte = 0b1011 // 服务端 ACK（音频帧走这个）
	ttsServerErr   byte = 0b1111 // 服务端错误
)

// flags（byte1 低 4bit）。
const (
	ttsFlagNone      byte = 0b0000
	ttsFlagNegSeq    byte = 0b0010 // 带序列号
	ttsFlagWithEvent byte = 0b0100 // 带事件号
)

// 序列化 / 压缩（byte2）。
const (
	ttsSerialNone byte = 0b0000
	ttsSerialJSON byte = 0b0001
	ttsCompNone   byte = 0b0000
	ttsCompGzip   byte = 0b0001
)

// 事件号（客户端发 / 服务端回）。
const (
	evStartConnection  int32 = 1
	evFinishConnection int32 = 2
	evConnectionStarted int32 = 50
	evConnectionFailed  int32 = 51
	evConnectionFinished int32 = 52

	evStartSession    int32 = 100
	evFinishSession   int32 = 102
	evSessionStarted  int32 = 150
	evSessionFinished int32 = 152
	evSessionFailed   int32 = 153

	evTaskRequest int32 = 200 // 客户端发文本（TTS 逐句推）

	evTTSSentenceStart int32 = 350
	evTTSSentenceEnd   int32 = 351
	evTTSResponse      int32 = 352 // 服务端回音频帧
)

func gzipBytes(b []byte) []byte {
	var buf bytes.Buffer
	w := gzip.NewWriter(&buf)
	_, _ = w.Write(b)
	_ = w.Close()
	return buf.Bytes()
}

func gunzip(b []byte) ([]byte, error) {
	r, err := gzip.NewReader(bytes.NewReader(b))
	if err != nil {
		return nil, err
	}
	defer r.Close()
	return io.ReadAll(r)
}

func ttsHeader(msgType, flags, serial, comp byte) []byte {
	return []byte{
		(0b0001 << 4) | 0b0001, // version=1, headerSize=1
		(msgType << 4) | (flags & 0x0F),
		(serial << 4) | comp,
		0x00,
	}
}

// connEventFrame 组连接级事件帧（无会话 ID），payload 为 JSON（gzip）。
func connEventFrame(event int32, jsonPayload []byte) []byte {
	var b bytes.Buffer
	b.Write(ttsHeader(ttsClientFull, ttsFlagWithEvent, ttsSerialJSON, ttsCompGzip))
	_ = binary.Write(&b, binary.BigEndian, event)
	pl := gzipBytes(jsonPayload)
	_ = binary.Write(&b, binary.BigEndian, uint32(len(pl)))
	b.Write(pl)
	return b.Bytes()
}

// sessionEventFrame 组会话级事件帧（带会话 ID），payload 为 JSON（gzip）。
func sessionEventFrame(event int32, sessionID string, jsonPayload []byte) []byte {
	var b bytes.Buffer
	b.Write(ttsHeader(ttsClientFull, ttsFlagWithEvent, ttsSerialJSON, ttsCompGzip))
	_ = binary.Write(&b, binary.BigEndian, event)
	_ = binary.Write(&b, binary.BigEndian, uint32(len(sessionID)))
	b.WriteString(sessionID)
	pl := gzipBytes(jsonPayload)
	_ = binary.Write(&b, binary.BigEndian, uint32(len(pl)))
	b.Write(pl)
	return b.Bytes()
}

// ttsResponse 是解析后的服务端帧。
type ttsResponse struct {
	MsgType byte
	Event   int32
	Payload []byte // 音频帧=raw PCM；JSON 事件=JSON 原文；错误=错误 JSON
	Code    uint32 // 错误码（MsgType==ttsServerErr 时）
	IsAudio bool
}

// parseTTSFrame 解析服务端帧（对齐参考实现 parse_response）。
func parseTTSFrame(data []byte) (*ttsResponse, error) {
	if len(data) < 4 {
		return nil, fmt.Errorf("帧过短 %d", len(data))
	}
	headerSize := int(data[0] & 0x0F)
	msgType := data[1] >> 4
	flags := data[1] & 0x0F
	comp := data[2] & 0x0F
	off := headerSize * 4
	if off > len(data) {
		return nil, fmt.Errorf("头长越界")
	}
	res := &ttsResponse{MsgType: msgType}

	if msgType == ttsServerErr {
		if len(data) < off+8 {
			return nil, fmt.Errorf("错误帧过短")
		}
		res.Code = binary.BigEndian.Uint32(data[off : off+4])
		sz := binary.BigEndian.Uint32(data[off+4 : off+8])
		res.Payload = clampTail(data, off+8, sz)
		return res, nil
	}

	p := off
	if flags&ttsFlagNegSeq != 0 { // 跳过序列号
		p += 4
	}
	if flags&ttsFlagWithEvent != 0 {
		if len(data) < p+4 {
			return nil, fmt.Errorf("事件号越界")
		}
		res.Event = int32(binary.BigEndian.Uint32(data[p : p+4]))
		p += 4
	}
	// 会话 ID（有符号长度；服务端正常响应都带）。
	if len(data) >= p+4 {
		sidLen := int32(binary.BigEndian.Uint32(data[p : p+4]))
		p += 4
		if sidLen > 0 && len(data) >= p+int(sidLen) {
			p += int(sidLen)
		}
	}
	if len(data) < p+4 {
		return res, nil // 无 payload 的控制帧（如 SessionFinished 可能无体）
	}
	sz := binary.BigEndian.Uint32(data[p : p+4])
	p += 4
	payload := clampTail(data, p, sz)

	// 音频响应（TTSResponse/SERVER_ACK 且非 JSON）→ raw PCM，按压缩位解压。
	res.IsAudio = res.Event == evTTSResponse || msgType == ttsServerAck
	if comp == ttsCompGzip && len(payload) > 0 {
		if dec, err := gunzip(payload); err == nil {
			payload = dec
		}
	}
	res.Payload = payload
	return res, nil
}

// clampTail 从 off 起最多取 sz 字节（防伪造长度越界）。
func clampTail(data []byte, off int, sz uint32) []byte {
	if off > len(data) {
		return nil
	}
	end := off + int(sz)
	if end > len(data) || sz == 0 {
		end = len(data)
	}
	return data[off:end]
}