宜博 / lemonboss

/**
 * CompanyHeartbeatService — 公司级心跳 + Pipeline 自动触发
 *
 * 职责:
 *   1. 定时扫描所有 heartbeat.enabled 的公司
 *   2. 检查公司内的 Pipeline 是否有 cron 调度
 *   3. 自动启动 Pipeline Run
 *   4. 驱动运行中的 Run：为每个 ready 步骤触发对应 Agent 执行
 *
 * 闭环:
 *   心跳 → 检查 Pipeline → 启动 Run → 触发 Agent → Agent 完成回调 → 推进步骤 → 触发下一个 Agent → ... → Run completed
 */

const fs = require('fs');
const path = require('path');
const http = require('http');
const config = require('../config');
const { CompanyService } = require('./company-service');
const { PipelineService } = require('./pipeline-service');
const { ScheduleService } = require('./schedule-service');

const DATA_DIR = path.resolve(__dirname, '../../data');

class CompanyHeartbeatService {

  constructor() {
    this.timers = new Map();      // companyId → NodeJS.Timeout
    this.companyService = new CompanyService();
    this.scheduleService = new ScheduleService();
    this.port = config.port || 3010;
    this._started = false;
  }

  /**
   * 启动公司级心跳服务
   */
  start() {
    if (this._started) return;
    this._started = true;

    console.log('[CompanyHB] 💓 启动公司级心跳服务...');
    const companies = this.companyService.list();

    for (const company of companies) {
      if (company.status === 'running') {
        this._setupCompanyHeartbeat(company.id);
      }
    }
    console.log(`[CompanyHB] ✅ 已挂载 ${this.timers.size} 家公司的心跳`);
  }

  /**
   * 停止所有心跳
   */
  stop() {
    for (const [id, timer] of this.timers.entries()) {
      clearInterval(timer);
    }
    this.timers.clear();
    this._started = false;
    console.log('[CompanyHB] 🛑 公司级心跳已停止');
  }

  /**
   * 添加/刷新单个公司的心跳
   */
  registerCompany(companyId) {
    const company = this.companyService.get(companyId);
    if (!company) return;

    // 先移除旧的
    if (this.timers.has(companyId)) {
      clearInterval(this.timers.get(companyId));
      this.timers.delete(companyId);
    }

    if (company.status === 'running' && company.heartbeat?.enabled) {
      this._setupCompanyHeartbeat(companyId);
    }
  }

  /**
   * 移除公司心跳
   */
  unregisterCompany(companyId) {
    if (this.timers.has(companyId)) {
      clearInterval(this.timers.get(companyId));
      this.timers.delete(companyId);
      console.log(`[CompanyHB] 🛑 移除 ${companyId} 的心跳`);
    }
  }

  // ═══════════════════════════════════════
  //  内部方法
  // ═══════════════════════════════════════

  _setupCompanyHeartbeat(companyId) {
    const company = this.companyService.get(companyId);
    if (!company || !company.heartbeat?.enabled) return;

    const intervalMs = (company.heartbeat.intervalMinutes || 30) * 60 * 1000;

    const timer = setInterval(() => {
      this._onHeartbeat(companyId);
    }, intervalMs);
    timer.unref();

    this.timers.set(companyId, timer);
    console.log(`[CompanyHB] ✅ ${companyId} 心跳已挂载 (间隔: ${company.heartbeat.intervalMinutes || 30} 分钟)`);
  }

  /**
   * 心跳触发时执行
   */
  async _onHeartbeat(companyId) {
    const timestamp = new Date().toISOString().replace('T', ' ').substring(0, 19);
    console.log(`[CompanyHB] 💓 ${companyId} 心跳触发 (${timestamp})`);

    const company = this.companyService.get(companyId);
    if (!company || company.status !== 'running') {
      this.unregisterCompany(companyId);
      return;
    }

    try {
      // 1. 检查并启动定时 Pipeline
      await this._checkAndStartPipelines(companyId);

      // 2. 驱动运行中的 Pipeline Run
      await this._driveActiveRuns(companyId);

      // 3. 执行日程驱动的任务 (Schedule)
      await this._executeScheduledTasks(companyId);

      // 4. 触发 CEO 心跳 (如果有 CEO 在公司内)
      if (company.ceo) {
        this._triggerCeoHeartbeat(companyId, company.ceo);
      }

      // 5. 检查是否需要生成 Master 晨会日报 (每天 08:00)
      this._checkDailyReport(companyId);
    } catch (err) {
      console.error(`[CompanyHB] ❌ ${companyId} 心跳执行失败:`, err.message);
      // 告警通道: 心跳本身异常也通知 Master
      this._sendAlertToMaster(companyId, `公司心跳执行异常: ${err.message}`);
    }
  }

