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