From 76ef49e9a688aeb4c185018b999183ebf459483b Mon Sep 17 00:00:00 2001 From: root Date: Sun, 2 Aug 2026 13:03:49 +0800 Subject: [PATCH] =?UTF-8?q?clawhubadapter:=20OpenClaw=20=E9=80=9A=E9=81=93?= =?UTF-8?q?=E6=8F=92=E4=BB=B6=E5=85=BC=E5=AE=B9=E4=BF=AE=E5=A4=8D=EF=BC=88?= =?UTF-8?q?gateway=20=E7=94=9F=E5=91=BD=E5=91=A8=E6=9C=9F=E6=A1=A5=20+=20c?= =?UTF-8?q?hannelRuntime=20+=20deliver=20=E4=BA=8B=E4=BB=B6=E5=BC=8F?= =?UTF-8?q?=E8=BE=93=E5=87=BA=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - manager/main.js：makeChannelRuntime（dispatchReplyWithBufferedBlockDispatcher 入站 + deliver 出站 绑定 + typingCallbacks）、startChannels/stopChannels（gateway.startAccount fire-and-forget 生命周期 桥、listAccountIds/resolveAccount 规范签名 cfg 传参）、tools/call 通道分支无 outbound 走 deliver 事件式发送、SIGTERM 优雅停靠、console 输出重定向 stderr 防 JSON-RPC 流污染 - plugin.go：channel_input 改 InjectInputSync 同步注入取回复并经 CallTool 回发通道（修复 InjectInterruptText 无 ResponseCh 致回复静默丢弃）；channel_status/channel_output 通知接入 - registry.go：channelStatus 状态缓存 - SDK：公共 IOInjector 增加 InjectInputSync(source, channel, text) string + ioAdapter 实现 - 生产验证：微信发消息 → pollLoop → dispatchReply → mock LLM 回文本 → deliver → ilink/bot/sendmessage status=200 送达 --- .../plugins/clawhubadapter/manager/main.js | 200 +++++++++++++++++- internal/plugins/clawhubadapter/plugin.go | 55 ++++- internal/plugins/clawhubadapter/registry.go | 2 + internal/sdk/plugin.go | 15 ++ third_party/homeagent-sdk/sdk/plugin.go | 13 ++ 5 files changed, 277 insertions(+), 8 deletions(-) diff --git a/internal/plugins/clawhubadapter/manager/main.js b/internal/plugins/clawhubadapter/manager/main.js index 1226b18..a04062d 100644 --- a/internal/plugins/clawhubadapter/manager/main.js +++ b/internal/plugins/clawhubadapter/manager/main.js @@ -3,6 +3,10 @@ const path = require('path'); const { execSync } = require('child_process'); // ---- Utility ---- +// stdout 仅承载 JSON-RPC；插件/manager 的 console 输出一律走 stderr，避免污染协议流 +const rawLog = console.log.bind(console); +console.log = (...args) => process.stderr.write(args.map(String).join(' ') + '\n'); + function writeJSON(obj) { process.stdout.write(JSON.stringify(obj) + '\n'); } @@ -93,7 +97,156 @@ function tryDetectOCPackage(pkgDir, pkgName) { const loadedPlugins = {}; // name -> { entry, tools: [{name, execute, ...}] } const allTools = []; // flat list of all tools across all plugins const allProviders = {}; // type -> { name, instance } across all plugins -const registeredChannels = {}; // name -> { pluginName, channelPlugin, output, send, type } +const registeredChannels = {}; // name -> { pluginName, channelPlugin, output, send, type, runtime, deliverers, accounts, status } + +// ---- OpenClaw channelRuntime mock ---- +// 真实通道插件依赖 channelRuntime 完成：入站转发（dispatchReplyWithBufferedBlockDispatcher）、 +// 出站回调（dispatcherOptions.deliver，插件自带的发送实现）、poll 轮询与 call 透传。 +function makeChannelRuntime(chName, ch) { + return { + id: chName, + // OC 通用通道轮询输入：manager 模式消息由插件自身 pollLoop 推送（经 deliver 入站）， + // 此处空转防止插件把 poll 判定为断连。 + chatPolls: async () => ({ msgs: [] }), + getPolls: async () => ({ msgs: [] }), + // 插件经 runtime 直接调用服务器 API：无目标服务器，转发 Go 端作日志/降级 + call: async (method, args) => { + notify('channel_output', { channel: chName, type: 'call', method, args }); + return { ok: true }; + }, + reply: { + // 入站消息 + 出站 deliver 绑定。ctx 来自真实插件（Body/From/SessionKey/AccountId...） + dispatchReplyWithBufferedBlockDispatcher: async (opts) => { + const ctx = (opts && opts.ctx) || {}; + const dopts = (opts && opts.dispatcherOptions) || {}; + const accountId = ctx.AccountId || ctx.accountId || 'default'; + ch.deliverers.set(accountId, { + deliver: dopts.deliver, + typingCallbacks: dopts.typingCallbacks || {}, + }); + notify('channel_input', { + channel: chName, + payload: { + content: ctx.BodyForAgent || ctx.Body || ctx.RawBody || '', + from: ctx.From || ctx.SenderId || '', + sessionKey: ctx.SessionKey || '', + accountId, + messageSid: ctx.MessageSid || '', + chatType: ctx.ChatType || 'direct', + raw: ctx, + }, + }); + const buffer = []; + return { + sendNow: async (items) => { + for (const item of (items || [])) await deliverItem(ch, accountId, item); + }, + addToBuffer: async (item) => { buffer.push(item); }, + sendBuffer: async () => { + for (const item of buffer) await deliverItem(ch, accountId, item); + buffer.length = 0; + }, + closeBuffer: async () => {}, + }; + }, + }, + }; +} + +async function deliverItem(ch, accountId, item) { + const d = ch.deliverers.get(accountId); + if (!d || typeof d.deliver !== 'function') return; + const payload = { + text: (item && item.text) || '', + mediaUrls: (item && (item.mediaUrls || (item.mediaUrl ? [item.mediaUrl] : []))) || [], + }; + if (d.typingCallbacks && typeof d.typingCallbacks.onReplyStart === 'function') { + try { await d.typingCallbacks.onReplyStart(payload); } catch {} + } + await d.deliver(payload); + if (d.typingCallbacks && typeof d.typingCallbacks.onCleanup === 'function') { + try { await d.typingCallbacks.onCleanup(payload); } catch {} + } +} + +// ---- OpenClaw gateway 生命周期桥 ---- +// 通道插件的心跳/收消息/状态上报都挂在 gateway.startAccount 上；startAccount 会永久挂起 +// （await new Promise(()=>{})），配合 OC health-monitor 判定账号存活，必须 fire-and-forget。 +async function startChannels(name) { + const ch = registeredChannels[name]; + if (!ch || !ch.channelPlugin) return; + const gateway = ch.channelPlugin.gateway; + if (!gateway || typeof gateway.startAccount !== 'function') return; + + ch.runtime = ch.runtime || makeChannelRuntime(name, ch); + ch.accounts = ch.accounts || {}; + ch.status = ch.status || { running: false, connected: false }; + + const chCfg = ch.channelPlugin.config || {}; + const passCfg = ch.pluginConfig || chCfg; + let accountIds = []; + try { + if (typeof chCfg.listAccountIds === 'function') { + // OC 规范签名: listAccountIds(cfg) + const ids = chCfg.listAccountIds(passCfg); + accountIds = (ids && ids.then ? await ids : (Array.isArray(ids) ? ids : [])); + } + } catch (e) { + process.stderr.write(`[manager] ${name}: listAccountIds failed: ${e.message}\n`); + } + if (!Array.isArray(accountIds) || accountIds.length === 0) accountIds = ['default']; + + for (const accountId of accountIds) { + let account = null; + try { + if (typeof chCfg.resolveAccount === 'function') { + // OC 规范签名: resolveAccount(cfg, accountId) + account = chCfg.resolveAccount(passCfg, accountId); + if (account && account.then) account = await account; + } + } catch (e) { + process.stderr.write(`[manager] ${name}: resolveAccount(${accountId}) failed: ${e.message}\n`); + } + if (!account) account = { accountId }; + + if (ch.accounts[accountId]) continue; + ch.accounts[accountId] = { started: true }; + + const getStatus = () => ({ ...ch.status, accountId }); + const setStatus = (patch) => { + ch.status = { ...ch.status, ...(patch || {}) }; + notify('channel_status', { channel: name, status: { ...ch.status, accountId } }); + }; + + const ctx = { + account, + cfg: ch.pluginConfig || chCfg, + channelRuntime: ch.runtime, + getStatus, + setStatus, + }; + + gateway.startAccount(ctx).catch((err) => { + process.stderr.write(`[manager] ${name}: startAccount(${accountId}) crashed: ${err.message}\n`); + ch.status = { ...ch.status, running: false, connected: false, error: err.message }; + notify('channel_status', { channel: name, status: { ...ch.status, accountId } }); + }); + } +} + +function stopChannels(name) { + const ch = registeredChannels[name]; + if (!ch || !ch.channelPlugin || !ch.accounts) return; + const gateway = ch.channelPlugin.gateway; + if (!gateway || typeof gateway.stopAccount !== 'function') return; + for (const accountId of Object.keys(ch.accounts)) { + const account = { accountId }; + gateway.stopAccount({ account, channelRuntime: ch.runtime, cfg: ch.pluginConfig || {} }).catch((err) => { + process.stderr.write(`[manager] ${name}: stopAccount(${accountId}) failed: ${err.message}\n`); + }); + } + ch.accounts = {}; +} function registerPluginTools(name, tools, api) { for (const t of tools) { @@ -182,11 +335,17 @@ function loadPlugin(pluginDir, name) { let chType = ch.type || 'text'; const chPlugin = ch.plugin; - // OpenClaw ChannelPlugin 格式: { plugin: { id, outbound: { sendText, sendMedia }, ... } } + // OpenClaw ChannelPlugin 格式: { plugin: { id, gateway: {startAccount, stopAccount}, config: {...}, ... } } if (chPlugin && typeof chPlugin === 'object') { chName = chName || chPlugin.id || chPlugin.meta?.id || name + '-channel'; chType = chType || (chPlugin.capabilities?.media ? 'io' : 'text'); - registeredChannels[chName] = { pluginName: name, channelPlugin: chPlugin, type: chType }; + registeredChannels[chName] = { + pluginName: name, channelPlugin: chPlugin, type: chType, + deliverers: new Map(), accounts: {}, status: {}, + }; + // gateway 生命周期桥：fire-and-forget，绝不阻塞 registerChannel + startChannels(chName).catch((e) => + process.stderr.write(`[manager] ${name}: startChannels failed: ${e.message}\n`)); } else { // 简单格式: { name, type, output } registeredChannels[chName] = { pluginName: name, output: ch.output || ch.send, type: chType }; @@ -717,12 +876,14 @@ rl.on('line', async (line) => { if (ch) { try { const channelPlugin = ch.channelPlugin; + let metaObj = {}; + try { metaObj = typeof args.meta === 'string' ? JSON.parse(args.meta) : (args.meta || {}); } catch {} + const accountId = metaObj.accountId || 'default'; + if (channelPlugin && channelPlugin.outbound) { - const meta = args.meta || ''; - let metaObj = {}; - try { metaObj = typeof meta === 'string' ? JSON.parse(meta) : meta; } catch {} + // 旧格式 outbound 直发 const to = metaObj.user_id || metaObj.to || metaObj.group_id || ''; - const ctx = { to, text: args.payload || '', mediaUrl: metaObj.mediaUrl || '', cfg: {}, accountId: metaObj.accountId || null }; + const ctx = { to, text: args.payload || '', mediaUrl: metaObj.mediaUrl || '', cfg: {}, accountId }; let result; if (ctx.mediaUrl && channelPlugin.outbound.sendMedia) { result = await channelPlugin.outbound.sendMedia(ctx); @@ -738,6 +899,20 @@ rl.on('line', async (line) => { } else if (typeof ch.send === 'function') { const result = await ch.send(args.payload, args.meta); writeJSON({ jsonrpc: '2.0', id, result: { status: 'sent', result } }); + } else if (ch.deliverers && ch.deliverers.has(accountId)) { + // OpenClaw ChannelPlugin 事件式发送：回复交给插件在入站时挂载的 deliver（插件自带发送实现） + await deliverItem(ch, accountId, { + text: args.payload || '', + mediaUrls: metaObj.mediaUrl ? [metaObj.mediaUrl] : [], + }); + writeJSON({ jsonrpc: '2.0', id, result: { status: 'sent', via: 'channelRuntime.deliver' } }); + } else if (ch.runtime) { + // 通道已启动但尚未收到入站消息（deliver 未建立）：降级为 channel_output 事件 + notify('channel_output', { + channel: toolName, type: 'message', + text: args.payload || '', mediaUrl: metaObj.mediaUrl || '', to: metaObj.user_id || '', + }); + writeJSON({ jsonrpc: '2.0', id, result: { status: 'queued', via: 'channel_output' } }); } else { sendError(id, -32601, `channel ${toolName} has no output handler`); } @@ -788,3 +963,14 @@ rl.on('line', async (line) => { sendError(id, -32601, `Method not found: ${method}`); }); + +// ---- 优雅停靠：进程退出前逐个 stopAccount（插件停止心跳/轮询） ---- +let shuttingDown = false; +async function shutdown() { + if (shuttingDown) return; + shuttingDown = true; + for (const name of Object.keys(registeredChannels)) stopChannels(name); + setTimeout(() => process.exit(0), 2000); +} +process.on('SIGTERM', shutdown); +process.on('SIGINT', shutdown); diff --git a/internal/plugins/clawhubadapter/plugin.go b/internal/plugins/clawhubadapter/plugin.go index cb7ae65..6f02c84 100644 --- a/internal/plugins/clawhubadapter/plugin.go +++ b/internal/plugins/clawhubadapter/plugin.go @@ -748,7 +748,60 @@ func (p *Plugin) translateAndRegister(n OCNotification, sp *sidecarProcess, s *s data, _ := json.Marshal(params.Payload) content = string(data) } - s.InjectInterruptText(pluginName, params.Channel, fmt.Sprintf("[%s] %s", params.Channel, content)) + // 同步注入并取回复，再把回复送回通道（agent → output_send__<通道> → deliver → 微信） + out := s.InjectInputSync(pluginName, params.Channel, "text", map[string]interface{}{ + "content": content, + }) + if out != nil { + reply, _ := out.Payload["content"].(string) + if reply != "" { + meta := map[string]interface{}{} + if from, _ := params.Payload["from"].(string); from != "" { + meta["user_id"] = from + } + if acc, _ := params.Payload["accountId"].(string); acc != "" { + meta["accountId"] = acc + } + go func() { + if _, err := sp.CallTool(params.Channel, map[string]interface{}{ + "payload": reply, + "meta": meta, + }); err != nil { + log.Printf("[clawhubadapter] channel %s reply dispatch failed: %v", params.Channel, err) + } + }() + } + } + return + } + if n.Method == "channel_status" { + var params struct { + Channel string `json:"channel"` + Status map[string]interface{} `json:"status"` + } + if err := json.Unmarshal(n.Params, ¶ms); err != nil || params.Channel == "" { + return + } + channelStatusMu.Lock() + if channelStatus == nil { + channelStatus = make(map[string]map[string]interface{}) + } + channelStatus[params.Channel] = params.Status + channelStatusMu.Unlock() + log.Printf("[clawhubadapter] channel_status %s: running=%v connected=%v", params.Channel, + params.Status["running"], params.Status["connected"]) + return + } + if n.Method == "channel_output" { + // 降级输出事件（通道已启动但 deliver 尚未建立）：仅记录，消息不丢失于协议层 + var params struct { + Channel string `json:"channel"` + Type string `json:"type"` + Text string `json:"text"` + } + if json.Unmarshal(n.Params, ¶ms) == nil && params.Channel != "" { + log.Printf("[clawhubadapter] channel_output %s (type=%s): %.120s", params.Channel, params.Type, params.Text) + } return } if n.Method != "register" { diff --git a/internal/plugins/clawhubadapter/registry.go b/internal/plugins/clawhubadapter/registry.go index ee93654..ec87505 100644 --- a/internal/plugins/clawhubadapter/registry.go +++ b/internal/plugins/clawhubadapter/registry.go @@ -13,6 +13,8 @@ import ( var ( channelInputBuf = map[string][]map[string]interface{}{} channelInputBufMu sync.Mutex + channelStatus map[string]map[string]interface{} // 通道运行状态（channel_status 通知） + channelStatusMu sync.Mutex ) type ToolRegistry struct{} diff --git a/internal/sdk/plugin.go b/internal/sdk/plugin.go index 9afa0c2..83c22ab 100644 --- a/internal/sdk/plugin.go +++ b/internal/sdk/plugin.go @@ -118,6 +118,21 @@ func (a ioAdapter) InjectInterruptText(source, channel, text string) { } } +// InjectInputSync 同步注入输入并等待回复（阻塞直至 agent 处理完成），返回回复文本。 +func (a ioAdapter) InjectInputSync(source, channel, text string) string { + if a.iom == nil { + return "" + } + out := a.iom.InjectInputSyncTo(source, channel, "text", map[string]interface{}{ + "content": text, + }) + if out == nil { + return "" + } + reply, _ := out.Payload["content"].(string) + return reply +} + func (a ioAdapter) InjectText(source, channel, text string) { if a.iom != nil { a.iom.InjectInputTo(source, channel, "text", map[string]interface{}{"content": text}) diff --git a/third_party/homeagent-sdk/sdk/plugin.go b/third_party/homeagent-sdk/sdk/plugin.go index 8285eba..3a446c0 100644 --- a/third_party/homeagent-sdk/sdk/plugin.go +++ b/third_party/homeagent-sdk/sdk/plugin.go @@ -111,6 +111,9 @@ type IOInjector interface { InjectInterruptText(source, channel, text string) InjectText(source, channel, text string) InjectTextNoMemory(source, channel, text string) + // InjectInputSync 注入输入事件并同步等待 agent 回复，返回回复文本（无回复时返回空串）。 + // 用于通道消息的完整闭环：收到入站 → agent 处理 → 回复取回 → 送回通道。 + InjectInputSync(source, channel, text string) string } // EventType identifies the kind of system event. @@ -364,6 +367,16 @@ func (s *PluginSDK) InjectTextNoMemory(source, channel, text string) { } } +// InjectInputSync injects a text message and synchronously waits for the agent reply, +// returning the reply text (empty string if none). Replies must be dispatched back +// to the source channel by the caller. +func (s *PluginSDK) InjectInputSync(source, channel, text string) string { + if s.io == nil { + return "" + } + return s.io.InjectInputSync(source, channel, text) +} + // SetAutoRestart 设置插件是否允许内核自动重启（崩溃后自动重载）。 // 默认 true。如果插件有无法恢复的状态（如外部连接），应设为 false。 func (s *PluginSDK) SetAutoRestart(enabled bool) { s.autoRestart = enabled }