/** * plugin-manager.js — 插件生命周期管理 * * 负责插件系统的初始化、Channel 启动/停止、优雅关闭。 * 这是一个通用模块,不绑定任何具体 Channel。 * * 用法: * const { PluginManager } = require('./src/plugins/plugin-manager'); * const pm = new PluginManager(); * await pm.init(); // 加载配置 + 编译 + 注册 * await pm.start(); // 启动所有 channel monitor * await pm.shutdown(); // 优雅关闭 */ const path = require('path'); const fs = require('fs'); const { bootstrapPlugins } = require('./bootstrap'); const { loadPluginConfig, bridgeConfig } = require('./config-bridge'); const { createAgentBridge } = require('./agent-bridge'); class PluginManager { constructor(options = {}) { this.logger = options.logger || { info: (...a) => console.log('[plugins]', ...a), warn: (...a) => console.warn('[plugins]', ...a), error: (...a) => console.error('[plugins]', ...a), debug: () => { }, }; this._ctx = null; // bootstrapPlugins 返回值 this._pluginConfig = null; // plugins.json 原始配置 this._clawdbotConfig = null; // 桥接后的 ClawdbotConfig this._bridge = null; // Agent 桥接器 this._abortController = null; this._monitors = []; // 运行中的 channel monitor Promise this._state = 'idle'; // idle → initialized → running → stopped } /** 当前状态 */ get state() { return this._state; } /** 插件是否已启用(plugins.json 存在且有 channel 配置) */ get enabled() { return this._pluginConfig?.channels && Object.keys(this._pluginConfig.channels).length > 0; } /** 获取所有已注册的 tools */ get tools() { return this._ctx?.getToolNames() || []; } /** 获取所有已注册的 channels */ get channels() { return this._ctx?.getChannels() || []; } /** 获取桥接统计 */ get stats() { return this._bridge?.getStats() || { total: 0, success: 0, error: 0 }; } /** * 阶段 1:初始化——加载配置、通过 OpenClaw 原生流程发现 + 注册插件。 */ async init() { if (this._state !== 'idle') { this.logger.warn(`init() 时状态为 ${this._state},跳过`); return; } // 1. 加载配置 this.logger.info('加载插件配置...'); this._pluginConfig = loadPluginConfig(); if (!this.enabled) { this.logger.info('未配置任何 channel,插件系统不启动'); this._state = 'initialized'; return; } const channelNames = Object.keys(this._pluginConfig.channels); this.logger.info(`发现 ${channelNames.length} 个 channel 配置: [${channelNames.join(', ')}]`); // 2. 桥接配置(把 plugins.json 格式 → OpenClaw 原生格式) this._clawdbotConfig = bridgeConfig(this._pluginConfig); // 3. 解析路径 const configDir = path.resolve(__dirname, '../../config'); let openclawRoot; if (this._pluginConfig.openclawRoot) { openclawRoot = path.resolve(configDir, this._pluginConfig.openclawRoot); } else { try { openclawRoot = require('@mas/plugin-bootstrap').resolveOpenclawDist(__dirname); } catch (err) { this.logger.error('无法定位 @mas/openclaw-dist,插件系统不可用:', err.message); this._state = 'initialized'; return; } } // 4. 加载——调用 1:1 复刻的 OpenClaw loadOpenClawPlugins this.logger.info('初始化插件系统 (OpenClaw 原生流程)...'); const t0 = Date.now(); this._ctx = await bootstrapPlugins({ openclawRoot, clawdbotConfig: this._clawdbotConfig, pluginsConfig: { enabled: true, load: { paths: (this._pluginConfig.extensionPaths || []).map(p => path.resolve(configDir, p)) }, entries: this._clawdbotConfig.plugins?.entries || {}, }, logger: this.logger, }); const elapsed = ((Date.now() - t0) / 1000).toFixed(1); this.logger.info(`插件加载完成 (${elapsed}s): ${this._ctx.getToolNames().length} tools, ${this._ctx.getChannels().length} channels`); this._state = 'initialized'; } /** * 阶段 2:启动所有 channel 的 monitor(如 feishu WebSocket)。 * * @param {Object} [orchestrator] - 编排器实例,提供 runOrchestration 方法 */ async start(orchestrator) { if (this._state !== 'initialized') { this.logger.warn(`start() 时状态为 ${this._state},需先 init()`); throw new Error(`Cannot start: current state is "${this._state}", need "initialized"`); } if (!this.enabled || !this._ctx) { this.logger.info('无 channel 需要启动'); return; } // ── 核心路由:MasterSession(与 Web UI 一致) ── if (orchestrator) { this._bridge = createAgentBridge(orchestrator, { logger: this.logger }); this.logger.info('Agent 桥接器已创建'); // MasterSession 会话池(按 feishu sessionKey 区分用户) const { MasterSession } = require('../core/master-session'); const { loadConfigFromSkills } = require('../core/skill-config-loader'); const masterSessions = new Map(); // sessionKey → { session, lastActive, timer } const IDLE_TIMEOUT = 15 * 60 * 1000; // 15 分钟空闲超时 function getMasterSession(sessionKey) { let entry = masterSessions.get(sessionKey); if (entry) { entry.lastActive = Date.now(); if (entry.timer) { clearTimeout(entry.timer); entry.timer = null; } return entry.session; } // 加载最新配置(读 SQLite skills DB via skill-config-loader) const configDir = path.resolve(__dirname, '../../config'); const backends = JSON.parse(fs.readFileSync(path.join(configDir, 'backends.json'), 'utf-8')); const skillsDir = path.resolve(__dirname, '../../../skills'); const config = loadConfigFromSkills(skillsDir, backends); const workDir = skillsDir; const session = new MasterSession(config, workDir); entry = { session, lastActive: Date.now(), timer: null }; masterSessions.set(sessionKey, entry); return session; } function resetIdleTimer(sessionKey) { const entry = masterSessions.get(sessionKey); if (!entry) return; if (entry.timer) clearTimeout(entry.timer); entry.timer = setTimeout(async () => { try { await entry.session.shutdown(); } catch { } masterSessions.delete(sessionKey); }, IDLE_TIMEOUT); } const logger = this.logger; this._ctx.runtime.channel.reply.dispatchReplyFromConfig = async (params) => { const { ctx, cfg, dispatcher, replyOptions } = params; const userMessage = ctx.BodyForAgent || ctx.RawBody || ctx.Body || ''; const sessionKey = ctx.SessionKey || 'default'; const senderId = ctx.SenderId || 'unknown'; const senderName = ctx.SenderName || senderId; logger.info(`[master] 收到飞书消息: sender=${senderName}, length=${userMessage.length}`); try { // 通过 MasterSession 处理(与 Web UI /api/master/chat 一致) const session = getMasterSession(sessionKey); // 先发一条"处理中"提示 dispatcher.sendBlockReply({ text: '🤖 正在处理,请稍候...' }); const result = await session.handleMessage(userMessage, { onData: () => { }, onToolCall: (tc) => { if (tc.tool === 'call_sub_agent' && tc.params.agent) { logger.info(`[master] 调用 Sub Agent: ${tc.params.agent}`); } }, onSubAgentData: () => { }, maxToolRounds: 10, }); const replyText = result.output || result.text || '(处理完成,无输出)'; logger.info(`[master] 回复 (${replyText.length}字): ${replyText.substring(0, 200)}${replyText.length > 200 ? '...' : ''}`); if (typeof replyOptions?.onReplyStart === 'function') replyOptions.onReplyStart(); const queuedFinal = dispatcher.sendFinalReply({ text: replyText }); const counts = dispatcher.getQueuedCounts(); resetIdleTimer(sessionKey); return { queuedFinal, counts }; } catch (err) { logger.error(`[master] 处理失败: ${err.message}`); try { dispatcher.sendFinalReply({ text: `处理失败: ${err.message}` }); } catch (sendErr) { logger.error(`[master] 发送错误回复也失败了: ${sendErr.message}`); } return { queuedFinal: false, counts: dispatcher.getQueuedCounts?.() || { final: 0, block: 0 } }; } }; // 保存 masterSessions 引用供 shutdown 使用 this._masterSessions = masterSessions; this.logger.info('已劫持 dispatchReplyFromConfig → MasterSession (与 Web UI 一致)'); } // 为所有已注册的 channel 启动 monitor this._abortController = new AbortController(); const channels = this._ctx.getChannels(); for (const ch of channels) { const id = ch.pluginId || ch.id || ch.plugin?.id; // 检查该 channel 在配置中是否真正可用(有凭证) const channelConfig = this._pluginConfig.channels[id]; if (!channelConfig || channelConfig.enabled === false) { this.logger.info(`channel/${id}: 未启用,跳过`); continue; } this.logger.info(`channel/${id}: 启动 monitor...`); try { const monitorPromise = this._startChannelMonitor(id, channelConfig); // 捕获异步错误,防止静默失败 monitorPromise.catch(err => { this.logger.error(`channel/${id}: monitor 异步错误: ${err.message}`); this.logger.error(err.stack); }); this._monitors.push({ id, promise: monitorPromise }); } catch (err) { this.logger.error(`channel/${id}: 启动失败: ${err.message}`); this.logger.error(err.stack); } } this._state = 'running'; if (this._monitors.length > 0) { this.logger.info(`${this._monitors.length} 个 channel monitor 已启动`); } } /** * 启动单个 channel 的 monitor。 * 根据 channel id 找到对应的 monitor 函数并调用。 */ async _startChannelMonitor(channelId, channelConfig) { const channel = this._ctx.getChannel(channelId); if (!channel) { throw new Error(`channel/${channelId} 未注册`); } // OpenClaw RuntimeEnv 格式:{ log: Function, error: Function } const runtimeEnv = { log: (...a) => this.logger.info(`[${channelId}]`, ...a), error: (...a) => this.logger.error(`[${channelId}]`, ...a), }; // 方式 1:如果 plugin 有 gateway.startAccount(OpenClaw 标准方式) if (channel.plugin?.gateway?.startAccount) { this.logger.info(`channel/${channelId}: 使用 gateway.startAccount 启动`); return channel.plugin.gateway.startAccount({ cfg: this._clawdbotConfig, runtime: runtimeEnv, abortSignal: this._abortController.signal, accountId: 'default', // ctx.log 需要 { info, warn, error, debug } 结构 log: { info: (...a) => this.logger.info(`[${channelId}]`, ...a), warn: (...a) => this.logger.warn(`[${channelId}]`, ...a), error: (...a) => this.logger.error(`[${channelId}]`, ...a), debug: (...a) => this.logger.debug(`[${channelId}]`, ...a), }, setStatus: (status) => { this.logger.info(`channel/${channelId}: status=${JSON.stringify(status)}`); }, }); } // 方式 2:fallback —— 直接用 jiti 加载 monitor 模块 const configDir = path.resolve(__dirname, '../../config'); let openclawRoot; if (this._pluginConfig.openclawRoot) { openclawRoot = path.resolve(configDir, this._pluginConfig.openclawRoot); } else { try { openclawRoot = require('@mas/plugin-bootstrap').resolveOpenclawDist(__dirname); } catch (err) { this.logger.warn(`channel/${channelId}: 无法定位 openclaw-dist,跳过 fallback monitor`); return; } } const monitorPaths = [ path.join(openclawRoot, `extensions/${channelId}/src/monitor.ts`), path.join(openclawRoot, `extensions/${channelId}/src/monitor.js`), ]; const monitorPath = monitorPaths.find(p => fs.existsSync(p)); if (!monitorPath) { this.logger.warn(`channel/${channelId}: 无 gateway 也无 monitor 模块,仅注册 channel/tools`); return; } this.logger.info(`channel/${channelId}: 使用 fallback monitor 启动`); const monitorModule = this._ctx.jiti(monitorPath); const startFn = monitorModule.monitorFeishuProvider || monitorModule.startMonitor || monitorModule.default; if (typeof startFn !== 'function') { this.logger.warn(`channel/${channelId}: monitor 模块无可用启动函数`); return; } return startFn({ config: this._clawdbotConfig, runtime: runtimeEnv, abortSignal: this._abortController.signal, }); } /** * 优雅关闭所有 channel monitor。 */ async shutdown() { if (this._state === 'stopped' || this._state === 'idle') return; this.logger.info('正在关闭插件系统...'); this._abortController?.abort(); // 等待所有 monitor 退出 for (const { id, promise } of this._monitors) { try { await Promise.race([promise, new Promise(r => setTimeout(r, 5000))]); this.logger.info(`channel/${id}: 已停止`); } catch (err) { this.logger.warn(`channel/${id}: 关闭时出错: ${err.message}`); } } this._monitors = []; this._state = 'stopped'; if (this._masterSessions) { for (const [key, entry] of this._masterSessions) { if (entry.timer) clearTimeout(entry.timer); try { await entry.session.shutdown(); } catch { } } this._masterSessions.clear(); this.logger.info(`MasterSession 会话池已清理`); } if (this._bridge) { const stats = this._bridge.getStats(); this.logger.info(`运行统计 — 总消息: ${stats.total}, 成功: ${stats.success}, 失败: ${stats.error}`); } this.logger.info('插件系统已关闭'); } /** * 获取状态摘要(用于 health check API)。 */ summary() { return { state: this._state, channels: this._ctx?.getChannels().map(c => c.id) || [], tools: this._ctx?.getToolNames() || [], monitors: this._monitors.map(m => m.id), stats: this.stats, }; } } module.exports = { PluginManager };