diff --git a/data/templates/_registry.json b/data/templates/_registry.json index eea6e5f..6f60f45 100644 --- a/data/templates/_registry.json +++ b/data/templates/_registry.json @@ -6,7 +6,7 @@ "description": "公司战略规划、团队管理、OKR 制定、任务委派", "benchmarkScore": 97, "trainedAt": "2026-03-20", - "usedCount": 4 + "usedCount": 5 }, { "id": "qa-judge", @@ -15,6 +15,6 @@ "description": "质量验收、OKR 进度审查、标准化仲裁", "benchmarkScore": 95, "trainedAt": "2026-03-25", - "usedCount": 4 + "usedCount": 5 } ] \ No newline at end of file diff --git a/server/index.js b/server/index.js index 23739dc..e4c9e43 100644 --- a/server/index.js +++ b/server/index.js @@ -76,6 +76,9 @@ async function start() { resumeIncompleteTasks().catch(err => console.error('[Startup] 恢复任务失败:', err.message)); const heartbeatService = require('./services/heartbeat-service'); heartbeatService.start(); + // 公司级心跳 (Pipeline 自动驱动) + const { companyHeartbeatService } = require('./services/company-heartbeat-service'); + companyHeartbeatService.start(); }, 5000); } catch (error) { console.error('Failed to start server:', error); diff --git a/server/scripts/test-company-e2e.js b/server/scripts/test-company-e2e.js index f52e2ce..ae1e8c5 100644 --- a/server/scripts/test-company-e2e.js +++ b/server/scripts/test-company-e2e.js @@ -20,6 +20,7 @@ const fs = require('fs'); const { CompanyService } = require('../services/company-service'); const { GoalsService } = require('../services/goals'); const { PipelineService } = require('../services/pipeline-service'); +const { CompanyHeartbeatService } = require('../services/company-heartbeat-service'); const DATA_DIR = path.resolve(__dirname, '../../data'); const TEST_COMPANY_ID = 'test-media-co'; @@ -371,14 +372,60 @@ async function main() { const updPipeline = pSvc.updatePipeline('daily-content', { name: '每日内容产出 v2' }); assert(updPipeline.name === '每日内容产出 v2', '流水线名称更新成功'); - // ─── 15. 更新公司 ─── - console.log('\n✏️ Phase 15: 更新公司'); - const updated = svc.update(TEST_COMPANY_ID, { - status: 'running', - heartbeat: { enabled: true }, + // ─── 15. Phase 5: 心跳服务 + Pipeline 自动触发 ─── + console.log('\n💓 Phase 15: 心跳服务 + Pipeline 自动触发'); + + // 15a: 创建带 cron 的流水线 + const pSvc2 = new PipelineService(TEST_COMPANY_ID); + const cronPipeline = pSvc2.createPipeline({ + id: 'weekly-report', + name: '周报汇总', + description: '每周自动汇总运营数据', + cron: { intervalHours: 168 }, // 每168小时 = 每周 + steps: [ + { name: '数据汇总', agent: 'ceo', prompt: '汇总本周运营数据' }, + { name: '报告生成', agent: 'content-writer', prompt: '生成周报' }, + ], }); - assert(updated.status === 'running', '状态更新为 running'); - assert(updated.heartbeat.enabled === true, '心跳已开启'); + assert(cronPipeline.cron !== null, '流水线包含 cron 配置'); + assert(cronPipeline.cron.intervalHours === 168, 'cron 间隔 = 168小时'); + + // 15b: 心跳服务实例化 + const hbSvc = new CompanyHeartbeatService(); + assert(hbSvc.timers.size === 0, '初始无计时器'); + + // 15c: _shouldRunPipeline 检测 (cron 无运行记录 → 应该运行) + const shouldRun1 = hbSvc._shouldRunPipeline(pSvc2, cronPipeline); + assert(shouldRun1 === true, '无历史运行 → 应该触发'); + + // 15d: 手动启动一次 Run 后再检测 + pSvc2.startRun('weekly-report', { trigger: 'test' }); + const shouldRun2 = hbSvc._shouldRunPipeline(pSvc2, cronPipeline); + assert(shouldRun2 === false, '刚运行过 → 不应触发 (间隔未到)'); + + // 15e: 注册/取消公司心跳 + svc.update(TEST_COMPANY_ID, { status: 'running', heartbeat: { enabled: true, intervalMinutes: 5 } }); + hbSvc.registerCompany(TEST_COMPANY_ID); + assert(hbSvc.timers.has(TEST_COMPANY_ID), '公司心跳已注册'); + + hbSvc.unregisterCompany(TEST_COMPANY_ID); + assert(!hbSvc.timers.has(TEST_COMPANY_ID), '公司心跳已取消'); + + // 15f: 驱动运行中的 Run (检查 getCurrentTask) + const testRun = pSvc2.startRun('daily-content', { trigger: 'heartbeat-test' }); + const readyTask = pSvc2.getCurrentTask(testRun.runId); + assert(readyTask !== null, '新 Run 有 ready 任务'); + assert(readyTask.agent === 'hot-topic-hunter', 'Ready 任务分配给正确 Agent'); + + // 15g: 超时检测 (模拟超时场景) + pSvc2.markStepStarted(testRun.runId, 0); + // 手动篡改 startedAt 为很久以前 + const runData = pSvc2.getRun(testRun.runId); + runData.steps[0].startedAt = new Date(Date.now() - 60 * 60 * 1000).toISOString(); // 1小时前 + fs.writeFileSync(path.join(TEST_COMPANY_DIR, 'pipelines', 'runs', `${testRun.runId}.json`), JSON.stringify(runData, null, 2), 'utf-8'); + + // 清理 + hbSvc.stop(); // ─── 16. 关停公司 ─── console.log('\n🛑 Phase 16: 关停公司'); @@ -403,7 +450,7 @@ async function main() { console.log(`\n${'═'.repeat(50)}`); console.log(` 📊 测试结果: ${passed} 通过, ${failed} 失败 (共 ${passed + failed})`); if (failed === 0) { - console.log(' 🎉 ALL PASSED — Phase 1 + 2 + 3 + 4 生产就绪!\n'); + console.log(' 🎉 ALL PASSED — Phase 1 + 2 + 3 + 4 + 闭环 生产就绪!\n'); } else { console.log(` ⚠️ ${failed} 个测试未通过\n`); process.exit(1); diff --git a/server/services/company-heartbeat-service.js b/server/services/company-heartbeat-service.js new file mode 100644 index 0000000..9f82365 --- /dev/null +++ b/server/services/company-heartbeat-service.js @@ -0,0 +1,310 @@ +/** + * 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 }; diff --git a/server/services/pipeline-service.js b/server/services/pipeline-service.js index 1a154f5..061ae0d 100644 --- a/server/services/pipeline-service.js +++ b/server/services/pipeline-service.js @@ -36,7 +36,7 @@ class PipelineService { * 创建流水线定义 * @param {object} opts - { id, name, description, steps: [ { name, agent, prompt, ... } ] } */ - createPipeline({ id, name, description = '', steps = [] }) { + createPipeline({ id, name, description = '', steps = [], cron = null }) { if (!id || !name) throw new Error('id 和 name 为必填项'); const fp = path.join(this.pipelinesDir, `${id}.json`); @@ -60,6 +60,7 @@ class PipelineService { description, companyId: this.companyId, steps: numberedSteps, + cron: cron || null, // { intervalHours: 24 } — 自动触发间隔 status: 'active', createdAt: now, updatedAt: now, @@ -109,6 +110,7 @@ class PipelineService { if (updates.name !== undefined) pipeline.name = updates.name; if (updates.description !== undefined) pipeline.description = updates.description; if (updates.status !== undefined) pipeline.status = updates.status; + if (updates.cron !== undefined) pipeline.cron = updates.cron; if (updates.steps !== undefined) { pipeline.steps = updates.steps.map((step, idx) => ({ index: idx,