宜博 / 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 { CompanyService } = require('./company-service');
const { PipelineService } = require('./pipeline-service');

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

class CompanyHeartbeatService {

  constructor() {
    this.timers = new Map();      // companyId → NodeJS.Timeout
    this.companyService = new CompanyService();
    this.port = 3000;
    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. 触发 CEO 心跳 (如果有 CEO 在公司内)
      if (company.ceo) {
        this._triggerCeoHeartbeat(companyId, company.ceo);
      }
    } catch (err) {
      console.error(`[CompanyHB] ❌ ${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: '超时跳过' },
            });
          }
        }
        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 taskMessage = [
        `[PIPELINE TASK]`,
        `流水线: ${run.pipelineName}`,
        `步骤 ${task.stepIndex + 1}/${run.steps.length}: ${task.stepName}`,
        `指令: ${stepDef?.prompt || '请执行你的职责'}`,
        inputStr,
        '',
        `完成后请调用以下接口汇报结果:`,
        `curl -X POST http://localhost:${this.port}/api/company/${companyId}/pipelines/runs/${runMeta.runId}/advance \\`,
        `  -H "Content-Type: application/json" \\`,
        `  -d '{"stepIndex": ${task.stepIndex}, "output": {"你的输出key": "你的输出value"}}'`,
      ].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)}`);
  }

  /**
   * 通过 HTTP 向 Agent 发送消息 (异步, 不阻塞)
   * 公司内 Agent 通过 /api/agent/{agentId}/chat 触发
   * TODO: 未来改为公司级路由 /api/company/{companyId}/agent/{agentId}/chat
   */
  _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 };