mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-04 00:03:59 +00:00
feat(remotedevice): 设备能力 outputch 化 —— 每设备一个 device/<id> 通道 + 设备指令类工具补授权闸
用户指出 remotedevice 的能力应当 outputch 化。查证后发现比"应当"更严重：
设备方向**根本没有出站实现**。
## 查到的三个缺口
1. `devicectlDevice` 一直声明 `OutputCapabilities() = CapStructured`（对外宣称可作输出目标），
但 `Execute` 的 switch 里**没有 "output" 分支** ⇒ `output_send__devicectl` 必然拿到
`unknown device tool output`，回模型"通过 [devicectl] 通道发送失败"。
2. 全插件没有 `RegisterOutputChannel`，也没有任何 EmitOutput/output_send 路径：
设备方向只有「工具（请求-响应）」与「设备→agent 注入」，**agent → 设备是断的**
（唯一的下行通道是个 HTTP 端点 `/api/v1/device/push`，不在 agent 的工具/通道模型里）。
3. 寻址是聚合的：所有设备共用一个名字 `devicectl`，没有 `device/<id>`；且设备 caps 只在
插件内部软检查（`SupportsTool`），**绕过**了内核的 `AllowedOutputs` 授权闸 ——
驻留子只要拿到 `device_ctl_cmdrun` 就能指挥**任意**设备。
## 按方向切分（不是一刀切）
**出站/消息类 → 每设备一个输出通道 `device/<id>`**（与入站同名）：
- 上线注册、掉线注销（caps 由设备声明的 caps 映射：文本恒有；有屏→图/文件；
speaker→音频；可跑命令(cmd/cmdrun/cmdresult)或未声明已知能力→全能力，与
`deviceSupportsTool` 的旧设备兼容规则一致）。断连不注销会留下死通道骗模型。
- 于是自动获得：内核按 caps 在**发送前**拦（送图给纯文本音箱直接拒）；
`output_list_channels` 能列出设备；`AllowedOutputs` 可按设备收窄给驻留子。
- 上下线钩子用 `Registry.SetPresenceHandler`（**同步回调**）而不是既有的 `ChangeChan`
（那是 select+default，缓冲满会丢事件；丢一次就留下死通道或漏注册）。
- `devicectl` 保留为聚合通道，并把它"声明了却不实现"的 output 补实：按
`meta.device_id`（或 meta 就是设备 id / args.device_id）路由；缺省时返回**可执行**的
报错（列出在线设备），而不是含糊失败。
**RPC 类保留为工具**（`devicedetect`/`screensee`/`computeruse`/`clipboard*`/
`device_ctl_status|cmdrun|cmdresult`）：它们的返回值（图像/命令输出/状态）必须进模型
上下文，做成通道会丢掉这个语义。
**并给设备指令类工具补上同一道授权闸**（core/toolcall.go）：`device_id` 指向的设备
必须是本 agent 被授权的 `device/<id>`。根 agent 默认完整授权 ⇒ 无行为变化；
驻留子收窄后，"拿到工具就能指挥任意设备"的缺口被堵上（新增 3 条 core 测试钉住）。
**设备端参考实现**（`internal/devicebridge/client/bridge.go` + waiter）：新增 `op=push`
分发与 `OnPush` 回调（文本/结构化；二进制走既有 `cmd_speech_*` → `DataHandler`），
waiter 把它打到终端。
## 设计口径（用户当场纠偏，已写进代码注释与 harness README）
**主动转发只有 webui 与 cli 两个交互界面**（webui 订阅 EventAgentOutput 渲染气泡、
cli 用同步回程写回终端）。其它通道一律要求 agent **显式** `output_send__<通道>`。
我第一版给 remotedevice 加了 `EventAgentOutput` 订阅来自动回投设备——那是凭空造了
第三个转发者，违背"输出是 agent 的主动调用"，已撤回（该测试一并删除）。
## 验收
- 单测：caps 映射词表；设备上线→注册通道（含同名 inputch）/掉线→注销；push 真落到
WS 设备；聚合通道 `devicectl` 的寻址（无 device_id 报可执行错误、按 meta 投递、
指定不存在设备报错）；core 授权闸 3 例。
- 全量 `go test ./...` = 37 包 ok / 0 FAIL；`-race`（agent/plugins/plugin/devicebridge/sdk）干净。
- **真二进制端到端**（私有 netns + mock LLM + 真 WS 设备客户端 scripts/kernel-stress/devclient.py）：
设备上线 → 通道表出现 `device/pydev-1`（caps=[text file image audio structured]，由
`caps:["cmd"]` 映射）→ agent 经 `output_send__device/pydev-1` 主动发送 → 设备收到
`{"op":"push","payload":"内核推给你的消息","type":"text"}` → 设备掉线 → 通道从表中消失。
This commit is contained in:
JianFeeeee
@ -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":
|
||||
|
||||
255
internal/plugins/remotedevice/outputch.go
Normal file
255
internal/plugins/remotedevice/outputch.go
Normal file
@ -0,0 +1,255 @@
|
||||
package remotedevice
|
||||
|
||||
// 设备输出通道：把"agent 主动发给设备"做成**每设备一个输出通道** `device/<id>`。
|
||||
//
|
||||
// 为什么是输出通道而不是再加一批工具：
|
||||
// - **寻址**：`output_send__device/<id>` 直接指名道姓；模型看 `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/<id>`。
|
||||
//
|
||||
// 入站与出站**同名**：两者指的是同一台设备，分成两个名字只会让模型与授权表更难对。
|
||||
func deviceChannelName(id string) string { return "device/" + id }
|
||||
|
||||
// wireDeviceChannels 把"设备上下线"接到通道的登记/注销上。
|
||||
//
|
||||
// 一台设备 = 一对**同名**通道 `device/<id>`：入站（设备上报 → 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/<id>（幂等）。
|
||||
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:<mime>;base64,<data> 形式的内联负载。
|
||||
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/<id> 通道）")
|
||||
}
|
||||
return nil, fmt.Errorf("devicectl 需要 meta.device_id（或直接用通道 device/<id>）；当前在线设备: %s", strings.Join(ids, ", "))
|
||||
}
|
||||
return pushToDevice(d.reg, deviceID, args)
|
||||
}
|
||||
257
internal/plugins/remotedevice/outputch_test.go
Normal file
257
internal/plugins/remotedevice/outputch_test.go
Normal file
@ -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("不存在的设备应报错")
|
||||
}
|
||||
}
|
||||
@ -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/<id>`（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()
|
||||
|
||||
@ -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/<id> 输出通道），再发变更通知。
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user