  /**
   * 检查是否需要自动启动 Pipeline Run
   * 根据 Pipeline 的 cron 字段决定是否该触发 (简化版: 检查距上次运行的间隔)
   */
  async _checkAndStartPipelines(companyId) {
    const pSvc = new PipelineService(companyId);
    const pipelines = pSvc.listPipelines();

    for (const pipelineMeta of pipelines) {
      if (pipelineMeta.status !== 'active') continue;

      const pipeline = pSvc.getPipeline(pipelineMeta.id);
      if (!pipeline || !pipeline.cron) continue;

      // 检查是否应该启动 (基于 cron.intervalHours)
      const shouldRun = this._shouldRunPipeline(pSvc, pipeline);
      if (shouldRun) {
        const run = pSvc.startRun(pipeline.id, {
          trigger: 'heartbeat',
          input: { date: new Date().toISOString().slice(0, 10) },
        });
        console.log(`[CompanyHB] 🚀 ${companyId} 自动启动 Pipeline Run: ${run.runId} (${pipeline.name})`);
      }
    }
  }

  /**
   * 简化的 cron 判断: 距离上次运行是否超过了 interval
   */
  _shouldRunPipeline(pSvc, pipeline) {
    if (!pipeline.cron?.intervalHours) return false;

    const runs = pSvc.listRuns(pipeline.id);
    if (runs.length === 0) return true;  // 从未运行过

    const lastRun = runs[0]; // 已按创建时间倒序排列
    const lastRunTime = new Date(lastRun.createdAt).getTime();
    const now = Date.now();
    const intervalMs = pipeline.cron.intervalHours * 60 * 60 * 1000;

    return (now - lastRunTime) >= intervalMs;
  }

  /**
   * 驱动运行中的 Pipeline — 为 ready 步骤触发 Agent 执行
   */
  async _driveActiveRuns(companyId) {
    const pSvc = new PipelineService(companyId);
    const runs = pSvc.listRuns();

    for (const runMeta of runs) {
      if (runMeta.status !== 'running') continue;

      const task = pSvc.getCurrentTask(runMeta.runId);
      if (!task) continue;

      const run = pSvc.getRun(runMeta.runId);
      const step = run.steps[task.stepIndex];

      // 如果步骤已经是 in_progress，Agent 还在执行，跳过
      if (step.status === 'in_progress') {
        // 检查是否超时
        if (step.startedAt) {
          const pipeline = pSvc.getPipeline(run.pipelineId);
          const stepDef = pipeline?.steps?.[task.stepIndex];
          const timeoutMs = (stepDef?.timeout || 30) * 60 * 1000;
          const elapsed = Date.now() - new Date(step.startedAt).getTime();
          if (elapsed > timeoutMs) {
            console.log(`[CompanyHB] ⏰ ${companyId} Run ${run.runId} 步骤 ${task.stepIndex} 超时 (${Math.round(elapsed/60000)}分)，跳过`);
            pSvc.advanceStep(run.runId, task.stepIndex, {
              status: 'skipped',
              output: { reason: '超时跳过' },
            });
            // 🚨 告警通道: 超时通知 Master
            this._sendAlertToMaster(companyId,
              `Pipeline "${run.pipelineName}" 步骤 ${task.stepIndex + 1} (${task.stepName}) 执行超时 (${Math.round(elapsed/60000)}分钟)，已自动跳过。Agent: ${task.agent}`
            );
          }
        }
        continue;
      }

      // 步骤 ready → 触发 Agent 执行
      console.log(`[CompanyHB] 🎯 ${companyId} 触发 ${task.agent} 执行步骤: ${task.stepName}`);
      pSvc.markStepStarted(runMeta.runId, task.stepIndex);

      // 构建任务消息
      const pipeline = pSvc.getPipeline(run.pipelineId);
      const stepDef = pipeline?.steps?.[task.stepIndex];
      const inputStr = task.input ? `\n上一步的输出(作为你的输入):\n${JSON.stringify(task.input, null, 2)}` : '';

      // 检查是否有驳回信息
      const rejectionInfo = task.input?.__rejection
        ? `\n⚠️ ${task.input.__rejection.instruction}\n这是第 ${task.input.__rejection.attempt} 次重做。\n`
        : '';

      const taskMessage = [
        `[PIPELINE TASK]`,
        `流水线: ${run.pipelineName}`,
        `步骤 ${task.stepIndex + 1}/${run.steps.length}: ${task.stepName}`,
        `指令: ${stepDef?.prompt || '请执行你的职责'}`,
        rejectionInfo,
        inputStr,
        '',
        `完成后请调用以下命令汇报结果:`,
        `node bin/lemon.js task report --taskId pipeline-${runMeta.runId}-step-${task.stepIndex} --from ${companyId}:${task.agent} --to ${companyId}:coo --status done --summary "(你的完成摘要)"`,
      ].join('\n');

      // 通过 HTTP 触发 Agent Chat (异步，不阻塞心跳)
      this._sendAgentChat(companyId, task.agent, taskMessage, `pipeline-${runMeta.runId}-step-${task.stepIndex}`);
    }
  }

