From 3cc8d84def2428b8d9a676386fdaa82c25138155 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 2 Aug 2026 13:07:31 +0800 Subject: [PATCH] =?UTF-8?q?clawhubadapter:=20=E8=A1=A5=E5=85=85=20chatPoll?= =?UTF-8?q?s/getPolls=20=E8=BD=AE=E8=AF=A2=E6=B6=88=E6=81=AF=E6=BA=90?= =?UTF-8?q?=EF=BC=88=E9=80=9A=E7=94=A8=E9=80=9A=E9=81=93=E6=8F=92=E4=BB=B6?= =?UTF-8?q?=E8=BE=93=E5=85=A5=E9=97=AD=E7=8E=AF=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - manager：chatPolls/getPolls 改读 per-account pollQueue（poll 取走即消费）； 新增 channel/send RPC（Go 端注入外部输入） - plugin.go：SendToChannel(channel, payload) + ChannelSender() 单例 - 验证：mock-poll（纯 chatPolls 轮询型通道插件）channel/send → poll got msg → dispatchReply → channel_input 完整入站；微信通道回归正常 --- .../plugins/clawhubadapter/manager/main.js | 40 +++++++++++++++++-- internal/plugins/clawhubadapter/plugin.go | 28 ++++++++++++- 2 files changed, 62 insertions(+), 6 deletions(-) diff --git a/internal/plugins/clawhubadapter/manager/main.js b/internal/plugins/clawhubadapter/manager/main.js index a04062d..ea1c148 100644 --- a/internal/plugins/clawhubadapter/manager/main.js +++ b/internal/plugins/clawhubadapter/manager/main.js @@ -105,10 +105,22 @@ const registeredChannels = {}; // name -> { pluginName, channelPlugin, output, s function makeChannelRuntime(chName, ch) { return { id: chName, - // OC 通用通道轮询输入：manager 模式消息由插件自身 pollLoop 推送（经 deliver 入站）， - // 此处空转防止插件把 poll 判定为断连。 - chatPolls: async () => ({ msgs: [] }), - getPolls: async () => ({ msgs: [] }), + // OC 通用通道轮询输入：Go 端经 channel/send 注入的消息放入 ch.pollQueue， + // 插件每次 chatPolls/getPolls 取走（poll 语义：取走即消费，不重复投递）。 + chatPolls: async (opts) => { + const accountId = (opts && opts.accountId) || 'default'; + const limit = (opts && opts.limit) || 20; + const q = ch.pollQueues.get(accountId) || []; + const msgs = q.splice(0, limit).map((m) => ({ ...m })); + return { msgs }; + }, + getPolls: async (opts) => { + const accountId = (opts && opts.accountId) || 'default'; + const limit = (opts && opts.limit) || 20; + const q = ch.pollQueues.get(accountId) || []; + const msgs = q.splice(0, limit).map((m) => ({ ...m })); + return { msgs }; + }, // 插件经 runtime 直接调用服务器 API：无目标服务器，转发 Go 端作日志/降级 call: async (method, args) => { notify('channel_output', { channel: chName, type: 'call', method, args }); @@ -342,6 +354,7 @@ function loadPlugin(pluginDir, name) { registeredChannels[chName] = { pluginName: name, channelPlugin: chPlugin, type: chType, deliverers: new Map(), accounts: {}, status: {}, + pollQueues: new Map(), }; // gateway 生命周期桥：fire-and-forget，绝不阻塞 registerChannel startChannels(chName).catch((e) => @@ -726,6 +739,25 @@ rl.on('line', async (line) => { return; } + // 向通道注入输入：Go 端外部输入（webui 会话/其他插件）→ pollQueue → 插件 chatPolls 轮询取走 + if (method === 'channel/send') { + const chName = req.params?.channel; + const payload = req.params?.payload || {}; + const accountId = (req.params && req.params.accountId) || payload.accountId || 'default'; + const ch = registeredChannels[chName]; + if (!ch) { sendError(id, -32601, `channel not found: ${chName}`); return; } + ch.pollQueues.set(accountId, ch.pollQueues.get(accountId) || []); + ch.pollQueues.get(accountId).push({ + sessionKey: payload.sessionKey || `${chName}:${accountId}:${payload.from || 'poll'}`, + from: payload.from || '', + text: payload.content || payload.text || '', + type: payload.type || 'text', + timestamp: Date.now(), + }); + writeJSON({ jsonrpc: '2.0', id, result: { status: 'queued', channel: chName, accountId } }); + return; + } + if (method === 'plugins/list') { const list = Object.entries(loadedPlugins).map(([name, p]) => ({ name, diff --git a/internal/plugins/clawhubadapter/plugin.go b/internal/plugins/clawhubadapter/plugin.go index 6f02c84..3903768 100644 --- a/internal/plugins/clawhubadapter/plugin.go +++ b/internal/plugins/clawhubadapter/plugin.go @@ -55,6 +55,9 @@ type Plugin struct { httpClient *http.Client } +// pluginSingleton 内核单例引用（Start 时设置），供 SendToChannel/ChannelSender 使用 +var pluginSingleton *Plugin + func New(name, skillsDir string) *Plugin { sd := "" if skillsDir != "" { @@ -73,6 +76,7 @@ func (p *Plugin) Name() string { return p.name } func (p *Plugin) Start(s *sdk.PluginSDK) error { s.SetAutoRestart(true) p.sdk = s + pluginSingleton = p s.Settings().RegisterDef(sdk.ConfigDef{ Key: "skills_dir", Type: "string", DisplayName: "Skill 加载目录", @@ -817,8 +821,28 @@ func (p *Plugin) translateAndRegister(n OCNotification, sp *sidecarProcess, s *s p.dispatcher.Dispatch(params.Type, params.Data, pluginName, sp, s) } -func (p *Plugin) loadPySidecar(s *sdk.PluginSDK, dir, name string) error { - simPath := filepath.Join(p.simulatorDir, "pysim.py") +// 向通道注入外部输入（经 manager channel/send → pollQueue → 插件 chatPolls 轮询取走）。 +// 供内核其他组件（webui 会话、其他插件）向依赖 runtime 轮询的通用通道插件投递消息。 +func (p *Plugin) SendToChannel(channel string, payload map[string]interface{}) error { + p.mu.Lock() + m := p.manager + p.mu.Unlock() + if m == nil { + return fmt.Errorf("clawhubadapter manager not running") + } + _, err := m.call("channel/send", map[string]interface{}{ + "channel": channel, + "payload": payload, + }) + return err +} + +// ChannelSender 返回 clawhubadapter 单例，供内核其他组件注入通道输入（nil 表示未启动） +func ChannelSender() *Plugin { + return pluginSingleton +} + +func (p *Plugin) loadPySidecar(s *sdk.PluginSDK, dir, name string) error { simPath := filepath.Join(p.simulatorDir, "pysim.py") if err := os.MkdirAll(p.simulatorDir, 0755); err != nil { return fmt.Errorf("create simulator dir: %w", err) }