From fad490dca0113fac356c140badeb78627663737c Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Wed, 26 Aug 2026 01:25:40 +0800 Subject: [PATCH] =?UTF-8?q?fix(remotedevice):=20=E5=AA=92=E4=BD=93?= =?UTF-8?q?=E5=9B=9E=E4=BC=A0=E8=90=BD=E7=9B=98=20+=20webui=20SSE=20panic?= =?UTF-8?q?=20+=20GUI=20=E7=9B=B8=E6=9C=BA=E8=B7=A8=E5=B9=B3=E5=8F=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. remotedevice 媒体落盘（核心改动） 设备录像/照片二进制聚合后写入 /device_media/.， cmd_result 返回 file 路径，不再 base64 内联——10s 录像数 MB 的 base64 会撑爆 LLM 上下文与工具结果管道。未配置目录时保持旧内联行为。 新增 TestWSBinaryMediaToFile 覆盖。 2. webui SSE 'send on closed channel' panic（生产单日 4924 次） handleChatEvents 的 defer close(writeCh) 与 Subscribe 回调闭包竞态： handler 退出后总线仍可能异步触发回调向已关闭 channel 发送。 改为 writer goroutine select on done 退出，不 close channel； defer 中等待 writerDone 保证无残余写入。 顺带补 mockPluginMgr.StopAndUnload（ae42e48 接口变更漏改测试）。 3. GUI camerasue 平台分支 ffmpeg 参数原硬编码 Linux v4l2(/dev/video0)，Windows 上必然失败。 现按平台探测：win32=dshow(枚举设备名取第一个视频设备)、 darwin=avfoundation、linux=v4l2；录像编码 Windows 交给 mp4 muxer 默认。 --- cmd/gui/main.js | 75 +++++++++++++------ internal/plugins/remotedevice/binary_test.go | 69 +++++++++++++++++ internal/plugins/remotedevice/plugin.go | 10 +++ internal/plugins/remotedevice/registry.go | 64 +++++++++++++++- internal/plugins/webui/handler.go | 19 +++-- internal/plugins/webui/handler_plugin_test.go | 1 + 6 files changed, 208 insertions(+), 30 deletions(-) diff --git a/cmd/gui/main.js b/cmd/gui/main.js index 8b7f0cb..042b83f 100644 --- a/cmd/gui/main.js +++ b/cmd/gui/main.js @@ -1380,7 +1380,8 @@ function executeHomeagentCmd(capability, reqId) { .split(/[ >\n]/)[0]; switch (name) { case "camerasue": { - // 摄像头：camerasue=抓拍单张；camerasue <秒>=录制 N 秒视频，返回 base64 + // 摄像头：camerasue=抓拍单张；camerasue <秒>=录制 N 秒视频 + // 平台分支：Windows=dshow(设备名自动探测)，macOS=avfoundation，Linux=v4l2 const argStr = String(capability || "") .replace(/^camerasue/, "") .trim(); @@ -1390,24 +1391,54 @@ function executeHomeagentCmd(capability, reqId) { const os = require("os"); const path = require("path"); const fs = require("fs"); + // 探测平台可用的 ffmpeg 输入参数（缓存结果避免重复探测） + let camInput = null; + function resolveCameraInput(cb) { + if (camInput) return cb(camInput); + const plat = process.platform; + if (plat === "win32") { + // dshow：先枚举设备名取第一个视频设备 + cp.execFile( + "ffmpeg", + ["-hide_banner", "-list_devices", "true", "-f", "dshow", "-i", "video= dummy"], + { timeout: 8000 }, + (err, _so, se) => { + const out = String(se || ""); + const m = out.match(/"([^"]+)"\s*\((?:video|默认)|"([^"]+)"[\s\S]{0,200}?\(video/) + || out.match(/"([^"]+)"[^\n]*\(video/i); + const name = m ? (m[1] || m[2]) : null; + if (name) { + camInput = { pre: ["-f", "dshow", "-i", "video=" + name] }; + } else { + camInput = { pre: ["-f", "dshow", "-i", "video=USB Camera"] }; // 常见默认名兑底 + } + cb(camInput); + }, + ); + return; + } + if (plat === "darwin") { + camInput = { pre: ["-f", "avfoundation", "-i", "0:0"] }; // 默认摄像头 + return cb(camInput); + } + camInput = { pre: ["-f", "v4l2", "-i", "/dev/video0"] }; // Linux + return cb(camInput); + } + resolveCameraInput((cam) => { if (isVideo) { // 录像：ffmpeg 录 N 秒 mp4 到临时文件 const outFile = path.join(os.tmpdir(), "ha_cam_" + Date.now() + ".mp4"); const args = [ - "-f", - "v4l2", - "-i", - "/dev/video0", + ...cam.pre, "-t", String(durMatch), "-pix_fmt", "yuv420p", - "-c:v", - "libx264", - "-f", - "mp4", - outFile, ]; + if (process.platform !== "win32") { + args.push("-c:v", "libx264"); // Windows dshow→mp4 由扩展名驱动原生编码器 + } + args.push("-f", "mp4", outFile); cp.execFile( "ffmpeg", args, @@ -1446,19 +1477,16 @@ function executeHomeagentCmd(capability, reqId) { return; } // 抓拍单张 jpeg - const args = [ - "-f", - "v4l2", - "-i", - "/dev/video0", - "-frames:v", - "1", - "-f", - "image2pipe", - "-vcodec", - "mjpeg", - "pipe:1", - ]; + const args = [ + ...cam.pre, + "-frames:v", + "1", + "-f", + "image2pipe", + "-vcodec", + "mjpeg", + "pipe:1", + ]; cp.execFile( "ffmpeg", args, @@ -1483,6 +1511,7 @@ function executeHomeagentCmd(capability, reqId) { ); }, ); + }); // resolveCameraInput 回调闭合 return; } case "screensue": { diff --git a/internal/plugins/remotedevice/binary_test.go b/internal/plugins/remotedevice/binary_test.go index 4b778e1..6145555 100644 --- a/internal/plugins/remotedevice/binary_test.go +++ b/internal/plugins/remotedevice/binary_test.go @@ -9,6 +9,8 @@ import ( "net" "net/http" "net/http/httptest" + "os" + "path/filepath" "strings" "sync" "testing" @@ -207,6 +209,73 @@ func TestWSBinaryChunkUpload(t *testing.T) { } } +// ===== 媒体落盘：SetMediaDir 后 cmd_result 返回 file 路径而非 base64 内联 ===== +func TestWSBinaryMediaToFile(t *testing.T) { + reg := NewRegistry() + token := "test-token-123" + reg.SetAcceptToken(func(provided string) bool { return provided == token }) + mediaDir := t.TempDir() + reg.SetMediaDir(mediaDir) + + 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":"gui-media","name":"媒体机","kind":"computer","caps":["cmd"]}}`)) + if _, _, err := cli.readMsg(); err != nil { // hello_ack + t.Fatalf("read hello_ack: %v", err) + } + + videoData := make([]byte, 30000) + for i := range videoData { + videoData[i] = byte(i % 253) + } + go func() { + time.Sleep(50 * time.Millisecond) + cli.sendText(mustJSON(map[string]interface{}{ + "op": "cmd_data_start", "req_id": "req-file-1", + "kind": "camera_video", "mime": "video/mp4", + "total": len(videoData), + })) + 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-file-1", "status": "ok", + })) + }() + + res, err := reg.AwaitResult("req-file-1", 5*time.Second) + if err != nil { + t.Fatalf("await result: %v", err) + } + // 落盘模式：file 字段存在且内容一致；不应再有 data_base64 + fp, ok := res["file"].(string) + if !ok || fp == "" { + t.Fatalf("expected file path in result, got %v", res) + } + if _, hasB64 := res["data_base64"]; hasB64 { + t.Fatal("data_base64 should be absent in file mode") + } + if want := filepath.Join(mediaDir, "req-file-1.mp4"); fp != want { + t.Fatalf("file path = %s, want %s", fp, want) + } + got, err := os.ReadFile(fp) + if err != nil { + t.Fatalf("read media file: %v", err) + } + if string(got) != string(videoData) { + t.Fatal("media file content mismatch") + } +} + // ===== 端到端：PushData 下发音频（网关→设备 cmd_speech 协议）===== func TestWSPushDataAudio(t *testing.T) { diff --git a/internal/plugins/remotedevice/plugin.go b/internal/plugins/remotedevice/plugin.go index 2c764c6..7193c5f 100644 --- a/internal/plugins/remotedevice/plugin.go +++ b/internal/plugins/remotedevice/plugin.go @@ -8,6 +8,7 @@ import ( "fmt" "log" "net/http" + "path/filepath" "strings" "sync" "time" @@ -104,6 +105,15 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { log.Printf("[remotedevice] register devicectl channel: %v", err) } + // ---- 媒体落盘目录：/device_media ---------------- + // 设备回传的录像/照片等二进制聚合后写入此目录，cmd_result 返回 file 路径， + // 避免 base64 内联撑爆 LLM 上下文。目录由 logManager/运维定期清理。 + if dataDir, err := s.Settings().GetCore("daemon.data_dir"); err == nil { + if dd, ok := dataDir.(string); ok && dd != "" { + p.registry.SetMediaDir(filepath.Join(dd, "device_media")) + } + } + // ---- 设备主动上报事件 → agent 注入 ---------------- // 摄像头发现异常/传感器报警等场景：设备经 WS op=event 上报， // 插件将其格式化为文本经 SDK InjectText 异步注入 agent（source=device/{id}， diff --git a/internal/plugins/remotedevice/registry.go b/internal/plugins/remotedevice/registry.go index f2f57b6..bc8db01 100644 --- a/internal/plugins/remotedevice/registry.go +++ b/internal/plugins/remotedevice/registry.go @@ -11,6 +11,8 @@ import ( "log" "net" "net/http" + "os" + "path/filepath" "strings" "sync" "time" @@ -49,6 +51,7 @@ type Registry struct { acceptFn func(token string) bool cmdPending map[string]chan map[string]interface{} // reqID -> 结果 channel results map[string]resultEntry // reqID -> 已留档结果 + mediaDir string // 设备回传媒体落盘目录；空则退化为 base64 内联 } // resultEntry 保存一次 cmdrun 的结果（供 device_ctl_cmdresult 查询）。 @@ -120,6 +123,43 @@ func deviceSupportsTool(caps []string, tool string) bool { return !hasKnown // 未声明任何已知能力 → 全能力兼容 } +// SetMediaDir 设置设备回传媒体的落盘目录。 +// 非空时 cmd_data_end 聚合完成后写入该目录，cmd_result 返回 file 路径 +// （大体积 base64 内联会撑爆 LLM 上下文与工具结果管道）；空则保持旧的内联行为。 +func (r *Registry) SetMediaDir(dir string) { + r.mu.Lock() + r.mediaDir = dir + r.mu.Unlock() +} + +// mediaExt 按 mime/kind 推断扩展名。 +func mediaExt(mime, kind string) string { + m := strings.ToLower(mime) + switch { + case strings.Contains(m, "mp4"): + return ".mp4" + case strings.Contains(m, "webm"): + return ".webm" + case strings.Contains(m, "jpeg"), strings.Contains(m, "jpg"): + return ".jpg" + case strings.Contains(m, "png"): + return ".png" + case strings.Contains(m, "wav"): + return ".wav" + case strings.Contains(m, "mpeg"), strings.Contains(m, "mp3"): + return ".mp3" + } + k := strings.ToLower(kind) + if strings.Contains(k, "video") { + return ".mp4" + } + if strings.Contains(k, "image") || strings.Contains(k, "camera_photo") { + return ".jpg" + } + return ".bin" +} + +// NewRegistry 返回初始化后的设备注册表。 func NewRegistry() *Registry { return &Registry{ devices: make(map[string]*DeviceMeta), @@ -743,8 +783,28 @@ func (r *Registry) handleWS(conn net.Conn, rw *bufio.ReadWriter) { "mime": acc.mime, "size": len(data), "expected": acc.total, - // base64 编码完整二进制（录像 mp4 等），供 agent/上层取回后解码使用 - "data_base64": base64.StdEncoding.EncodeToString(data), + } + // 媒体落盘模式：写入 /.，cmd_result 返回 file 路径。 + // 大体积 base64 内联会撑爆 LLM 上下文（一段 10s 录像即数 MB）， + // agent 应拿路径后用 files/describe_image/ocr 等工具消费。 + r.mu.RLock() + mediaDir := r.mediaDir + r.mu.RUnlock() + if mediaDir != "" { + if err := os.MkdirAll(mediaDir, 0755); err == nil { + fp := filepath.Join(mediaDir, reqID+mediaExt(acc.mime, acc.kind)) + if werr := os.WriteFile(fp, data, 0644); werr == nil { + res["file"] = fp + } else { + log.Printf("[remotedevice] media write %s: %v", fp, werr) + } + } else { + log.Printf("[remotedevice] media dir %s: %v", mediaDir, err) + } + } + // 未配置落盘目录时保持旧行为：base64 内联返回（小体积数据仍可用） + if _, hasFile := res["file"]; !hasFile { + res["data_base64"] = base64.StdEncoding.EncodeToString(data) } r.SaveResult(reqID, res) r.deliverResult(reqID, res) diff --git a/internal/plugins/webui/handler.go b/internal/plugins/webui/handler.go index a877085..f2bbcb3 100644 --- a/internal/plugins/webui/handler.go +++ b/internal/plugins/webui/handler.go @@ -1412,18 +1412,26 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) { ticker := time.NewTicker(15 * time.Second) defer ticker.Stop() + // writeCh 不 close：Subscribe 回调闭包持有它，handler 退出后回调仍可能被 + // 总线异步触发，close 后再发送会 panic（send on closed channel，生产日志中 + // 单日数千次）。writer goroutine 通过 done 退出；发送侧 select on done 防泄漏。 writeCh := make(chan string, 64) - defer close(writeCh) - + writerDone := make(chan struct{}) go func() { defer func() { if r := recover(); r != nil { log.Printf("[SSE] writer panic: %v", r) } + close(writerDone) }() - for line := range writeCh { - fmt.Fprintf(w, "%s\n", line) - flusher.Flush() + for { + select { + case line := <-writeCh: + fmt.Fprintf(w, "%s\n", line) + flusher.Flush() + case <-done: + return + } } }() @@ -1501,6 +1509,7 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) { for _, unsub := range unsubs { unsub() } + <-writerDone // 等 writer 退出，保证 handler 返回后无残余写入 }() for { select { diff --git a/internal/plugins/webui/handler_plugin_test.go b/internal/plugins/webui/handler_plugin_test.go index 15e6771..e768ddd 100644 --- a/internal/plugins/webui/handler_plugin_test.go +++ b/internal/plugins/webui/handler_plugin_test.go @@ -52,6 +52,7 @@ func (m *mockPluginMgr) ReloadPlugins() (string, error) { return "reloaded", nil } func (m *mockPluginMgr) ReloadOne(name string) error { return nil } +func (m *mockPluginMgr) StopAndUnload(name string) error { return nil } func (m *mockPluginMgr) PluginMetas() map[string]sdk.PluginMeta { return nil }