  /**
   * 触发 CEO 心跳
   */
  _triggerCeoHeartbeat(companyId, ceoId) {
    const timestamp = new Date().toISOString().replace('T', ' ').substring(0, 19);
    const msg = [
      `[COMPANY HEARTBEAT]`,
      `公司: ${companyId}`,
      `当前时间: ${timestamp}`,
      ``,
      `请执行以下检查:`,
      `1. 查看 todo.md 是否有待办任务`,
      `2. 查看团队成员的工作进度`,
      `3. 检查 Pipeline 运行状态`,
      `4. 如有偏差，主动 dispatch 纠偏指令`,
      `5. 无事则回复 "[心跳正常]"`,
    ].join('\n');

    this._sendAgentChat(companyId, ceoId, msg, `heartbeat-${Date.now().toString(36)}`);
  }

  // ═══════════════════════════════════════
  //  日程驱动执行 (Schedule)
  // ═══════════════════════════════════════

  /**
   * 执行当前时间到期的日程任务
   */
  async _executeScheduledTasks(companyId) {
    const dueSlots = this.scheduleService.getDueSlots(companyId);
    if (dueSlots.length === 0) return;

    for (const slot of dueSlots) {
      const agentId = `${companyId}:${slot.agent}`;
      const typeLabel = slot.type === 'meeting' ? '📋 会议' : slot.type === 'review' ? '🔍 复盘' : '📌 任务';
      const msg = [
        `[SCHEDULED TASK]`,
        `${typeLabel}: ${slot.task}`,
        `日程时间: ${slot.time}`,
        ``,
        `这是你的日程中预定的任务，请立即执行。`,
      ].join('\n');

      console.log(`[Schedule] 📅 ${companyId} 触发日程: ${slot.time} → ${slot.agent} (${slot.task})`);
      this._sendAgentChat(companyId, agentId, msg, `schedule-${slot.id}-${Date.now().toString(36)}`);
    }
  }

  // ═══════════════════════════════════════
  //  告警通道 (Alert Channel)
  // ═══════════════════════════════════════

  /**
   * 向 Master Agent 发送告警消息
   */
  _sendAlertToMaster(companyId, alertMessage) {
    const timestamp = new Date().toISOString().replace('T', ' ').substring(0, 19);
    const msg = [
      `[🚨 COMPANY ALERT]`,
      `公司: ${companyId}`,
      `时间: ${timestamp}`,
      ``,
      `告警内容:`,
      alertMessage,
      ``,
      `请评估是否需要介入处理。`,
    ].join('\n');

    // 发给 Master Agent
    this._sendAgentChat(null, 'master', msg, `alert-${companyId}-${Date.now().toString(36)}`);
    console.log(`[Alert] 🚨 已向 Master 发送告警: ${companyId} - ${alertMessage.substring(0, 80)}`);
  }

  // ═══════════════════════════════════════
  //  Master 晨会日报 (Daily Report)
  // ═══════════════════════════════════════

  /**
   * 检查是否需要生成每日晨会报告 (08:00 触发)
   */
  _checkDailyReport(companyId) {
    const now = new Date();
    const tz = 'Asia/Shanghai';
    const localTime = new Date(now.toLocaleString('en-US', { timeZone: tz }));
    const hour = localTime.getHours();
    const minute = localTime.getMinutes();

    // 仅在 08:00~08:14 范围内触发
    if (hour !== 8 || minute >= 15) return;

    // 防重复: 检查今天是否已生成
    const today = localTime.toISOString().slice(0, 10);
    const flagPath = path.join(DATA_DIR, companyId, '.daily-report-flag');
    try {
      const lastDate = fs.readFileSync(flagPath, 'utf-8').trim();
      if (lastDate === today) return;
    } catch { /* first time */ }

    // 生成日报
    const report = this._generateDailyReport(companyId);
    if (!report) return;

    // 写入标记
    fs.writeFileSync(flagPath, today, 'utf-8');

    // 发给 Master
    const msg = [
      `[📊 DAILY COMPANY REPORT]`,
      `公司: ${companyId}`,
      `日期: ${today}`,
      ``,
      report,
      ``,
      `请查看以上日报。如需调整策略或设定新目标，请下发指令给 CEO。`,
    ].join('\n');

    this._sendAgentChat(null, 'master', msg, `daily-report-${companyId}-${today}`);
    console.log(`[DailyReport] 📊 ${companyId} 晨会日报已发送给 Master`);
  }

