mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-15 12:15:16 +00:00
clawhubadapter: 补充 chatPolls/getPolls 轮询消息源(通用通道插件输入闭环)
- 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 完整入站;微信通道回归正常
This commit is contained in:
@ -105,10 +105,22 @@ const registeredChannels = {}; // name -> { pluginName, channelPlugin, output, s
|
|||||||
function makeChannelRuntime(chName, ch) {
|
function makeChannelRuntime(chName, ch) {
|
||||||
return {
|
return {
|
||||||
id: chName,
|
id: chName,
|
||||||
// OC 通用通道轮询输入:manager 模式消息由插件自身 pollLoop 推送(经 deliver 入站),
|
// OC 通用通道轮询输入:Go 端经 channel/send 注入的消息放入 ch.pollQueue,
|
||||||
// 此处空转防止插件把 poll 判定为断连。
|
// 插件每次 chatPolls/getPolls 取走(poll 语义:取走即消费,不重复投递)。
|
||||||
chatPolls: async () => ({ msgs: [] }),
|
chatPolls: async (opts) => {
|
||||||
getPolls: async () => ({ msgs: [] }),
|
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 端作日志/降级
|
// 插件经 runtime 直接调用服务器 API:无目标服务器,转发 Go 端作日志/降级
|
||||||
call: async (method, args) => {
|
call: async (method, args) => {
|
||||||
notify('channel_output', { channel: chName, type: 'call', method, args });
|
notify('channel_output', { channel: chName, type: 'call', method, args });
|
||||||
@ -342,6 +354,7 @@ function loadPlugin(pluginDir, name) {
|
|||||||
registeredChannels[chName] = {
|
registeredChannels[chName] = {
|
||||||
pluginName: name, channelPlugin: chPlugin, type: chType,
|
pluginName: name, channelPlugin: chPlugin, type: chType,
|
||||||
deliverers: new Map(), accounts: {}, status: {},
|
deliverers: new Map(), accounts: {}, status: {},
|
||||||
|
pollQueues: new Map(),
|
||||||
};
|
};
|
||||||
// gateway 生命周期桥:fire-and-forget,绝不阻塞 registerChannel
|
// gateway 生命周期桥:fire-and-forget,绝不阻塞 registerChannel
|
||||||
startChannels(chName).catch((e) =>
|
startChannels(chName).catch((e) =>
|
||||||
@ -726,6 +739,25 @@ rl.on('line', async (line) => {
|
|||||||
return;
|
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') {
|
if (method === 'plugins/list') {
|
||||||
const list = Object.entries(loadedPlugins).map(([name, p]) => ({
|
const list = Object.entries(loadedPlugins).map(([name, p]) => ({
|
||||||
name,
|
name,
|
||||||
|
|||||||
@ -55,6 +55,9 @@ type Plugin struct {
|
|||||||
httpClient *http.Client
|
httpClient *http.Client
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// pluginSingleton 内核单例引用(Start 时设置),供 SendToChannel/ChannelSender 使用
|
||||||
|
var pluginSingleton *Plugin
|
||||||
|
|
||||||
func New(name, skillsDir string) *Plugin {
|
func New(name, skillsDir string) *Plugin {
|
||||||
sd := ""
|
sd := ""
|
||||||
if skillsDir != "" {
|
if skillsDir != "" {
|
||||||
@ -73,6 +76,7 @@ func (p *Plugin) Name() string { return p.name }
|
|||||||
func (p *Plugin) Start(s *sdk.PluginSDK) error {
|
func (p *Plugin) Start(s *sdk.PluginSDK) error {
|
||||||
s.SetAutoRestart(true)
|
s.SetAutoRestart(true)
|
||||||
p.sdk = s
|
p.sdk = s
|
||||||
|
pluginSingleton = p
|
||||||
|
|
||||||
s.Settings().RegisterDef(sdk.ConfigDef{
|
s.Settings().RegisterDef(sdk.ConfigDef{
|
||||||
Key: "skills_dir", Type: "string", DisplayName: "Skill 加载目录",
|
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)
|
p.dispatcher.Dispatch(params.Type, params.Data, pluginName, sp, s)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *Plugin) loadPySidecar(s *sdk.PluginSDK, dir, name string) error {
|
// 向通道注入外部输入(经 manager channel/send → pollQueue → 插件 chatPolls 轮询取走)。
|
||||||
simPath := filepath.Join(p.simulatorDir, "pysim.py")
|
// 供内核其他组件(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 {
|
if err := os.MkdirAll(p.simulatorDir, 0755); err != nil {
|
||||||
return fmt.Errorf("create simulator dir: %w", err)
|
return fmt.Errorf("create simulator dir: %w", err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user