From f2e3215c77e6770452d121cc3f458ee05434d5cd Mon Sep 17 00:00:00 2001
From: JianFeeeee <root@qq.com>
Date: Fri, 21 Aug 2026 10:53:38 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20remotedevice=20=E4=BA=8C=E8=BF=9B?=
 =?UTF-8?q?=E5=88=B6=E5=88=86=E5=9D=97=E5=8D=8F=E8=AE=AE(=E5=BD=95?=
 =?UTF-8?q?=E5=83=8F=E5=9B=9E=E4=BC=A0+=E9=9F=B3=E9=A2=91=E4=B8=8B?=
 =?UTF-8?q?=E5=8F=91)=20+=20=E5=B7=A5=E5=85=B7=E6=8F=8F=E8=BF=B0=E5=A2=9E?=
 =?UTF-8?q?=E5=BC=BA?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

回应 GUI c29abe9 需服务端配套项([阻塞]两项 + [中]一项):

1. readFrame 支持 0x2 二进制帧(之前遇 0x2 直接断连):
   - 返回 opcode, 二进制帧上限放宽至 8MB
2. handleWS 二进制聚合协议(设备→网关, 录像回传):
   - cmd_data_start 开启按 req_id 聚合 → 0x2 帧追加 → cmd_data_end 聚合完成
   - 结果存 cmdresult(data_base64 字段), device_ctl_cmdresult 可取回
   - 超限防护(声明 total×2 或硬上限 64MB)
3. PushData 下发协议(网关→设备, 音频/TTS):
   - cmd_speech_start → 0x2 分块(8KB) → cmd_speech_end
   - GUI 端已实现接收侧(speakeruse 播放链路打通)
4. device_ctl_cmdrun 工具描述补齐:
   - screensue 带参示例(<内容>/<秒> <内容>)
   - camerasue 录像说明(cmdresult 含 data_base64)
   - speakeruse 文字朗读

测试: 最小 WS 客户端端到端验证二进制分块上传/PushData 下发/离线报错
---
 internal/plugins/remotedevice/binary_test.go | 297 +++++++++++++++++++
 internal/plugins/remotedevice/device.go      |  10 +-
 internal/plugins/remotedevice/registry.go    | 170 +++++++++--
 3 files changed, 457 insertions(+), 20 deletions(-)
 create mode 100644 internal/plugins/remotedevice/binary_test.go