  /**
   * 生成公司日报内容
   */
  _generateDailyReport(companyId) {
    try {
      const company = this.companyService.get(companyId);
      if (!company) return null;

      const lines = [`# ${company.name} 运营日报`, ''];

      // 1. 公司状态
      lines.push(`## 公司状态`);
      lines.push(`- 状态: ${company.status}`);
      lines.push(`- CEO: ${company.ceo || '(未任命)'}`);
      lines.push(`- Agent 数量: ${company.agents?.length || 0}`);
      lines.push('');

      // 2. Pipeline 运行状况
      const pSvc = new PipelineService(companyId);
      const pipelines = pSvc.listPipelines();
      const runs = pSvc.listRuns();
      const activeRuns = runs.filter(r => r.status === 'running');
      const completedToday = runs.filter(r => {
        const created = new Date(r.createdAt);
        const today = new Date();
        return r.status === 'completed' && created.toDateString() === today.toDateString();
      });

      lines.push(`## Pipeline 状况`);
      lines.push(`- 已定义流水线: ${pipelines.length} 条`);
      lines.push(`- 运行中: ${activeRuns.length} 个`);
      lines.push(`- 今日完成: ${completedToday.length} 个`);
      if (activeRuns.length > 0) {
        lines.push(`- 活跃详情:`);
        for (const r of activeRuns) {
          lines.push(`  - ${r.pipelineName}: 步骤 ${r.currentStep + 1}/${r.totalSteps}`);
        }
      }
      lines.push('');

      // 3. OKR 进度
      try {
        const { GoalsService } = require('./goals');
        const goals = new GoalsService(companyId);
        const goalList = goals.list();
        if (goalList.length > 0) {
          lines.push(`## OKR 进度`);
          for (const g of goalList) {
            const pct = Math.round((g.progress || 0) * 100);
            lines.push(`- ${g.id} ${g.title}: ${pct}% (${g.doneCount || 0}/${g.totalCount || 0})`);
          }
          lines.push('');
        }
      } catch { /* no goals */ }

      // 4. Agent 活跃度
      const agentsDir = path.join(DATA_DIR, companyId, 'agents');
      if (fs.existsSync(agentsDir)) {
        lines.push(`## Agent 活跃度`);
        const agents = fs.readdirSync(agentsDir, { withFileTypes: true })
          .filter(d => d.isDirectory());
        for (const a of agents) {
          const convDir = path.join(agentsDir, a.name, 'conversations');
          let lastActive = '(无记录)';
          if (fs.existsSync(convDir)) {
            const convs = fs.readdirSync(convDir).sort().reverse();
            if (convs.length > 0) {
              const metaPath = path.join(convDir, convs[0], 'meta.json');
              try {
                const meta = JSON.parse(fs.readFileSync(metaPath, 'utf-8'));
                lastActive = meta.updatedAt?.substring(0, 16) || '(未知)';
              } catch {}
            }
          }
          lines.push(`- ${a.name}: 最近活跃 ${lastActive}`);
        }
        lines.push('');
      }

      return lines.join('\n');
    } catch (err) {
      console.error(`[DailyReport] ❌ 生成日报失败:`, err.message);
      return null;
    }
  }

  /**
   * 通过 HTTP 向 Agent 发送消息 (异步, 不阻塞)
   * companyId 为 null 时发给全局 Agent (如 master)
   */
  _sendAgentChat(companyId, agentId, message, conversationId) {
    const postData = JSON.stringify({
      message,
      conversationId,
      isHeartbeat: true,
    });

    const options = {
      hostname: 'localhost',
      port: this.port,
      path: `/api/agent/${agentId}/chat`,
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Content-Length': Buffer.byteLength(postData),
      },
    };

    const req = http.request(options, (res) => {
      let body = '';
      res.on('data', (chunk) => body += chunk);
      res.on('end', () => {
        if (res.statusCode >= 200 && res.statusCode < 300) {
          console.log(`[CompanyHB] 📨 ${companyId}/${agentId} 任务下发成功 (${conversationId})`);
        } else {
          console.warn(`[CompanyHB] ⚠️ ${companyId}/${agentId} 触发异常: ${res.statusCode}`);
        }
      });
    });

    req.on('error', (err) => {
      console.error(`[CompanyHB] ❌ ${companyId}/${agentId} 请求失败: ${err.message}`);
    });

    req.write(postData);
    req.end();
  }
}

// 导出单例
const companyHeartbeatService = new CompanyHeartbeatService();
module.exports = { CompanyHeartbeatService, companyHeartbeatService };