宜博 / lemonboss

/**
 * 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 };