diff --git a/internal/plugins/remotedevice/binary_test.go b/internal/plugins/remotedevice/binary_test.go
new file mode 100644
index 0000000..e70c927
--- /dev/null
+++ b/internal/plugins/remotedevice/binary_test.go
@@ -0,0 +1,297 @@
+package remotedevice
+
+import (
+	"bufio"
+	"encoding/base64"
+	"encoding/binary"
+	"encoding/json"
+	"math/rand"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+// ===== 测试用最小 WS 客户端（模拟 GUI 设备桥）=====
+
+type testWSClient struct {
+	conn net.Conn
+	rw   *bufio.ReadWriter
+}
+
+func dialTestWS(t *testing.T, url, token string) *testWSClient {
+	t.Helper()
+	req := "GET /api/v1/device/ws?token=" + token + " HTTP/1.1\r\n" +
+		"Host: test\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" +
+		"Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n"
+	conn, err := net.Dial("tcp", strings.TrimPrefix(url, "http://"))
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	if _, err := conn.Write([]byte(req)); err != nil {
+		t.Fatalf("write upgrade: %v", err)
+	}
+	br := bufio.NewReader(conn)
+	resp, err := readLine(br)
+	if err != nil {
+		t.Fatalf("read upgrade resp: %v", err)
+	}
+	if !strings.Contains(resp, "101") {
+		t.Fatalf("expected 101, got %s", resp)
+	}
+	for {
+		line, err := readLine(br)
+		if err != nil {
+			t.Fatalf("read headers: %v", err)
+		}
+		if line == "\r\n" || line == "" {
+			break
+		}
+	}
+	return &testWSClient{conn: conn, rw: &bufio.ReadWriter{Reader: br, Writer: bufio.NewWriter(conn)}}
+}
+
+func readLine(br *bufio.Reader) (string, error) {
+	var sb strings.Builder
+	for {
+		b, err := br.ReadByte()
+		if err != nil {
+			return sb.String(), err
+		}
+		sb.WriteByte(b)
+		if b == '\n' {
+			return sb.String(), nil
+		}
+	}
+}
+
+// sendText 发送客户端文本帧（带掩码，RFC6455 要求客户端帧必须掩码）
+func (c *testWSClient) sendText(payload []byte) {
+	c.sendFrame(0x1, payload)
+}
+
+func (c *testWSClient) sendBinary(payload []byte) {
+	c.sendFrame(0x2, payload)
+}
+
+func (c *testWSClient) sendFrame(opcode byte, payload []byte) {
+	maskKey := make([]byte, 4)
+	rand.Read(maskKey)
+	masked := make([]byte, len(payload))
+	for i := range payload {
+		masked[i] = payload[i] ^ maskKey[i%4]
+	}
+	var hdr []byte
+	hdr = append(hdr, 0x80|opcode)
+	n := len(payload)
+	switch {
+	case n < 126:
+		hdr = append(hdr, 0x80|byte(n))
+	case n <= 0xffff:
+		hdr = append(hdr, 0x80|126)
+		ext := make([]byte, 2)
+		binary.BigEndian.PutUint16(ext, uint16(n))
+		hdr = append(hdr, ext...)
+	default:
+		hdr = append(hdr, 0x80|127)
+		ext := make([]byte, 8)
+		binary.BigEndian.PutUint64(ext, uint64(n))
+		hdr = append(hdr, ext...)
+	}
+	c.rw.Write(hdr)
+	c.rw.Write(maskKey)
+	c.rw.Write(masked)
+	c.rw.Flush()
+}
+
+// readMsg 读一帧（跳过 pong），返回 opcode 与 payload
+func (c *testWSClient) readMsg() (byte, []byte, error) {
+	for {
+		payload, isClose, opcode, err := readFrame(c.rw.Reader)
+		if err != nil || isClose {
+			return 0, nil, err
+		}
+		if opcode == 0xa {
+			continue
+		}
+		return opcode, payload, nil
+	}
+}
+
+func (c *testWSClient) close() { c.conn.Close() }
+
+// ===== 端到端：hello/bind/cmd + 二进制分块回传（录像协议）=====
+
+func TestWSBinaryChunkUpload(t *testing.T) {
+	reg := NewRegistry()
+	token := "test-token-123"
+	reg.SetAcceptToken(func(provided string) bool { return provided == token })
+
+	srv := httptest.NewServer(http.HandlerFunc(reg.ServeWS))
+	defer srv.Close()
+	url := srv.URL
+
+	cli := dialTestWS(t, url, token)
+	defer cli.close()
+
+	// hello 登记
+	cli.sendText([]byte(`{"op":"hello","device":{"device_id":"gui-test","name":"测试机","kind":"computer","caps":["cmd"]}}`))
+	op, payload, err := cli.readMsg()
+	if err != nil {
+		t.Fatalf("read hello_ack: %v", err)
+	}
+	if op != 0x1 {
+		t.Fatalf("expected text frame, got %x", op)
+	}
+	var ack map[string]interface{}
+	json.Unmarshal(payload, &ack)
+	if ack["op"] != "hello_ack" {
+		t.Fatalf("expected hello_ack, got %v", ack)
+	}
+
+	// 模拟设备收到 cmd 后以二进制分块回传（cmd_data_start → 0x2×N → cmd_data_end）
+	videoData := make([]byte, 20000) // 跨多个 8KB 块
+	for i := range videoData {
+		videoData[i] = byte(i % 251)
+	}
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		cli.sendText(mustJSON(map[string]interface{}{
+			"op": "cmd_data_start", "req_id": "req-video-1",
+			"kind": "camera_video", "mime": "video/mp4",
+			"total": len(videoData), "chunk_size": 8192,
+		}))
+		const chunk = 8192
+		for off := 0; off < len(videoData); off += chunk {
+			end := off + chunk
+			if end > len(videoData) {
+				end = len(videoData)
+			}
+			cli.sendBinary(videoData[off:end])
+		}
+		cli.sendText(mustJSON(map[string]interface{}{
+			"op": "cmd_data_end", "req_id": "req-video-1", "status": "ok", "total": len(videoData),
+		}))
+	}()
+
+	// 服务端等待聚合结果（AwaitResult 由 deliverResult 唤醒）
+	res, err := reg.AwaitResult("req-video-1", 5*time.Second)
+	if err != nil {
+		t.Fatalf("await aggregated result: %v", err)
+	}
+	if res["status"] != "ok" {
+		t.Fatalf("expected status ok, got %v", res["status"])
+	}
+	if got, _ := res["size"].(int); got != len(videoData) {
+		t.Fatalf("size mismatch: got %v want %d", res["size"], len(videoData))
+	}
+
+	// 校验 base64 数据完整性
+	stored, ok := reg.GetResult("req-video-1")
+	if !ok {
+		t.Fatal("result not persisted")
+	}
+	b64, _ := stored["data_base64"].(string)
+	if len(b64) == 0 {
+		t.Fatal("data_base64 empty")
+	}
+	decoded, err := base64.StdEncoding.DecodeString(b64)
+	if err != nil {
+		t.Fatalf("decode base64: %v", err)
+	}
+	if string(decoded) != string(videoData) {
+		t.Fatal("decoded data mismatch")
+	}
+}
+
+// ===== 端到端：PushData 下发音频（网关→设备 cmd_speech 协议）=====
+
+func TestWSPushDataAudio(t *testing.T) {
+	reg := NewRegistry()
+	token := "test-token-456"
+	reg.SetAcceptToken(func(provided string) bool { return provided == token })
+
+	srv := httptest.NewServer(http.HandlerFunc(reg.ServeWS))
+	defer srv.Close()
+
+	cli := dialTestWS(t, srv.URL, token)
+	defer cli.close()
+
+	cli.sendText([]byte(`{"op":"hello","device":{"device_id":"audio-dev","name":"音频机","kind":"speaker"}}`))
+	if _, _, err := cli.readMsg(); err != nil {
+		t.Fatalf("read hello_ack: %v", err)
+	}
+
+	audioData := []byte("RIFF....fake-wav-audio-data-for-testing....")
+
+	// 异步下发音频
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- reg.PushData("audio-dev", "req-speech-1", "speech", "audio/wav", audioData)
+	}()
+
+	// 设备侧按协议读取：start 文本帧 → N 个二进制帧 → end 文本帧
+	var start map[string]interface{}
+	var chunks [][]byte
+	var end map[string]interface{}
+	deadline := time.After(5 * time.Second)
+	for end == nil {
+		select {
+		case <-deadline:
+			t.Fatal("timeout reading speech protocol frames")
+		default:
+		}
+		op, payload, err := cli.readMsg()
+		if err != nil {
+			t.Fatalf("read frame: %v", err)
+		}
+		switch op {
+		case 0x1:
+			var msg map[string]interface{}
+			json.Unmarshal(payload, &msg)
+			switch msg["op"] {
+			case "cmd_speech_start":
+				start = msg
+			case "cmd_speech_end":
+				end = msg
+			}
+		case 0x2:
+			chunks = append(chunks, payload)
+		}
+	}
+	if err := <-errCh; err != nil {
+		t.Fatalf("PushData error: %v", err)
+	}
+
+	if start == nil || start["op"] != "cmd_speech_start" {
+		t.Fatal("missing cmd_speech_start")
+	}
+	if start["mime"] != "audio/wav" || start["kind"] != "speech" {
+		t.Fatalf("unexpected start fields: %v", start)
+	}
+	if int(start["total"].(float64)) != len(audioData) {
+		t.Fatalf("total mismatch: %v", start["total"])
+	}
+	if end["req_id"] != "req-speech-1" {
+		t.Fatalf("unexpected end: %v", end)
+	}
+	var got []byte
+	for _, c := range chunks {
+		got = append(got, c...)
+	}
+	if string(got) != string(audioData) {
+		t.Fatalf("audio data mismatch: got %d bytes want %d", len(got), len(audioData))
+	}
+}
+
+// ===== PushData 对离线设备报错 =====
+
+func TestPushDataOfflineDevice(t *testing.T) {
+	reg := NewRegistry()
+	err := reg.PushData("no-such-device", "req-x", "speech", "audio/wav", []byte{1})
+	if err == nil || !strings.Contains(err.Error(), "not online") {
+		t.Fatalf("expected not online error, got %v", err)
+	}
+}
diff --git a/internal/plugins/remotedevice/device.go b/internal/plugins/remotedevice/device.go
index c4c71d3..4547925 100644
--- a/internal/plugins/remotedevice/device.go
+++ b/internal/plugins/remotedevice/device.go
@@ -59,8 +59,12 @@ func (d *devicectlDevice) Tools() []agentIO.ToolDef {
 			Description: "向设备下发命令/操作（异步，accepted=true 后用 device_ctl_cmdresult 轮询结果）。" +
 				"command 支持两类（前缀区分）：\n" +
 				"- shell-cmd: 在设备上执行原生 shell 命令，如 shell-cmd ls -la /tmp\n" +
-				"- homeagent-cmd: 调用设备端 HomeAgent 内置能力，如 homeagent-camerasue（调用用户侧摄像头）、" +
-				"homeagent-screensue（用户侧屏幕显示内容）\n" +
+				"- homeagent-cmd: 调用设备端 HomeAgent 内置能力：\n" +
+				"  · homeagent-screensue <显示内容/HTML> — 用户侧屏幕弹窗显示自定义内容（带参显示，如 homeagent-screensue 会议提醒：三点开会）\n" +
+				"  · homeagent-screensue <秒> <内容> — 显示指定时长后自动关闭；不带参数则常驻\n" +
+				"  · homeagent-camerasue — 抓拍单张 jpeg（结果为 base64 data URL）\n" +
+				"  · homeagent-camerasue <N秒> — 录像 N 秒 mp4（二进制分块回传，cmdresult 含 data_base64 字段）\n" +
+				"  · homeagent-speakeruse <文字> — 设备端 TTS 语音朗读文字\n" +
 				"⚡ 高危：设备必须已授权，且该操作会改变设备行为。" +
 				"返回 accepted=true 表示已下发并等待设备执行，之后可用 device_ctl_cmdresult 查询结果。" +
 				"若设备未授权或离线，返回错误信息。",
@@ -68,7 +72,7 @@ func (d *devicectlDevice) Tools() []agentIO.ToolDef {
 				"type": "object",
 				"properties": map[string]interface{}{
 					"device_id": map[string]interface{}{"type": "string", "description": "目标设备 ID"},
-					"command":   map[string]interface{}{"type": "string", "description": "以 shell-cmd 或 homeagent-cmd 前缀开头。如 shell-cmd pwd、homeagent-camerasue"},
+					"command":   map[string]interface{}{"type": "string", "description": "以 shell-cmd 或 homeagent-cmd 前缀开头。如 shell-cmd pwd、homeagent-screensue 三点开会、homeagent-camerasue 5（录5秒）、homeagent-speakeruse 你好"},
 				},
 				"required": []interface{}{"device_id", "command"},
 			},
diff --git a/internal/plugins/remotedevice/registry.go b/internal/plugins/remotedevice/registry.go
index a196364..4ec477a 100644
--- a/internal/plugins/remotedevice/registry.go
+++ b/internal/plugins/remotedevice/registry.go
@@ -260,6 +260,45 @@ func (r *Registry) PushCmd(deviceID, reqID, command, cmdType string) error {
 	})
 }
 
