diff --git a/cmd/waiter/device.go b/cmd/waiter/device.go index 60dcb15..a58c7f1 100644 --- a/cmd/waiter/device.go +++ b/cmd/waiter/device.go @@ -72,6 +72,11 @@ func startDeviceBridge(addr, token string) error { cmdRouter.Dispatch(reqID, command) }) + // agent 主动投递（output_send__device/）→ 终端显示。 + // 设备侧参考实现：文本/结构化直接打出来；二进制负载走 OnData（TTS 音频等）。 + bridge.OnPush(func(reqID, typ, payload, meta string) { + printlnC("cyan", fmt.Sprintf("[push:%s] %s", typ, payload)) + }) if err := bridge.Start(); err != nil { return fmt.Errorf("device bridge: %w", err) } diff --git a/internal/agent/core/device_tool_auth_test.go b/internal/agent/core/device_tool_auth_test.go new file mode 100644 index 0000000..0683785 --- /dev/null +++ b/internal/agent/core/device_tool_auth_test.go @@ -0,0 +1,92 @@ +package core + +// 设备类工具的**授权闸**：设备指令类工具走的是工具面，而 AllowedOutputs 只作用于 +// output_send__<通道> —— 不补闸的话"授权"对指令类完全无效（驻留子拿到 +// device_ctl_cmdrun 就能指挥任意设备）。这里按目标设备的通道名 device/ 查同一道闸。 + +import ( + "strings" + "testing" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" +) + +// registerFakeDevice 注册一台假设备，带一个"需要 device_id 的指令类工具"和一个无参枚举工具。 +func registerFakeDevice(t *testing.T, a *Agent, name string, called *[]string) { + t.Helper() + dev := &mockOutputDevice{ + name: name, + caps: agentIO.CapStructured, + tools: []agentIO.ToolDef{ + {Name: "device_ctl_cmdrun", Description: "在设备上执行命令"}, + {Name: "devicedetect", Description: "枚举设备"}, + }, + toolFn: func(tool string, args map[string]interface{}) (interface{}, error) { + if called != nil { + *called = append(*called, tool) + } + return "ok:" + tool, nil + }, + } + if err := a.io.RegisterDevice(dev); err != nil { + t.Fatalf("注册测试设备 %s 失败: %v", name, err) + } +} + +func deviceToolCall(name string, args map[string]interface{}) agentAPI.ToolCall { + return agentAPI.ToolCall{ID: "call_1", Name: name, Arguments: args} +} + +// 完整授权（根 agent 默认）：设备指令工具照常可用。 +func TestDeviceToolAuth_RootHasFullGrant(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + var called []string + registerFakeDevice(t, a, "devicectl", &called) + + got := a.executeToolCall(deviceToolCall("device_ctl_cmdrun", map[string]interface{}{ + "device_id": "pc-1", "command": "ls", + }), "cli") + if !strings.Contains(got, "ok:device_ctl_cmdrun") { + t.Fatalf("根 agent 应可指挥任意设备，实际: %s", got) + } +} + +// 收窄授权（驻留子）：只授权了 device/ok-1，指挥别的设备必须被拒，且**不落到设备**。 +func TestDeviceToolAuth_NarrowedGrantRefusesOtherDevice(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + a.allowedOutputs = []string{"device/ok-1"} + var called []string + registerFakeDevice(t, a, "devicectl", &called) + + got := a.executeToolCall(deviceToolCall("device_ctl_cmdrun", map[string]interface{}{ + "device_id": "other-2", "command": "rm -rf /", + }), "cli") + if !strings.Contains(got, "未授权") { + t.Fatalf("未授权设备应被拒，实际: %s", got) + } + if len(called) != 0 { + t.Fatalf("被拒的调用不得落到设备，实际执行了 %v", called) + } + + // 已授权的设备照常可用 + got = a.executeToolCall(deviceToolCall("device_ctl_cmdrun", map[string]interface{}{ + "device_id": "ok-1", "command": "ls", + }), "cli") + if !strings.Contains(got, "ok:device_ctl_cmdrun") { + t.Fatalf("已授权设备应可用，实际: %s", got) + } +} + +// 无参枚举类（devicedetect）不受闸门影响：它不指向具体设备。 +func TestDeviceToolAuth_EnumerationNotGated(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + a.allowedOutputs = []string{"device/ok-1"} + var called []string + registerFakeDevice(t, a, "devicectl", &called) + + got := a.executeToolCall(deviceToolCall("devicedetect", map[string]interface{}{}), "cli") + if !strings.Contains(got, "ok:devicedetect") { + t.Fatalf("枚举类工具不应被设备授权闸拦，实际: %s", got) + } +} diff --git a/internal/agent/core/toolcall.go b/internal/agent/core/toolcall.go index c948611..7bf4a1e 100644 --- a/internal/agent/core/toolcall.go +++ b/internal/agent/core/toolcall.go @@ -96,6 +96,18 @@ func (a *Agent) executeToolCallInner(tc agentAPI.ToolCall, channel string) strin } } + // 设备类工具的**授权闸**（最小授权的缺口在这里）。 + // + // 设备指令类工具（device_ctl_cmdrun/screensee/computeruse/...）走的是工具面， + // 而 AllowedOutputs 只作用于 output_send__<通道> —— 于是"授权"对指令类工具完全无效： + // 驻留子只要拿到 device_ctl_cmdrun 就能指挥**任意**设备。 + // 这里按目标设备的通道名 device/ 查同一道闸：父授权了哪台设备，才允许指挥哪台。 + if _, isDeviceTool := a.io.DeviceOfTool(tc.Name); isDeviceTool { + if id, _ := tc.Arguments["device_id"].(string); id != "" && !a.IsOutputAllowed("device/"+id) { + return fmt.Sprintf("设备 [%s] 未授权给本 agent（可用设备见 output_list_channels 的 device/ 通道，或 devicedetect）", id) + } + } + if a.tracker != nil { a.tracker.PreAction(tc.Name) } diff --git a/internal/agent/io/channel.go b/internal/agent/io/channel.go index 16e89a0..1f11fbb 100644 --- a/internal/agent/io/channel.go +++ b/internal/agent/io/channel.go @@ -507,6 +507,23 @@ func (m *IOManager) GetAllTools() []ToolDef { return tools } +// DeviceOfTool 返回提供该工具的**设备/输出通道名**（设备类工具才有）。 +// +// 用途：设备类工具（device_ctl_*/screensee/computeruse/...）需要按"目标设备" +// 做授权判断，调用方得先知道这个工具属于哪个设备通道。 +func (m *IOManager) DeviceOfTool(name string) (string, bool) { + m.mu.RLock() + defer m.mu.RUnlock() + for _, dev := range m.devices { + for _, t := range dev.Tools() { + if t.Name == name { + return dev.Name(), true + } + } + } + return "", false +} + func (m *IOManager) ExecuteTool(name string, args map[string]interface{}) (ret interface{}, err error) { m.mu.RLock() type nameDevice struct { diff --git a/internal/devicebridge/client/bridge.go b/internal/devicebridge/client/bridge.go index be8b275..af74ebe 100644 --- a/internal/devicebridge/client/bridge.go +++ b/internal/devicebridge/client/bridge.go @@ -22,6 +22,15 @@ type CmdResultHandler func(reqID, status, output, errMsg string) // DataHandler 是二进制数据接收回调（如 TTS 音频）。 type DataHandler func(reqID, kind, mime string, data []byte) +// PushHandler 接收 agent **主动投递**给本设备的内容。 +// +// 与 cmd 的区别：cmd 是"让设备做一件事"（请求-响应，结果要回传）， +// push 是"把这段内容交给设备"（agent 经 output_send__device/ 发起， +// 一种单向投递）。宿主按自己的形态落地：终端打出来、音箱念出来、屏幕显示。 +// +// typ: text / structured / image / file / audio（二进制走 DataHandler，不走这里） +type PushHandler func(reqID, typ, payload, meta string) + // Bridge 是设备桥客户端核心结构体。 // 管理 WebSocket 连接、消息路由、心跳保活和命令分发。 // 授权状态由设备端本地存储（客户端鉴权），服务端不存储； @@ -46,6 +55,7 @@ type Bridge struct { cmdHandler BridgeCmdHandler resultHandler CmdResultHandler dataHandler DataHandler + pushHandler PushHandler // 二进制数据聚合（服务端→设备，如 TTS 音频） speechAccum *speechBuffer @@ -157,6 +167,13 @@ func (b *Bridge) OnData(handler DataHandler) { b.dataHandler = handler } +// OnPush 注册 agent 主动投递内容的回调（服务端 op=push）。 +func (b *Bridge) OnPush(handler PushHandler) { + b.mu.Lock() + defer b.mu.Unlock() + b.pushHandler = handler +} + // SetPingInterval 设置心跳间隔（默认 30 秒）。 func (b *Bridge) SetPingInterval(d time.Duration) { b.mu.Lock() @@ -478,6 +495,25 @@ func (b *Bridge) handleMessage(msg map[string]interface{}) { dh(reqID, acc.kind, acc.mime, data) } + case "push": + // agent 主动投递（output_send__device/）。二进制负载走 + // cmd_speech_* → DataHandler，这里只处理文本/结构化。 + reqID, _ := msg["req_id"].(string) + typ, _ := msg["type"].(string) + payload, _ := msg["payload"].(string) + meta, _ := msg["meta"].(string) + if typ == "" { + typ = "text" + } + b.mu.RLock() + ph := b.pushHandler + b.mu.RUnlock() + if ph != nil { + ph(reqID, typ, payload, meta) + } else { + log.Printf("[devicebridge] push req=%s type=%s payload=%s", reqID, typ, truncateString(payload, 120)) + } + default: log.Printf("[devicebridge] unhandled op=%s", op) } diff --git a/internal/plugin/registry.go b/internal/plugin/registry.go index 9682f61..2a39c17 100644 --- a/internal/plugin/registry.go +++ b/internal/plugin/registry.go @@ -318,6 +318,15 @@ func (r *Registry) buildSDK(name string) *sdk.PluginSDK { return nil } + regOutputUnreg := func(chName string) error { + if r.iom == nil { + return nil + } + r.iom.UnregisterDevice(chName) + r.forgetChannel(name, chName) + return nil + } + regInput := func(chName string, def sdk.ChannelDef) error { if r.iom == nil { return nil @@ -334,18 +343,19 @@ func (r *Registry) buildSDK(name string) *sdk.PluginSDK { EventBus: r.evBus, // 带 media 的包装：插件提交的三元组/文档/文本事件里的媒体会落进 CAS // 并挂上引用。传入插件名仅用于日志溯源（哪个插件写的媒体）。 - Memory: sdk.NewGraphMemoryWithMedia(name, r.memDB, r.mediaStore), - TextMemory: sdk.NewTextMemoryWithMedia(name, r.textMem, r.mediaStore), - DocMemory: sdk.NewDocMemoryWithMedia(name, r.docStore, r.mediaStore), - Knowledge: sdk.NewKnowledge(r.ks), - LLM: sdk.NewLLM(r.mgr, r.cfgReg, r.lua, r.baseKey), - Settings: sett, - RegTool: regTool, - RegStage: regStage, - RegAPI: regAPI, - RegOutput: regOutput, - RegInput: regInput, - PluginMgr: r, + Memory: sdk.NewGraphMemoryWithMedia(name, r.memDB, r.mediaStore), + TextMemory: sdk.NewTextMemoryWithMedia(name, r.textMem, r.mediaStore), + DocMemory: sdk.NewDocMemoryWithMedia(name, r.docStore, r.mediaStore), + Knowledge: sdk.NewKnowledge(r.ks), + LLM: sdk.NewLLM(r.mgr, r.cfgReg, r.lua, r.baseKey), + Settings: sett, + RegTool: regTool, + RegStage: regStage, + RegAPI: regAPI, + RegOutput: regOutput, + RegOutputUnreg: regOutputUnreg, + RegInput: regInput, + PluginMgr: r, Status: r.status, Supervisor: r.sup, @@ -570,6 +580,24 @@ func (r *Registry) stageRegistrarFor() (func(plugin string, stage sdk.Stage, han } // noteChannel 记住插件注册了哪个通道，供卸载/崩溃时摘除。 +// forgetChannel 把某个通道从"本插件注册过哪些通道"的记账里摘掉（注销通道时用）。 +// +// 不摘的话 status 的 channels 列表与 input_channels 视图会一直列着已死通道， +// 模型会以为它还在（远程设备掉线后尤其明显）。 +func (r *Registry) forgetChannel(plugin, channel string) { + if plugin == "" || channel == "" { + return + } + r.channelsMu.Lock() + defer r.channelsMu.Unlock() + set := r.pluginChannels[plugin] + if set == nil { + return + } + delete(set.outputs, channel) + delete(set.inputs, channel) +} + func (r *Registry) noteChannel(plugin, channel string, output bool) { if plugin == "" || channel == "" { return diff --git a/internal/plugins/remotedevice/device.go b/internal/plugins/remotedevice/device.go index febef98..380e14c 100644 --- a/internal/plugins/remotedevice/device.go +++ b/internal/plugins/remotedevice/device.go @@ -182,6 +182,9 @@ func (d *devicectlDevice) Tools() []agentIO.ToolDef { func (d *devicectlDevice) Execute(tool string, args map[string]interface{}) (interface{}, error) { switch tool { + case "output": + // 出站：见 outputch.go。声明了 CapStructured 就必须真的能收输出。 + return d.output(args) case "devicedetect": return d.detect(args) case "device_ctl_status": diff --git a/internal/plugins/remotedevice/outputch.go b/internal/plugins/remotedevice/outputch.go new file mode 100644 index 0000000..de1ea0a --- /dev/null +++ b/internal/plugins/remotedevice/outputch.go @@ -0,0 +1,255 @@ +package remotedevice + +// 设备输出通道：把"agent 主动发给设备"做成**每设备一个输出通道** `device/`。 +// +// 为什么是输出通道而不是再加一批工具： +// - **寻址**：`output_send__device/` 直接指名道姓；模型看 `output_list_channels` +// 就知道当前有哪些设备在线，不必先 `devicedetect` 再往参数里塞 device_id。 +// - **能力**：caps 由设备声明的 caps 映射，**内核**在发送前就按 caps 拦 +// （把图片发给只支持文本的音箱会被拒，而不是等设备侧报错）。 +// - **授权**：`AllowedOutputs` 是内核级的授权闸（`executeOutputSendTool` 里先查 +// `IsOutputAllowed`）。父 agent 因此可以"只授权某一台设备"给驻留子 —— +// 这在工具模型下做不到（拿到 `device_ctl_cmdrun` 就能对任意设备下指令）。 +// +// 而 `screensee`/`computeruse`/`device_ctl_*` 这类**请求-响应**仍留作工具： +// 它们的返回值（图像/命令输出/状态）必须进模型上下文，做成通道会丢掉这个语义。 + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "log" + "strings" + "time" + + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// deviceOutputCaps 把设备声明的 caps 映射成输出通道能力位。 +// +// 映射依据（与 registry.go 的 capabilityTools/compatFullCaps 同一套词表）： +// - 文本：任何设备都收（消息/指令都是文本）→ 总是 CapText +// - 结构化：能跑命令（cmd/cmdrun/cmdresult 视为历史全能力）→ 能渲染结构化结果 +// - 音频：speaker/speakeruse，或 kind=speaker +// - 图片/文件：有屏（screen/screensue/screensee/gui）、剪切板、摄像头，或 kind 是带屏设备 +// - **未声明任何已知能力** → 视为全能力（与 deviceSupportsTool 的旧设备兼容规则一致） +func deviceOutputCaps(caps []string, kind string) agentIO.OutputCapability { + full := agentIO.CapText | agentIO.CapFile | agentIO.CapImage | agentIO.CapAudio | agentIO.CapStructured + + out := agentIO.CapText + known := false + for _, c := range caps { + switch c { + case "cmd", "cmdrun", "cmdresult": + // 历史"全能力"标记：这类设备能跑命令、能收结构化结果。 + return full + case "screen", "screensue", "screensee", "gui", "display": + known = true + out |= agentIO.CapImage | agentIO.CapFile + case "clipboard", "clipboardsee", "clipboardsue": + known = true + out |= agentIO.CapFile + case "camera", "camerasue": + known = true + out |= agentIO.CapImage | agentIO.CapFile + case "speaker", "speakeruse", "audio": + known = true + out |= agentIO.CapAudio + case "computeruse": + known = true + out |= agentIO.CapStructured + } + } + // kind 兜底：带屏设备即便没声明 caps，也能收图和文件。 + switch kind { + case "computer", "phone", "tablet", "tv": + known = true + out |= agentIO.CapImage | agentIO.CapFile | agentIO.CapStructured + case "speaker": + known = true + out |= agentIO.CapAudio + } + if !known { + return full // 旧设备兼容：未声明已知能力 ⇒ 全能力 + } + return out +} + +// deviceChannelName 是设备输出（也是输入）通道名：`device/`。 +// +// 入站与出站**同名**：两者指的是同一台设备，分成两个名字只会让模型与授权表更难对。 +func deviceChannelName(id string) string { return "device/" + id } + +// wireDeviceChannels 把"设备上下线"接到通道的登记/注销上。 +// +// 一台设备 = 一对**同名**通道 `device/`：入站（设备上报 → agent）与出站 +// （agent → 设备）。用**同步回调**而不是 ChangeChan（后者是 select+default， +// 缓冲满会丢事件；丢一次就留下死通道或漏注册）。 +// +// 抽成方法而不是内联在 Start 里：测试要能走**同一条**接线， +// 否则测试自己塞 handler，Start 忘了接线也照样绿。 +func (p *Plugin) wireDeviceChannels() { + p.registry.SetPresenceHandler( + func(meta DeviceMeta) { + _ = p.sdk.RegisterInputChannel(deviceChannelName(meta.DeviceID), sdk.ChannelDef{}) + p.ensureDeviceOutputChannel(meta.DeviceID) + }, + func(id string) { p.dropDeviceOutputChannel(id) }, + ) +} + +// ensureDeviceOutputChannel 给在线设备注册输出通道 device/（幂等）。 +func (p *Plugin) ensureDeviceOutputChannel(id string) { + if p.sdk == nil || id == "" { + return + } + meta, ok := p.registry.Get(id) + if !ok || !meta.Online { + return + } + ch := deviceChannelName(id) + caps := deviceOutputCaps(meta.Caps, meta.Kind) + desc := fmt.Sprintf("远程设备 %s（%s）：agent 主动向该设备发送内容；能力位 %s", + id, fallback(meta.Name, meta.Kind), agentIO.OutputCapability(caps).String()) + // 重复注册是安全的：芯片侧 Register 会合并（owner/capacity 取旧值）。 + if err := p.sdk.RegisterOutputChannel(ch, int(caps), desc, sdk.ChannelDef{}, func(args map[string]interface{}) (interface{}, error) { + return pushToDevice(p.registry, id, args) + }); err != nil { + p.logf("register output channel %s: %v", ch, err) + return + } + p.logf("device %s online → 输出通道 %s（caps=%s）", id, ch, agentIO.OutputCapability(caps).String()) +} + +// dropDeviceOutputChannel 设备下线时注销它的输出通道。 +// +// 不注销的后果：`output_list_channels` 一直列着它，模型会往死通道发消息， +// 拿到的却只是"发送已提交"之类的假回执。 +func (p *Plugin) dropDeviceOutputChannel(id string) { + if p.sdk == nil || id == "" { + return + } + ch := deviceChannelName(id) + if err := p.sdk.UnregisterOutputChannel(ch); err != nil { + p.logf("unregister output channel %s: %v", ch, err) + return + } + p.logf("device %s offline → 注销输出通道 %s", id, ch) +} + +// pushToDevice 把一次 output_send 的 {payload,type,meta} 转成下行帧发给设备。 +// +// 线上格式（新增 op=push，与既有 op=cmd/cmd_speech_* 并列）： +// +// {"op":"push","req_id":"...","type":"text|image|file|audio|structured","payload":"...","meta":"..."} +// +// 大负载（data URL 形式的图片/音频/文件）走既有分块通道 PushData， +// 避免把 base64 塞进一个超大文本帧。 +func pushToDevice(reg *Registry, id string, args map[string]interface{}) (interface{}, error) { + payload, _ := args["payload"].(string) + typ, _ := args["type"].(string) + metaStr, _ := args["meta"].(string) + if payload == "" { + return nil, fmt.Errorf("payload 不能为空") + } + if typ == "" { + typ = "text" + } + reqID := fmt.Sprintf("push_%d", time.Now().UnixNano()) + + if data, mime, ok := decodeDataURL(payload); ok && typ != "text" && typ != "structured" { + if err := reg.PushData(id, reqID, typ, mime, data); err != nil { + return nil, err + } + return map[string]interface{}{"status": "sent", "req_id": reqID}, nil + } + + frame := map[string]interface{}{ + "op": "push", + "req_id": reqID, + "type": typ, + "payload": payload, + } + if metaStr != "" { + frame["meta"] = metaStr + } + if err := reg.PushJSON(id, frame); err != nil { + return nil, err + } + return map[string]interface{}{"status": "sent", "req_id": reqID}, nil +} + +// decodeDataURL 解析 data:;base64, 形式的内联负载。 +func decodeDataURL(s string) (data []byte, mime string, ok bool) { + if !strings.HasPrefix(s, "data:") { + return nil, "", false + } + rest := strings.TrimPrefix(s, "data:") + comma := strings.Index(rest, ",") + if comma < 0 { + return nil, "", false + } + head, body := rest[:comma], rest[comma+1:] + if !strings.HasSuffix(head, ";base64") { + return nil, "", false + } + mime = strings.TrimSuffix(head, ";base64") + b, err := base64.StdEncoding.DecodeString(body) + if err != nil { + return nil, "", false + } + return b, mime, true +} + +func fallback(s, def string) string { + if strings.TrimSpace(s) == "" { + return def + } + return s +} + +func (p *Plugin) logf(format string, a ...interface{}) { + log.Printf("[remotedevice] "+format, a...) +} + +// output 是**聚合通道** `devicectl` 的出站实现（每设备通道之外的另一条路）。 +// +// 历史状态：devicectlDevice 一直声明 OutputCapabilities=CapStructured， +// 但 Execute 里根本没有 "output" 分支 ⇒ `output_send__devicectl` 必然报 +// "unknown device tool output"。这里把它补实：按 meta/device_id 指到具体设备。 +// +// 寻址方式（两者都收，模型的写法越少歧义越好）： +// - args.meta 是 JSON 且含 device_id：{"device_id":"phone-1"} +// - args.meta 直接就是设备 id：phone-1 +// - args.device_id +// +// 留空则返回**可执行**的提示（列出在线设备），而不是含糊报错 —— 模型据此重试。 +func (d *devicectlDevice) output(args map[string]interface{}) (interface{}, error) { + deviceID, _ := args["device_id"].(string) + if deviceID == "" { + if metaStr, _ := args["meta"].(string); metaStr != "" { + var m map[string]interface{} + if json.Unmarshal([]byte(metaStr), &m) == nil { + deviceID, _ = m["device_id"].(string) + if deviceID == "" { + deviceID, _ = m["device"].(string) + } + } + if deviceID == "" { + deviceID = strings.TrimSpace(metaStr) + } + } + } + if deviceID == "" { + ids := []string{} + for _, m := range d.reg.OnlineList() { + ids = append(ids, m.DeviceID) + } + if len(ids) == 0 { + return nil, fmt.Errorf("devicectl 需要 meta.device_id 才能投递；当前没有在线设备（device_list_channels 可看每台设备的 device/ 通道）") + } + return nil, fmt.Errorf("devicectl 需要 meta.device_id（或直接用通道 device/）；当前在线设备: %s", strings.Join(ids, ", ")) + } + return pushToDevice(d.reg, deviceID, args) +} diff --git a/internal/plugins/remotedevice/outputch_test.go b/internal/plugins/remotedevice/outputch_test.go new file mode 100644 index 0000000..5a74ca8 --- /dev/null +++ b/internal/plugins/remotedevice/outputch_test.go @@ -0,0 +1,257 @@ +package remotedevice + +// 设备输出通道（outputch.go）的测试： +// - caps 映射词表 +// - 上下线 → 通道登记/注销 + push 真能落到设备（走真 WS 帧） +// - 聚合通道 devicectl 的出站寻址 + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +func TestDeviceOutputCapsMapping(t *testing.T) { + full := agentIO.CapText | agentIO.CapFile | agentIO.CapImage | agentIO.CapAudio | agentIO.CapStructured + cases := []struct { + name string + caps []string + kind string + want agentIO.OutputCapability + }{ + {"未声明已知能力 ⇒ 全能力（旧设备兼容）", nil, "", full}, + {"未知 caps ⇒ 全能力", []string{"whatever"}, "", full}, + {"cmd ⇒ 历史全能力", []string{"cmd"}, "", full}, + {"speaker ⇒ 文本+音频", []string{"speaker"}, "", agentIO.CapText | agentIO.CapAudio}, + {"screen ⇒ 文本+图+文件", []string{"screen"}, "", agentIO.CapText | agentIO.CapImage | agentIO.CapFile}, + {"clipboard ⇒ 文本+文件", []string{"clipboard"}, "", agentIO.CapText | agentIO.CapFile}, + {"kind=computer 兜底（未声明 caps）", nil, "computer", agentIO.CapText | agentIO.CapImage | agentIO.CapFile | agentIO.CapStructured}, + {"kind=speaker 兜底", nil, "speaker", agentIO.CapText | agentIO.CapAudio}, + } + for _, c := range cases { + if got := deviceOutputCaps(c.caps, c.kind); got != c.want { + t.Errorf("%s: deviceOutputCaps(%v,%q)=%s，期望 %s", c.name, c.caps, c.kind, got, c.want) + } + } +} + +// channelRecorder 记录通道注册/注销。 +// +// 必须加锁：注册/注销发生在设备 WS 的处理 goroutine（上下线回调）里， +// 而测试在主线读 —— 裸 map/slice 会被 -race 抓住（第一版就是这么被抓住的）。 +type channelRecorder struct { + mu sync.Mutex + registered map[string]int + unregistered []string +} + +func (r *channelRecorder) caps(name string) (int, bool) { + r.mu.Lock() + defer r.mu.Unlock() + c, ok := r.registered[name] + return c, ok +} + +func (r *channelRecorder) unregList() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.unregistered...) +} + +// recordingSDK 造一个能记账的 PluginSDK：通道注册/注销都落到 recorder 里。 +func recordingSDK(t *testing.T, iom *agentIO.IOManager, rec *channelRecorder) *sdk.PluginSDK { + t.Helper() + return sdk.New("remotedevice", sdk.SDKConfig{ + IOManager: iom, + RegOutput: func(name string, caps int, desc string, def sdk.ChannelDef, handler sdk.ToolHandler) error { + rec.mu.Lock() + rec.registered[name] = caps + rec.mu.Unlock() + // 通道 handler 也要真的可调用 —— 记进 iom 才能从外面触发。 + return iom.RegisterDevice(&recordingDevice{name: name, handler: handler}) + }, + RegOutputUnreg: func(name string) error { + rec.mu.Lock() + rec.unregistered = append(rec.unregistered, name) + rec.mu.Unlock() + iom.UnregisterDevice(name) + return nil + }, + RegInput: func(name string, def sdk.ChannelDef) error { + iom.RegisterInputChannel(name, agentIO.ChannelDef(def)) + return nil + }, + }) +} + +// recordingDevice 把插件注册的输出通道在 io 层落地，便于用 Execute("output") 触发。 +type recordingDevice struct { + name string + handler sdk.ToolHandler +} + +func (d *recordingDevice) Name() string { return d.name } +func (d *recordingDevice) Type() agentIO.DeviceType { return agentIO.DeviceIO } +func (d *recordingDevice) OutputCapabilities() agentIO.OutputCapability { return agentIO.CapText } +func (d *recordingDevice) Description() string { return "recording device" } +func (d *recordingDevice) ChannelDef() agentIO.ChannelDef { return agentIO.ChannelDef{} } +func (d *recordingDevice) Start() error { return nil } +func (d *recordingDevice) Stop() error { return nil } +func (d *recordingDevice) Tools() []agentIO.ToolDef { return nil } +func (d *recordingDevice) Execute(tool string, args map[string]interface{}) (interface{}, error) { + return d.handler(args) +} + +func TestDeviceChannelLifecycleAndPush(t *testing.T) { + reg := NewRegistry() + token := "tk" + reg.SetAcceptToken(func(p string) bool { return p == token }) + + iom := agentIO.NewIOManager() + rec := &channelRecorder{registered: map[string]int{}} + + p := &Plugin{registry: reg} + p.sdk = recordingSDK(t, iom, rec) + p.wireDeviceChannels() // 走 Start 的同一条接线 + + srv := httptest.NewServer(http.HandlerFunc(reg.ServeWS)) + defer srv.Close() + + cli := dialTestWS(t, srv.URL, token) + defer cli.close() + + // 设备上线（caps=speaker ⇒ 通道能力应为 文本+音频） + cli.sendText([]byte(`{"op":"hello","device":{"device_id":"spk-1","name":"音箱","kind":"speaker","caps":["speaker"]}}`)) + cli.readHelloAckAndBind(t, token) + + ch := deviceChannelName("spk-1") + deadline := time.Now().Add(3 * time.Second) + caps, ok := rec.caps(ch) + for !ok && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + caps, ok = rec.caps(ch) + } + if !ok { + t.Fatalf("设备上线后应注册输出通道 %s，实际: %v", ch, rec.unregList()) + } + if want := int(agentIO.CapText | agentIO.CapAudio); caps != want { + t.Fatalf("通道能力位应为 文本+音频(%d)，实际 %d", want, caps) + } + // 入站 inputch 同名登记（父 agent 才能把"这台设备"划给驻留子） + if _, ok := iom.LookupInputChannel(ch); !ok { + t.Fatalf("设备上线后应同时登记同名 inputch %s", ch) + } + + // 触发一次出站：走 io 层的 output 分发（与 output_send__<通道> 同一条路） + dev := iom.GetDevice(ch) + if dev == nil { + t.Fatalf("输出通道 %s 未在 io 层注册", ch) + } + if _, err := dev.Execute("output", map[string]interface{}{"payload": "你好，设备", "type": "text"}); err != nil { + t.Fatalf("向设备发送失败: %v", err) + } + + // 设备侧应收到 op=push 的帧 + got := make(chan map[string]interface{}, 1) + go func() { + _, payload, err := cli.readMsg() + if err != nil { + return + } + var m map[string]interface{} + if json.Unmarshal(payload, &m) == nil { + got <- m + } + }() + select { + case m := <-got: + if m["op"] != "push" { + t.Fatalf("设备应收到 op=push，实际 %v", m) + } + if m["payload"] != "你好，设备" || m["type"] != "text" { + t.Fatalf("push 帧内容不符: %v", m) + } + case <-time.After(3 * time.Second): + t.Fatal("设备未收到 push 帧") + } + + // 设备下线 ⇒ 注销通道（不留死通道） + cli.close() + deadline = time.Now().Add(3 * time.Second) + for len(rec.unregList()) == 0 && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if got := rec.unregList(); len(got) == 0 || got[0] != ch { + t.Fatalf("设备下线后应注销 %s，实际 %v", ch, got) + } + if iom.GetDevice(ch) != nil { + t.Fatalf("注销后 io 层不应还有 %s", ch) + } +} + +func TestDevicectlAggregateOutputAddressing(t *testing.T) { + reg := NewRegistry() + token := "tk2" + reg.SetAcceptToken(func(p string) bool { return p == token }) + srv := httptest.NewServer(http.HandlerFunc(reg.ServeWS)) + defer srv.Close() + + dev := &devicectlDevice{reg: reg} + + // ① 没指定设备 ⇒ 报错要**可执行**（列出在线设备），而不是含糊失败 + if _, err := dev.Execute("output", map[string]interface{}{"payload": "x", "type": "text"}); err == nil { + t.Fatal("无 device_id 时应报错") + } + + cli := dialTestWS(t, srv.URL, token) + defer cli.close() + cli.sendText([]byte(`{"op":"hello","device":{"device_id":"pc-1","name":"PC","kind":"computer","caps":["cmd"]}}`)) + cli.readHelloAckAndBind(t, token) + + deadline := time.Now().Add(3 * time.Second) + for !reg.Online("pc-1") && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if !reg.Online("pc-1") { + t.Fatal("设备未上线") + } + + // ② meta 是 JSON 且含 device_id ⇒ 投递到该设备 + if _, err := dev.Execute("output", map[string]interface{}{ + "payload": "hi", "type": "text", "meta": `{"device_id":"pc-1"}`, + }); err != nil { + t.Fatalf("按 meta.device_id 投递失败: %v", err) + } + m := make(chan map[string]interface{}, 1) + go func() { + _, payload, err := cli.readMsg() + if err != nil { + return + } + var got map[string]interface{} + if json.Unmarshal(payload, &got) == nil { + m <- got + } + }() + select { + case got := <-m: + if got["op"] != "push" { + t.Fatalf("应为 push 帧，实际 %v", got) + } + case <-time.After(3 * time.Second): + t.Fatal("聚合通道未投递到设备") + } + + // ③ 设备在线但指定了不存在的设备 ⇒ 报错（online 列表里有 pc-1） + if _, err := dev.Execute("output", map[string]interface{}{ + "payload": "hi", "type": "text", "device_id": "ghost", + }); err == nil { + t.Fatal("不存在的设备应报错") + } +} diff --git a/internal/plugins/remotedevice/plugin.go b/internal/plugins/remotedevice/plugin.go index 865f289..b3b2da0 100644 --- a/internal/plugins/remotedevice/plugin.go +++ b/internal/plugins/remotedevice/plugin.go @@ -114,10 +114,17 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { } } + // ---- 设备通道随在线状态生灭（见 outputch.go 的 wireDeviceChannels）---- + p.wireDeviceChannels() + // ---- 设备主动上报事件 → agent 注入 ---------------- // 摄像头发现异常/传感器报警等场景：设备经 WS op=event 上报， - // 插件将其格式化为文本经 SDK InjectText 异步注入 agent（source=device/{id}， - // 回复路由回 device/{id} 通道），同时发 EventBus 供 WebUI 展示。 + // 插件将其格式化为文本经 SDK InjectText 异步注入 agent（source=device/{id}）， + // 同时发 EventBus 供 WebUI 展示。 + // + // **注意**：agent 的输出**不会**被自动转回设备 —— 主动转发只有 webui 与 cli 两个 + // 交互界面（它们把最终回复渲染成对话气泡是本职）。设备要走 + // `output_send__device/`（agent 主动调用），这才与"输出是 agent 的主动调用"一致。 // 节流：同设备同类型事件 10s 内去重，防传感器风暴。 lastEventAt := map[string]time.Time{} var eventMu sync.Mutex @@ -299,6 +306,13 @@ func (p *Plugin) describeScreen(dataURL string, provider string) string { } func (p *Plugin) Stop() error { + // 注销全部设备通道：插件卸载/重载后这些通道不再有实现， + // 留着会让 output_list_channels 骗模型。 + if p.sdk != nil { + for _, m := range p.registry.List() { + p.dropDeviceOutputChannel(m.DeviceID) + } + } if p.server != nil { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() diff --git a/internal/plugins/remotedevice/registry.go b/internal/plugins/remotedevice/registry.go index f861a8e..54b3965 100644 --- a/internal/plugins/remotedevice/registry.go +++ b/internal/plugins/remotedevice/registry.go @@ -60,10 +60,14 @@ func (c *wconn) unlockWrite() { // Registry 是设备接入网关的注册表：管理在线连接、设备元数据。线程安全。 // 鉴权在设备端执行，服务端不存储授权状态。 type Registry struct { - mu sync.RWMutex - devices map[string]*DeviceMeta // deviceID -> meta（在线/历史） - conns map[string]*wconn // deviceID -> 活跃连接（支持 push） - onlineCh chan string + mu sync.RWMutex + devices map[string]*DeviceMeta // deviceID -> meta（在线/历史） + conns map[string]*wconn // deviceID -> 活跃连接（支持 push） + onlineCh chan string + + // onOnline/onOffline：设备上下线的同步回调（见 SetPresenceHandler）。 + onOnline func(DeviceMeta) + onOffline func(string) onStatus func(msg map[string]interface{}) onEvent func(deviceID string, msg map[string]interface{}) acceptFn func(token string) bool @@ -281,7 +285,12 @@ func (r *Registry) register(meta DeviceMeta) { meta.LastSeen = time.Now().Unix() // 保留设备自报的授权状态（客户端鉴权，服务端不覆盖） r.devices[meta.DeviceID] = &meta + onOnline := r.onOnline r.mu.Unlock() + // 先回调（可能注册 device/ 输出通道），再发变更通知。 + if onOnline != nil { + onOnline(meta) + } r.notifyChange(meta.DeviceID) } @@ -291,10 +300,27 @@ func (r *Registry) markOffline(id string) { m.Online = false } delete(r.conns, id) + onOffline := r.onOffline r.mu.Unlock() + if onOffline != nil { + onOffline(id) + } r.notifyChange(id) } +// SetPresenceHandler 注册设备上线/下线回调。 +// +// 为什么不用 ChangeChan：那是 `select { case ch <- id: default: }`，缓冲满了会**丢事件** +// （设备上下线是要跟"注册/注销输出通道"绑定的，丢一次就会留下一个死通道或漏注册）。 +// 这里同步调用，且在**释放锁之后**调 —— 回调内部会回查 registry（Get/List）， +// 持锁调用会自己锁死自己。 +func (r *Registry) SetPresenceHandler(onOnline func(DeviceMeta), onOffline func(string)) { + r.mu.Lock() + r.onOnline = onOnline + r.onOffline = onOffline + r.mu.Unlock() +} + func (r *Registry) notifyChange(id string) { select { case r.onlineCh <- id: diff --git a/internal/sdk/plugin.go b/internal/sdk/plugin.go index 74f7490..468b3d4 100644 --- a/internal/sdk/plugin.go +++ b/internal/sdk/plugin.go @@ -50,6 +50,7 @@ const ( type APIRegistrar = pubsdk.APIRegistrar type OutputChannelRegistrar = pubsdk.OutputChannelRegistrar +type OutputChannelUnregistrar = pubsdk.OutputChannelUnregistrar type InputChannelRegistrar = pubsdk.InputChannelRegistrar type ChannelDef = pubsdk.ChannelDef @@ -284,20 +285,21 @@ type AudioURL = pubsdk.AudioURL // SDKConfig holds all dependencies for creating a PluginSDK. type SDKConfig struct { - IOManager *agentIO.IOManager - EventBus *events.Bus - Memory MemoryAPI - TextMemory TextMemoryAPI - DocMemory DocMemoryAPI - Knowledge KnowledgeAPI - LLM LLMAPI - Settings SettingsAPI - RegTool ToolRegistrar - RegStage StageRegistrar - RegAPI APIRegistrar - RegOutput OutputChannelRegistrar - RegInput InputChannelRegistrar - PluginMgr PluginManager + IOManager *agentIO.IOManager + EventBus *events.Bus + Memory MemoryAPI + TextMemory TextMemoryAPI + DocMemory DocMemoryAPI + Knowledge KnowledgeAPI + LLM LLMAPI + Settings SettingsAPI + RegTool ToolRegistrar + RegStage StageRegistrar + RegAPI APIRegistrar + RegOutput OutputChannelRegistrar + RegOutputUnreg OutputChannelUnregistrar + RegInput InputChannelRegistrar + PluginMgr PluginManager Status StatusAPI Supervisor SupervisorAPI @@ -316,6 +318,9 @@ func New(name string, cfg SDKConfig) *PluginSDK { if cfg.RegInput != nil { base.SetInputChannelRegistrar(cfg.RegInput) } + if cfg.RegOutputUnreg != nil { + base.SetOutputChannelUnregistrar(cfg.RegOutputUnreg) + } base.SetMemoryAPI(cfg.Memory) base.SetTextMemoryAPI(cfg.TextMemory) base.SetDocMemoryAPI(cfg.DocMemory) diff --git a/scripts/kernel-stress/README.md b/scripts/kernel-stress/README.md index 55b782c..861cb70 100644 --- a/scripts/kernel-stress/README.md +++ b/scripts/kernel-stress/README.md @@ -46,6 +46,30 @@ export KCLI_KEY=$(sqlite3 $DATA/config.db "select value from config_webui where ./stress.py $DATA/cli.sock "$KCLI_KEY" 1 1 0 0 resident 0.0 # 驻留子全链路（mock 见 !resident 标记） ``` +## 远程设备（agent ↔ 设备）端到端 + +设备网关在实例的 netns 内监听 `127.0.0.1:9890`，所以设备客户端要**进同一个 netns** 跑： + +```bash +TOKEN=$(sqlite3 $DATA/config.db "select value from config_remotedevice where key='ws_token';") +nsenter -t $(cat $DATA/pid) -n python3 ./devclient.py --port 9890 --token "$TOKEN" \ + --id pydev-1 --caps cmd --seconds 30 --out /var/tmp/push.txt +``` + +然后让 agent 主动发一条（mock 里 `!push` 会回一个 `output_send__device/pydev-1` 的工具调用）： + +```bash +# 经 CLI socket 发 "!push"，设备侧应收到 {"op":"push", ...} +``` + +设备上线/下线会在内核里登记/注销输出通道 `device/`，可用 `/kernel` 的 channels 观察 +（在线时出现、掉线后消失）。 + +> **设计口径**：agent → 设备**必须**是 agent 的主动调用（`output_send__device/`）； +> 设备的上报（`op=event`）虽然会被注入成输入，但 agent 的回复**不会**被插件自动转回设备 +> —— 全仓只有 **webui 与 cli** 两个交互界面"主动转发"（把最终回复渲染成气泡/终端输出）， +> 其它通道（qq、设备等）一律要求显式 `output_send__<通道>`。 + ## 读结果 | 指标 | 含义 | diff --git a/scripts/kernel-stress/devclient.py b/scripts/kernel-stress/devclient.py new file mode 100644 index 0000000..4216b50 --- /dev/null +++ b/scripts/kernel-stress/devclient.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +"""最小远程设备客户端（WebSocket，无第三方依赖）。 + +用途：在**真内核**上验证"agent → 设备"这条出站链路 —— +`output_send__device/` 应该以 `op=push` 的帧落到设备。 + +流程：握手 → hello（自报 id/kind/caps）→ bind → 打印收到的帧； +看到 push 就往 --out 文件里写一行（便于 shell 断言）。 + +注意：实例跑在私有 netns 里，设备网关的 127.0.0.1:9890 在 netns 内， +所以本脚本要用 nsenter 进同一个 netns 跑，例如： + nsenter -t -n python3 devclient.py --port 9890 --token X --id pydev-1 --out /tmp/push.txt +""" +import argparse +import base64 +import hashlib +import json +import os +import socket +import struct +import threading +import time + +GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + + +class WS: + def __init__(self, host, port, path, timeout=30): + self.sock = socket.create_connection((host, port), timeout=timeout) + self.sock.settimeout(timeout) + self._handshake(host, port, path) + + def _handshake(self, host, port, path): + key = base64.b64encode(os.urandom(16)).decode() + req = ( + f"GET {path} HTTP/1.1\r\n" + f"Host: {host}:{port}\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + f"Sec-WebSocket-Key: {key}\r\n" + "Sec-WebSocket-Version: 13\r\n\r\n" + ) + self.sock.sendall(req.encode()) + buf = b"" + while b"\r\n\r\n" not in buf: + chunk = self.sock.recv(4096) + if not chunk: + raise RuntimeError("握手未完成: 连接关闭") + buf += chunk + head = buf.decode("utf-8", "replace") + if "101" not in head.split("\r\n")[0]: + raise RuntimeError("握手被拒: " + head.split("\r\n")[0]) + expect = base64.b64encode(hashlib.sha1((key + GUID).encode()).digest()).decode() + if expect.lower() not in head.lower(): + raise RuntimeError("Sec-WebSocket-Accept 校验失败") + self.buf = buf.split(b"\r\n\r\n", 1)[1] + + # ---- 发送 ---- + def send(self, opcode, payload=b""): + header = bytes([0x80 | opcode]) + mask = os.urandom(4) + n = len(payload) + if n < 126: + header += bytes([0x80 | n]) + elif n < 65536: + header += bytes([0x80 | 126]) + struct.pack(">H", n) + else: + header += bytes([0x80 | 127]) + struct.pack(">Q", n) + masked = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + self.sock.sendall(header + mask + masked) + + def send_json(self, obj): + self.send(0x1, json.dumps(obj).encode()) + + # ---- 接收 ---- + def _read(self, n): + while len(self.buf) < n: + chunk = self.sock.recv(65536) + if not chunk: + raise RuntimeError("连接关闭") + self.buf += chunk + out, self.buf = self.buf[:n], self.buf[n:] + return out + + def recv_frame(self): + b0, b1 = self._read(2) + opcode = b0 & 0x0F + ln = b1 & 0x7F + if ln == 126: + ln = struct.unpack(">H", self._read(2))[0] + elif ln == 127: + ln = struct.unpack(">Q", self._read(8))[0] + masked = b1 & 0x80 + mask = self._read(4) if masked else None + payload = self._read(ln) + if mask: + payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload)) + return opcode, payload + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--host", default="127.0.0.1") + ap.add_argument("--port", type=int, default=9890) + ap.add_argument("--token", required=True, help="设备接入 token（config_remotedevice.ws_token）") + ap.add_argument("--id", default="pydev-1") + ap.add_argument("--kind", default="computer") + ap.add_argument("--caps", default="cmd") + ap.add_argument("--seconds", type=float, default=25) + ap.add_argument("--out", default="", help="收到 push 时写一行到此文件") + args = ap.parse_args() + + ws = WS(args.host, args.port, "/api/v1/device/ws?token=" + args.token) + ws.send_json({"op": "hello", "device": { + "device_id": args.id, "name": "python 设备", "kind": args.kind, + "caps": [c for c in args.caps.split(",") if c], + }}) + op, payload = ws.recv_frame() + ack = json.loads(payload) + print("[devclient] hello_ack:", ack, flush=True) + + ws.send_json({"op": "bind", "device_id": args.id, "token": args.token}) + op, payload = ws.recv_frame() + bind = json.loads(payload) + print("[devclient] bind_ack:", bind, flush=True) + if not bind.get("ok"): + raise SystemExit("bind 被拒: %s" % bind) + + deadline = time.time() + args.seconds + ws.sock.settimeout(1.0) + while time.time() < deadline: + try: + op, payload = ws.recv_frame() + except socket.timeout: + continue + except Exception as e: # 服务端关闭 + print("[devclient] 连接结束:", e, flush=True) + break + if op == 0x9: # ping → pong + ws.send(0xA, payload) + continue + if op == 0x2: + print("[devclient] 收到二进制帧 %d 字节" % len(payload), flush=True) + continue + if op != 0x1: + continue + try: + m = json.loads(payload) + except Exception: + print("[devclient] 非 JSON 帧:", payload[:120], flush=True) + continue + print("[devclient] 收到帧:", json.dumps(m, ensure_ascii=False)[:300], flush=True) + if m.get("op") == "push" and args.out: + with open(args.out, "a") as f: + f.write(json.dumps(m, ensure_ascii=False) + "\n") + print("[devclient] ✅ 收到 push（agent → 设备链路通）", flush=True) + + +if __name__ == "__main__": + main() diff --git a/third_party/homeagent-sdk/sdk/plugin.go b/third_party/homeagent-sdk/sdk/plugin.go index 9f10d90..1f65774 100644 --- a/third_party/homeagent-sdk/sdk/plugin.go +++ b/third_party/homeagent-sdk/sdk/plugin.go @@ -284,6 +284,13 @@ type InputChannelRegistrar func(name string, def ChannelDef) error // OutputChannelRegistrar registers an output channel that the output_send tool can use. type OutputChannelRegistrar func(name string, caps int, desc string, def ChannelDef, handler ToolHandler) error +// OutputChannelUnregistrar 注销一个输出通道。 +// +// 为什么需要它：输出通道不止有"启动时注册一次"的静态通道，还有**随外部资源生灭**的 +// 动态通道 —— 典型是远程设备：`device/` 只在设备在线期间存在，设备掉线后 +// 必须注销，否则 output_list_channels 会一直列着它、模型会往一个死通道发消息。 +type OutputChannelUnregistrar func(name string) error + // Output capability flags const ( CapText = 1 @@ -296,22 +303,23 @@ const ( // PluginSDK is the main API surface provided to plugins at runtime. // It wraps tool registration, settings, memory, knowledge, LLM, and IO injection. type PluginSDK struct { - name string - regTool ToolRegistrar - regStage StageRegistrar - regAPI APIRegistrar - regOutput OutputChannelRegistrar - regInput InputChannelRegistrar - io IOInjector - mem MemoryAPI - textMem TextMemoryAPI - docMem DocMemoryAPI - know KnowledgeAPI - llm LLMAPI - sett SettingsAPI - social SocialAPI - events EventSubscriber - plgMgr PluginMgrAPI + name string + regTool ToolRegistrar + regStage StageRegistrar + regAPI APIRegistrar + regOutput OutputChannelRegistrar + regOutputUnreg OutputChannelUnregistrar + regInput InputChannelRegistrar + io IOInjector + mem MemoryAPI + textMem TextMemoryAPI + docMem DocMemoryAPI + know KnowledgeAPI + llm LLMAPI + sett SettingsAPI + social SocialAPI + events EventSubscriber + plgMgr PluginMgrAPI // apiMu 保护上面这些由内核注入的 API 字段，以及 autoRestart。 // @@ -487,6 +495,17 @@ func (s *PluginSDK) RegisterOutputChannel(name string, caps int, desc string, de return nil } +// UnregisterOutputChannel 注销一个输出通道（动态通道随资源生灭时必须调用）。 +func (s *PluginSDK) UnregisterOutputChannel(name string) error { + s.apiMu.RLock() + reg := s.regOutputUnreg + s.apiMu.RUnlock() + if reg != nil { + return reg(name) + } + return nil +} + // RegisterInputChannel registers an input channel with its memory behavior. // // 契约：**凡是用 InjectText*/InjectInput*/InjectInterrupt*(source, "", ...) @@ -518,6 +537,13 @@ func (s *PluginSDK) SetOutputChannelRegistrar(r OutputChannelRegistrar) { s.apiMu.Unlock() } +// SetOutputChannelUnregistrar sets the output channel unregistrar (called by the core at startup). +func (s *PluginSDK) SetOutputChannelUnregistrar(r OutputChannelUnregistrar) { + s.apiMu.Lock() + s.regOutputUnreg = r + s.apiMu.Unlock() +} + // SetInputChannelRegistrar sets the input channel registrar (called by the core at startup). func (s *PluginSDK) SetInputChannelRegistrar(r InputChannelRegistrar) { s.apiMu.Lock()