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
2026-09-13 12:23:28 +08:00
parent 907d92546c
commit df0f7042dd
15 changed files with 1008 additions and 48 deletions

View File

@ -72,6 +72,11 @@ func startDeviceBridge(addr, token string) error {
cmdRouter.Dispatch(reqID, command) cmdRouter.Dispatch(reqID, command)
}) })
// agent 主动投递(output_send__device/<id>)→ 终端显示。
// 设备侧参考实现:文本/结构化直接打出来;二进制负载走 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 { if err := bridge.Start(); err != nil {
return fmt.Errorf("device bridge: %w", err) return fmt.Errorf("device bridge: %w", err)
} }

View File

@ -0,0 +1,92 @@
package core
// 设备类工具的**授权闸**:设备指令类工具走的是工具面,而 AllowedOutputs 只作用于
// output_send__<通道> —— 不补闸的话"授权"对指令类完全无效(驻留子拿到
// device_ctl_cmdrun 就能指挥任意设备)。这里按目标设备的通道名 device/<id> 查同一道闸。
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)
}
}

View File

@ -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/<id> 查同一道闸:父授权了哪台设备,才允许指挥哪台。
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/<id> 通道,或 devicedetect)", id)
}
}
if a.tracker != nil { if a.tracker != nil {
a.tracker.PreAction(tc.Name) a.tracker.PreAction(tc.Name)
} }

View File