+// PushData 向设备分块下发二进制数据（网关→设备，如 TTS 音频）。
+// 协议（与 GUI 设备桥协商）：
+//   文本帧 cmd_speech_start {op, req_id, kind, mime, total} → N 个二进制帧(0x2, ≤8KB) → 文本帧 cmd_speech_end {op, req_id}
+// kind 为语义标记（如 speech），mime 为数据 MIME 类型。设备聚合后按自身能力处理（播放等）。
+func (r *Registry) PushData(deviceID, reqID, kind, mime string, data []byte) error {
+	r.mu.RLock()
+	c, ok := r.conns[deviceID]
+	r.mu.RUnlock()
+	if !ok {
+		return fmt.Errorf("device %s not online", deviceID)
+	}
+	if err := writeText(c.w, mustJSON(map[string]interface{}{
+		"op":     "cmd_speech_start",
+		"req_id": reqID,
+		"kind":   kind,
+		"mime":   mime,
+		"total":  len(data),
+	})); err != nil {
+		return fmt.Errorf("push data start: %w", err)
+	}
+	const chunkSize = 8192
+	for off := 0; off < len(data); off += chunkSize {
+		end := off + chunkSize
+		if end > len(data) {
+			end = len(data)
+		}
+		if err := writeBinary(c.w, data[off:end]); err != nil {
+			return fmt.Errorf("push data chunk: %w", err)
+		}
+	}
+	if err := writeText(c.w, mustJSON(map[string]interface{}{
+		"op":     "cmd_speech_end",
+		"req_id": reqID,
+	})); err != nil {
+		return fmt.Errorf("push data end: %w", err)
+	}
+	return nil
+}
+
 // AwaitResult 等待某请求的结果（带超时）。
 func (r *Registry) AwaitResult(reqID string, timeout time.Duration) (map[string]interface{}, error) {
 	ch := make(chan map[string]interface{}, 1)
@@ -342,43 +381,50 @@ func httpUpgrade(w http.ResponseWriter, r *http.Request) (net.Conn, *bufio.ReadW
 	return conn, rw, nil
 }
 
-func readFrame(r *bufio.Reader) ([]byte, bool, error) {
+// readFrame 读取一个 WS 帧。返回 (payload, isClose, err)。
+// opcode: 0x1 文本 / 0x2 二进制（设备→网关大体积数据分块，如录像回传）。
+func readFrame(r *bufio.Reader) ([]byte, bool, byte, error) {
 	b0, err := r.ReadByte()
 	if err != nil {
-		return nil, true, err
+		return nil, true, 0, err
 	}
 	opcode := b0 & 0x0f
 	b1, err := r.ReadByte()
 	if err != nil {
-		return nil, true, err
+		return nil, true, opcode, err
 	}
 	masked := b1&0x80 != 0
 	length := uint64(b1 & 0x7f)
 	if length == 126 {
 		var ext [2]byte
 		if _, err := io.ReadFull(r, ext[:]); err != nil {
-			return nil, true, err
+			return nil, true, opcode, err
 		}
 		length = uint64(binary.BigEndian.Uint16(ext[:]))
 	} else if length == 127 {
 		var ext [8]byte
 		if _, err := io.ReadFull(r, ext[:]); err != nil {
-			return nil, true, err
+			return nil, true, opcode, err
 		}
 		length = binary.BigEndian.Uint64(ext[:])
 	}
-	if length > 1<<20 {
-		return nil, true, fmt.Errorf("frame too large")
+	// 二进制帧允许更大（录像分块聚合，单帧仍限 8MB 防滥用）
+	maxFrame := uint64(1 << 20)
+	if opcode == 0x2 {
+		maxFrame = 8 << 20
+	}
+	if length > maxFrame {
+		return nil, true, opcode, fmt.Errorf("frame too large")
 	}
 	var maskKey [4]byte
 	if masked {
 		if _, err := io.ReadFull(r, maskKey[:]); err != nil {
-			return nil, true, err
+			return nil, true, opcode, err
 		}
 	}
 	payload := make([]byte, length)
 	if _, err := io.ReadFull(r, payload); err != nil {
-		return nil, true, err
+		return nil, true, opcode, err
 	}
 	if masked {
 		for i := range payload {
@@ -386,23 +432,32 @@ func readFrame(r *bufio.Reader) ([]byte, bool, error) {
 		}
 	}
 	switch opcode {
-	case 0x1:
-		return payload, false, nil
+	case 0x1, 0x2:
+		return payload, false, opcode, nil
 	case 0x8:
-		return nil, true, nil
+		return nil, true, opcode, nil
 	case 0xa:
-		return nil, false, nil
+		return nil, false, opcode, nil
 	case 0x9:
-		return nil, false, errPing
+		return nil, false, opcode, errPing
 	default:
-		return nil, false, fmt.Errorf("unsupported opcode %x", opcode)
+		return nil, false, opcode, fmt.Errorf("unsupported opcode %x", opcode)
 	}
 }
 
 var errPing = fmt.Errorf("ping")
 
 func writeText(w *bufio.Writer, payload []byte) error {
-	if err := writeFrameHeader(w, 0x1, len(payload)); err != nil {
+	return writeFrame(w, 0x1, payload)
+}
+
+// writeBinary 发送 WS 二进制帧（0x2）：网关→设备大体积数据（如 TTS 音频）分块下发。
+func writeBinary(w *bufio.Writer, payload []byte) error {
+	return writeFrame(w, 0x2, payload)
+}
+
+func writeFrame(w *bufio.Writer, opcode byte, payload []byte) error {
+	if err := writeFrameHeader(w, opcode, len(payload)); err != nil {
 		return err
 	}
 	if _, err := w.Write(payload); err != nil {
@@ -491,8 +546,12 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) {
 		}
 	}()
 
+	// 二进制分块聚合状态（设备→网关，如录像回传）：
+	// cmd_data_start 开启 → 0x2 帧追加 → cmd_data_end 聚合存入 cmdresult
+	var dataAccum *dataAccumulator
+
 	for {
-		payload, isClose, err := readFrame(rw.Reader)
+		payload, isClose, opcode, err := readFrame(rw.Reader)
 		if err != nil {
 			if err == errPing {
 				if werr := writePong(rw.Writer); werr != nil {
@@ -505,6 +564,23 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) {
 		if isClose {
 			return
 		}
+		if opcode == 0x2 {
+			// 二进制帧：处于聚合状态时追加数据块，否则忽略
+			if dataAccum != nil {
+				dataAccum.chunks = append(dataAccum.chunks, payload)
+				dataAccum.got += len(payload)
+				// 防滥用：超出声明 total 的 2 倍或硬上限 64MB 时放弃聚合
+				limit := int64(dataAccum.total)*2 + 1024
+				if limit < 64<<20 {
+					limit = 64 << 20
+				}
+				if int64(dataAccum.got) > limit {
+					log.Printf("[remotedevice] data accumulation exceeded limit for req %s, dropped", dataAccum.reqID)
+					dataAccum = nil
+				}
+			}
+			continue
+		}
 		var msg map[string]interface{}
 		if err := json.Unmarshal(payload, &msg); err != nil {
 			continue
@@ -563,10 +639,70 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) {
 			if reqID != "" {
 				r.deliverResult(reqID, msg)
 			}
+		case "cmd_data_start":
+			reqID, _ := msg["req_id"].(string)
+			if reqID == "" {
+				continue
+			}
+			total, _ := msg["total"].(float64)
+			kind, _ := msg["kind"].(string)
+			mime, _ := msg["mime"].(string)
+			dataAccum = &dataAccumulator{
+				reqID: reqID,
+				kind:  kind,
+				mime:  mime,
+				total: int(total),
+			}
+		case "cmd_data_end":
+			reqID, _ := msg["req_id"].(string)
+			status, _ := msg["status"].(string)
+			if dataAccum == nil || dataAccum.reqID != reqID {
+				continue
+			}
+			acc := dataAccum
+			dataAccum = nil
+			if status != "ok" {
+				r.SaveResult(reqID, map[string]interface{}{
+					"op": "cmd_result", "req_id": reqID, "status": "error",
+					"error": "device reported transfer failure",
+				})
+				r.deliverResult(reqID, map[string]interface{}{
+					"op": "cmd_result", "req_id": reqID, "status": "error",
+					"error": "device reported transfer failure",
+				})
+				continue
+			}
+			data := make([]byte, 0, acc.got)
+			for _, c := range acc.chunks {
+				data = append(data, c...)
+			}
+			res := map[string]interface{}{
+				"op":       "cmd_result",
+				"req_id":   reqID,
+				"status":   "ok",
+				"kind":     acc.kind,
+				"mime":     acc.mime,
+				"size":     len(data),
+				"expected": acc.total,
+				// base64 编码完整二进制（录像 mp4 等），供 agent/上层取回后解码使用
+				"data_base64": base64.StdEncoding.EncodeToString(data),
+			}
+			r.SaveResult(reqID, res)
+			r.deliverResult(reqID, res)
 		}
 	}
 }
 
+// dataAccumulator 聚合设备→网关的二进制分块传输（如录像回传）。
+type dataAccumulator struct {
+	reqID  string
+	kind   string
+	mime   string
+	total  int
+	chunks [][]byte
+	got    int
+}
+
 func metaFromMsg(msg map[string]interface{}) DeviceMeta {
 	var meta DeviceMeta
 	if d, ok := msg["device"].(map[string]interface{}); ok {