@ -507,6 +507,23 @@ func (m *IOManager) GetAllTools() []ToolDef {
return tools 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) { func (m *IOManager) ExecuteTool(name string, args map[string]interface{}) (ret interface{}, err error) {
m.mu.RLock() m.mu.RLock()
type nameDevice struct { type nameDevice struct {

View File

@ -22,6 +22,15 @@ type CmdResultHandler func(reqID, status, output, errMsg string)
// DataHandler 是二进制数据接收回调(如 TTS 音频)。 // DataHandler 是二进制数据接收回调(如 TTS 音频)。
type DataHandler func(reqID, kind, mime string, data []byte) type DataHandler func(reqID, kind, mime string, data []byte)
// PushHandler 接收 agent **主动投递**给本设备的内容。
//
// 与 cmd 的区别:cmd 是"让设备做一件事"(请求-响应,结果要回传),
// push 是"把这段内容交给设备"(agent 经 output_send__device/<id> 发起,
// 一种单向投递)。宿主按自己的形态落地:终端打出来、音箱念出来、屏幕显示。
//
// typ: text / structured / image / file / audio(二进制走 DataHandler,不走这里)
type PushHandler func(reqID, typ, payload, meta string)
// Bridge 是设备桥客户端核心结构体。 // Bridge 是设备桥客户端核心结构体。
// 管理 WebSocket 连接、消息路由、心跳保活和命令分发。 // 管理 WebSocket 连接、消息路由、心跳保活和命令分发。
// 授权状态由设备端本地存储(客户端鉴权),服务端不存储; // 授权状态由设备端本地存储(客户端鉴权),服务端不存储;
@ -46,6 +55,7 @@ type Bridge struct {
cmdHandler BridgeCmdHandler cmdHandler BridgeCmdHandler
resultHandler CmdResultHandler resultHandler CmdResultHandler
dataHandler DataHandler dataHandler DataHandler
pushHandler PushHandler
// 二进制数据聚合(服务端→设备,如 TTS 音频) // 二进制数据聚合(服务端→设备,如 TTS 音频)
speechAccum *speechBuffer speechAccum *speechBuffer
@ -157,6 +167,13 @@ func (b *Bridge) OnData(handler DataHandler) {
b.dataHandler = handler 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 秒)。 // SetPingInterval 设置心跳间隔(默认 30 秒)。
func (b *Bridge) SetPingInterval(d time.Duration) { func (b *Bridge) SetPingInterval(d time.Duration) {
b.mu.Lock() b.mu.Lock()
@ -478,6 +495,25 @@ func (b *Bridge) handleMessage(msg map[string]interface{}) {
dh(reqID, acc.kind, acc.mime, data) dh(reqID, acc.kind, acc.mime, data)
} }
case "push":
// agent 主动投递(output_send__device/<id>)。二进制负载走
// 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: default:
log.Printf("[devicebridge] unhandled op=%s", op) log.Printf("[devicebridge] unhandled op=%s", op)
} }

View File

@ -318,6 +318,15 @@ func (r *Registry) buildSDK(name string) *sdk.PluginSDK {
return nil 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 { regInput := func(chName string, def sdk.ChannelDef) error {
if r.iom == nil { if r.iom == nil {
return nil return nil
@ -344,6 +353,7 @@ func (r *Registry) buildSDK(name string) *sdk.PluginSDK {
RegStage: regStage, RegStage: regStage,
RegAPI: regAPI, RegAPI: regAPI,
RegOutput: regOutput, RegOutput: regOutput,
RegOutputUnreg: regOutputUnreg,
RegInput: regInput, RegInput: regInput,
PluginMgr: r, PluginMgr: r,
@ -570,6 +580,24 @@ func (r *Registry) stageRegistrarFor() (func(plugin string, stage sdk.Stage, han
} }
// noteChannel 记住插件注册了哪个通道,供卸载/崩溃时摘除。 // 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) { func (r *Registry) noteChannel(plugin, channel string, output bool) {
if plugin == "" || channel == "" { if plugin == "" || channel == "" {
return return

View File

@ -182,6 +182,9 @@ func (d *devicectlDevice) Tools() []agentIO.ToolDef {
func (d *devicectlDevice) Execute(tool string, args map[string]interface{}) (interface{}, error) { func (d *devicectlDevice) Execute(tool string, args map[string]interface{}) (interface{}, error) {
switch tool { switch tool {
case "output":
// 出站:见 outputch.go。声明了 CapStructured 就必须真的能收输出。
return d.output(args)
case "devicedetect": case "devicedetect":
return d.detect(args) return d.detect(args)
case "device_ctl_status": case "device_ctl_status":

View 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)
}

View 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("不存在的设备应报错")
}
}

View File

@ -114,10 +114,17 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error {
} }
} }
// ---- 设备通道随在线状态生灭(见 outputch.go 的 wireDeviceChannels)----
p.wireDeviceChannels()
// ---- 设备主动上报事件 → agent 注入 ---------------- // ---- 设备主动上报事件 → agent 注入 ----------------
// 摄像头发现异常/传感器报警等场景:设备经 WS op=event 上报, // 摄像头发现异常/传感器报警等场景:设备经 WS op=event 上报,
// 插件将其格式化为文本经 SDK InjectText 异步注入 agent(source=device/{id}, // 插件将其格式化为文本经 SDK InjectText 异步注入 agent(source=device/{id}),
// 回复路由回 device/{id} 通道),同时发 EventBus 供 WebUI 展示。 // 同时发 EventBus 供 WebUI 展示。
//
// **注意**:agent 的输出**不会**被自动转回设备 —— 主动转发只有 webui 与 cli 两个
// 交互界面(它们把最终回复渲染成对话气泡是本职)。设备要走
// `output_send__device/<id>`(agent 主动调用),这才与"输出是 agent 的主动调用"一致。
// 节流:同设备同类型事件 10s 内去重,防传感器风暴。 // 节流:同设备同类型事件 10s 内去重,防传感器风暴。
lastEventAt := map[string]time.Time{} lastEventAt := map[string]time.Time{}
var eventMu sync.Mutex var eventMu sync.Mutex
@ -299,6 +306,13 @@ func (p *Plugin) describeScreen(dataURL string, provider string) string {
} }
func (p *Plugin) Stop() error { 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 { if p.server != nil {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel() defer cancel()

View File

@ -64,6 +64,10 @@ type Registry struct {
devices map[string]*DeviceMeta // deviceID -> meta(在线/历史) devices map[string]*DeviceMeta // deviceID -> meta(在线/历史)
conns map[string]*wconn // deviceID -> 活跃连接(支持 push) conns map[string]*wconn // deviceID -> 活跃连接(支持 push)
onlineCh chan string onlineCh chan string
// onOnline/onOffline:设备上下线的同步回调(见 SetPresenceHandler)。
onOnline func(DeviceMeta)
onOffline func(string)
onStatus func(msg map[string]interface{}) onStatus func(msg map[string]interface{})
onEvent func(deviceID string, msg map[string]interface{}) onEvent func(deviceID string, msg map[string]interface{})
acceptFn func(token string) bool acceptFn func(token string) bool
@ -281,7 +285,12 @@ func (r *Registry) register(meta DeviceMeta) {
meta.LastSeen = time.Now().Unix() meta.LastSeen = time.Now().Unix()
// 保留设备自报的授权状态(客户端鉴权,服务端不覆盖) // 保留设备自报的授权状态(客户端鉴权,服务端不覆盖)
r.devices[meta.DeviceID] = &meta r.devices[meta.DeviceID] = &meta
onOnline := r.onOnline
r.mu.Unlock() r.mu.Unlock()
// 先回调(可能注册 device/<id> 输出通道),再发变更通知。
if onOnline != nil {
onOnline(meta)
}
r.notifyChange(meta.DeviceID) r.notifyChange(meta.DeviceID)
} }
@ -291,10 +300,27 @@ func (r *Registry) markOffline(id string) {
m.Online = false m.Online = false
} }
delete(r.conns, id) delete(r.conns, id)
onOffline := r.onOffline
r.mu.Unlock() r.mu.Unlock()
if onOffline != nil {
onOffline(id)
}
r.notifyChange(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) { func (r *Registry) notifyChange(id string) {
select { select {
case r.onlineCh <- id: case r.onlineCh <- id:

View File

@ -50,6 +50,7 @@ const (
type APIRegistrar = pubsdk.APIRegistrar type APIRegistrar = pubsdk.APIRegistrar
type OutputChannelRegistrar = pubsdk.OutputChannelRegistrar type OutputChannelRegistrar = pubsdk.OutputChannelRegistrar
type OutputChannelUnregistrar = pubsdk.OutputChannelUnregistrar
type InputChannelRegistrar = pubsdk.InputChannelRegistrar type InputChannelRegistrar = pubsdk.InputChannelRegistrar
type ChannelDef = pubsdk.ChannelDef type ChannelDef = pubsdk.ChannelDef
@ -296,6 +297,7 @@ type SDKConfig struct {
RegStage StageRegistrar RegStage StageRegistrar
RegAPI APIRegistrar RegAPI APIRegistrar
RegOutput OutputChannelRegistrar RegOutput OutputChannelRegistrar
RegOutputUnreg OutputChannelUnregistrar
RegInput InputChannelRegistrar RegInput InputChannelRegistrar
PluginMgr PluginManager PluginMgr PluginManager
@ -316,6 +318,9 @@ func New(name string, cfg SDKConfig) *PluginSDK {
if cfg.RegInput != nil { if cfg.RegInput != nil {
base.SetInputChannelRegistrar(cfg.RegInput) base.SetInputChannelRegistrar(cfg.RegInput)
} }
if cfg.RegOutputUnreg != nil {
base.SetOutputChannelUnregistrar(cfg.RegOutputUnreg)
}
base.SetMemoryAPI(cfg.Memory) base.SetMemoryAPI(cfg.Memory)
base.SetTextMemoryAPI(cfg.TextMemory) base.SetTextMemoryAPI(cfg.TextMemory)
base.SetDocMemoryAPI(cfg.DocMemory) base.SetDocMemoryAPI(cfg.DocMemory)

View File

@ -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 标记) ./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/<id>`,可用 `/kernel` 的 channels 观察
(在线时出现、掉线后消失)。
> **设计口径**:agent → 设备**必须**是 agent 的主动调用(`output_send__device/<id>`);
> 设备的上报(`op=event`)虽然会被注入成输入,但 agent 的回复**不会**被插件自动转回设备
> —— 全仓只有 **webui 与 cli** 两个交互界面"主动转发"(把最终回复渲染成气泡/终端输出),
> 其它通道(qq、设备等)一律要求显式 `output_send__<通道>`。
## 读结果 ## 读结果
| 指标 | 含义 | | 指标 | 含义 |

View File

@ -0,0 +1,160 @@
#!/usr/bin/env python3
"""最小远程设备客户端(WebSocket,无第三方依赖)。
用途:在**真内核**上验证"agent → 设备"这条出站链路 ——
`output_send__device/<id>` 应该以 `op=push` 的帧落到设备。
流程:握手 → hello(自报 id/kind/caps)→ bind → 打印收到的帧;
看到 push 就往 --out 文件里写一行(便于 shell 断言)。
注意:实例跑在私有 netns 里,设备网关的 127.0.0.1:9890 在 netns 内,
所以本脚本要用 nsenter 进同一个 netns 跑,例如:
nsenter -t <homed-pid> -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()

View File

@ -284,6 +284,13 @@ type InputChannelRegistrar func(name string, def ChannelDef) error
// OutputChannelRegistrar registers an output channel that the output_send tool can use. // 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 type OutputChannelRegistrar func(name string, caps int, desc string, def ChannelDef, handler ToolHandler) error
// OutputChannelUnregistrar 注销一个输出通道。
//
// 为什么需要它:输出通道不止有"启动时注册一次"的静态通道,还有**随外部资源生灭**的
// 动态通道 —— 典型是远程设备:`device/<id>` 只在设备在线期间存在,设备掉线后
// 必须注销,否则 output_list_channels 会一直列着它、模型会往一个死通道发消息。
type OutputChannelUnregistrar func(name string) error
// Output capability flags // Output capability flags
const ( const (
CapText = 1 CapText = 1
@ -301,6 +308,7 @@ type PluginSDK struct {
regStage StageRegistrar regStage StageRegistrar
regAPI APIRegistrar regAPI APIRegistrar
regOutput OutputChannelRegistrar regOutput OutputChannelRegistrar
regOutputUnreg OutputChannelUnregistrar
regInput InputChannelRegistrar regInput InputChannelRegistrar
io IOInjector io IOInjector
mem MemoryAPI mem MemoryAPI
@ -487,6 +495,17 @@ func (s *PluginSDK) RegisterOutputChannel(name string, caps int, desc string, de
return nil 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. // RegisterInputChannel registers an input channel with its memory behavior.
// //
// 契约:**凡是用 InjectText*/InjectInput*/InjectInterrupt*(source, "<name>", ...) // 契约:**凡是用 InjectText*/InjectInput*/InjectInterrupt*(source, "<name>", ...)
@ -518,6 +537,13 @@ func (s *PluginSDK) SetOutputChannelRegistrar(r OutputChannelRegistrar) {
s.apiMu.Unlock() 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). // SetInputChannelRegistrar sets the input channel registrar (called by the core at startup).
func (s *PluginSDK) SetInputChannelRegistrar(r InputChannelRegistrar) { func (s *PluginSDK) SetInputChannelRegistrar(r InputChannelRegistrar) {
s.apiMu.Lock() s.apiMu.Lock()