From ec77ccb9ea7d36eb4c13eb9931fb6979180c0b14 Mon Sep 17 00:00:00 2001 From: yibo Date: Wed, 25 Mar 2026 17:59:11 +0800 Subject: [PATCH] feat(company): implement Phase 4 - SOP Pipeline execution engine --- data/templates/_registry.json | 4 ++-- server/routes/company.js | 85 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ server/scripts/test-company-e2e.js | 89 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----- server/services/pipeline-service.js | 314 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 485 insertions(+), 7 deletions(-) create mode 100644 server/services/pipeline-service.js diff --git a/data/templates/_registry.json b/data/templates/_registry.json index 6fb0694..eea6e5f 100644 --- a/data/templates/_registry.json +++ b/data/templates/_registry.json @@ -6,7 +6,7 @@ "description": "公司战略规划、团队管理、OKR 制定、任务委派", "benchmarkScore": 97, "trainedAt": "2026-03-20", - "usedCount": 3 + "usedCount": 4 }, { "id": "qa-judge", @@ -15,6 +15,6 @@ "description": "质量验收、OKR 进度审查、标准化仲裁", "benchmarkScore": 95, "trainedAt": "2026-03-25", - "usedCount": 3 + "usedCount": 4 } ] \ No newline at end of file diff --git a/server/routes/company.js b/server/routes/company.js index 6ad8412..32ca9e6 100644 --- a/server/routes/company.js +++ b/server/routes/company.js @@ -192,4 +192,89 @@ router.get('/:companyId/goals/:goalId/progress', (req, res) => { res.json(progress); }); +// ═══════════════════════════════════════ +// Pipeline / SOP 流水线 +// ═══════════════════════════════════════ + +const { PipelineService } = require('../services/pipeline-service'); + +// POST /api/company/:companyId/pipelines — 创建流水线 +router.post('/:companyId/pipelines', (req, res) => { + try { + const pSvc = new PipelineService(req.params.companyId); + const pipeline = pSvc.createPipeline(req.body); + res.status(201).json(pipeline); + } catch (err) { + res.status(400).json({ error: err.message }); + } +}); + +// GET /api/company/:companyId/pipelines — 列出流水线 +router.get('/:companyId/pipelines', (req, res) => { + const pSvc = new PipelineService(req.params.companyId); + res.json(pSvc.listPipelines()); +}); + +// GET /api/company/:companyId/pipelines/:pipelineId — 获取流水线定义 +router.get('/:companyId/pipelines/:pipelineId', (req, res) => { + const pSvc = new PipelineService(req.params.companyId); + const pipeline = pSvc.getPipeline(req.params.pipelineId); + if (!pipeline) return res.status(404).json({ error: '流水线不存在' }); + res.json(pipeline); +}); + +// PUT /api/company/:companyId/pipelines/:pipelineId — 更新流水线 +router.put('/:companyId/pipelines/:pipelineId', (req, res) => { + const pSvc = new PipelineService(req.params.companyId); + const pipeline = pSvc.updatePipeline(req.params.pipelineId, req.body); + if (!pipeline) return res.status(404).json({ error: '流水线不存在' }); + res.json(pipeline); +}); + +// POST /api/company/:companyId/pipelines/:pipelineId/run — 启动运行 +router.post('/:companyId/pipelines/:pipelineId/run', (req, res) => { + try { + const pSvc = new PipelineService(req.params.companyId); + const run = pSvc.startRun(req.params.pipelineId, req.body); + res.status(201).json(run); + } catch (err) { + res.status(400).json({ error: err.message }); + } +}); + +// GET /api/company/:companyId/pipelines/runs/list — 列出运行记录 +router.get('/:companyId/pipelines/runs/list', (req, res) => { + const pSvc = new PipelineService(req.params.companyId); + res.json(pSvc.listRuns(req.query.pipelineId)); +}); + +// GET /api/company/:companyId/pipelines/runs/:runId — 获取运行详情 +router.get('/:companyId/pipelines/runs/:runId', (req, res) => { + const pSvc = new PipelineService(req.params.companyId); + const run = pSvc.getRun(req.params.runId); + if (!run) return res.status(404).json({ error: '运行记录不存在' }); + res.json(run); +}); + +// POST /api/company/:companyId/pipelines/runs/:runId/advance — 推进步骤 +router.post('/:companyId/pipelines/runs/:runId/advance', (req, res) => { + try { + const pSvc = new PipelineService(req.params.companyId); + const { stepIndex, output, status } = req.body; + if (stepIndex === undefined) return res.status(400).json({ error: 'stepIndex 为必填项' }); + const run = pSvc.advanceStep(req.params.runId, Number(stepIndex), { output, status }); + res.json(run); + } catch (err) { + res.status(400).json({ error: err.message }); + } +}); + +// GET /api/company/:companyId/pipelines/runs/:runId/current — 获取当前任务 +router.get('/:companyId/pipelines/runs/:runId/current', (req, res) => { + const pSvc = new PipelineService(req.params.companyId); + const task = pSvc.getCurrentTask(req.params.runId); + if (!task) return res.json({ message: '无就绪任务' }); + res.json(task); +}); + module.exports = router; diff --git a/server/scripts/test-company-e2e.js b/server/scripts/test-company-e2e.js index db9928e..f52e2ce 100644 --- a/server/scripts/test-company-e2e.js +++ b/server/scripts/test-company-e2e.js @@ -19,6 +19,7 @@ const fs = require('fs'); // 直接引入 service 层 (不需要 HTTP) const { CompanyService } = require('../services/company-service'); const { GoalsService } = require('../services/goals'); +const { PipelineService } = require('../services/pipeline-service'); const DATA_DIR = path.resolve(__dirname, '../../data'); const TEST_COMPANY_ID = 'test-media-co'; @@ -292,8 +293,86 @@ async function main() { const hasCompanyGoal = globalList.some(g => g.id === 'G-001' && g.title === '30天粉丝破万'); assert(!hasCompanyGoal, '全局 GoalsService 看不到公司级 OKR (隔离)'); - // ─── 14. 更新公司 ─── - console.log('\n✏️ Phase 14: 更新公司'); + // ─── 14. Phase 4: Pipeline SOP 流水线 ─── + console.log('\n🛠️ Phase 14: Pipeline SOP 流水线'); + + const pSvc = new PipelineService(TEST_COMPANY_ID); + + // 14a: 创建流水线 + const pipeline = pSvc.createPipeline({ + id: 'daily-content', + name: '每日内容产出', + description: '选题 → 写稿 → 审核 → 发布', + steps: [ + { name: '热点搜索与选题', agent: 'hot-topic-hunter', prompt: '搜索今日热点并提供3个选题建议' }, + { name: '内容写作', agent: 'content-writer', prompt: '根据选题写一篇3000字文章' }, + { name: '质量审核', agent: 'qa-judge', prompt: '审核文章质量，给出修改意见' }, + { name: '发布推送', agent: 'publisher', prompt: '排版并发布到公众号' }, + ], + }); + assert(pipeline.id === 'daily-content', '流水线 ID 正确'); + assert(pipeline.steps.length === 4, `步骤数 = 4 (实际: ${pipeline.steps.length})`); + assert(pipeline.steps[0].agent === 'hot-topic-hunter', '第1步 agent 正确'); + assert(pipeline.steps[1].dependsOn === 0, '第2步依赖第1步'); + + // 14b: 列出流水线 + const pipelines = pSvc.listPipelines(); + assert(pipelines.length === 1, `流水线数量 = 1 (实际: ${pipelines.length})`); + + // 14c: 获取流水线定义 + const fetPipeline = pSvc.getPipeline('daily-content'); + assert(fetPipeline !== null, '获取流水线定义成功'); + assert(fetPipeline.name === '每日内容产出', '流水线名称正确'); + + // 14d: 启动运行 + const run = pSvc.startRun('daily-content', { + trigger: 'heartbeat', + input: { date: '2026-03-25' }, + }); + assert(run.status === 'running', 'Run 状态为 running'); + assert(run.steps.length === 4, `Run 步骤数 = 4`); + assert(run.steps[0].status === 'ready', '第1步状态为 ready'); + assert(run.steps[1].status === 'waiting', '第2步状态为 waiting'); + assert(run.currentStep === 0, '当前步骤 = 0'); + + // 14e: 获取当前任务 + const curTask = pSvc.getCurrentTask(run.runId); + assert(curTask !== null, '当前任务不为空'); + assert(curTask.agent === 'hot-topic-hunter', '当前任务分配给 hot-topic-hunter'); + assert(curTask.stepIndex === 0, '当前任务是第0步'); + + // 14f: 标记步骤开始 + pSvc.markStepStarted(run.runId, 0); + const runAfterStart = pSvc.getRun(run.runId); + assert(runAfterStart.steps[0].status === 'in_progress', '第1步状态变为 in_progress'); + + // 14g: 推进步骤 0 → 1 + const run1 = pSvc.advanceStep(run.runId, 0, { + output: { topics: ['AI提效工具', '副业经济', '新能源资讯'] }, + }); + assert(run1.steps[0].status === 'done', '第1步完成'); + assert(run1.steps[1].status === 'ready', '第2步变为 ready'); + assert(run1.currentStep === 1, '当前步骤推进到 1'); + assert(run1.steps[1].input !== null, '第2步的 input 为第1步的 output'); + + // 14h: 推进到最后一步 + pSvc.advanceStep(run.runId, 1, { output: { draft: '文章初稿...' } }); + pSvc.advanceStep(run.runId, 2, { output: { approved: true, score: 92 } }); + const runFinal = pSvc.advanceStep(run.runId, 3, { output: { publishUrl: 'https://mp.weixin.qq.com/s/xxx' } }); + assert(runFinal.status === 'completed', 'Run 全部完成'); + assert(runFinal.output !== null, 'Run 携带最终 output'); + + // 14i: 列出运行记录 + const runs = pSvc.listRuns(); + assert(runs.length === 1, `运行记录数 = 1 (实际: ${runs.length})`); + assert(runs[0].status === 'completed', '运行状态为 completed'); + + // 14j: 更新流水线 + 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 }, @@ -301,8 +380,8 @@ async function main() { assert(updated.status === 'running', '状态更新为 running'); assert(updated.heartbeat.enabled === true, '心跳已开启'); - // ─── 15. 关停公司 ─── - console.log('\n🛑 Phase 15: 关停公司'); + // ─── 16. 关停公司 ─── + console.log('\n🛑 Phase 16: 关停公司'); const shutdown = svc.shutdown(TEST_COMPANY_ID); assert(shutdown.status === 'shutdown', '状态变为 shutdown'); assert(shutdown.heartbeat.enabled === false, '心跳已关闭'); @@ -324,7 +403,7 @@ async function main() { console.log(`\n${'═'.repeat(50)}`); console.log(` 📊 测试结果: ${passed} 通过, ${failed} 失败 (共 ${passed + failed})`); if (failed === 0) { - console.log(' 🎉 ALL PASSED — Phase 1 + Phase 2 + Phase 3 生产就绪!\n'); + console.log(' 🎉 ALL PASSED — Phase 1 + 2 + 3 + 4 生产就绪!\n'); } else { console.log(` ⚠️ ${failed} 个测试未通过\n`); process.exit(1); diff --git a/server/services/pipeline-service.js b/server/services/pipeline-service.js new file mode 100644 index 0000000..1a154f5 --- /dev/null +++ b/server/services/pipeline-service.js @@ -0,0 +1,314 @@ +/** + * PipelineService — SOP 流水线管理与执行 + * + * 数据目录: data/{companyId}/pipelines/ + * 流水线定义: data/{companyId}/pipelines/{pipelineId}.json + * 运行记录: data/{companyId}/pipelines/runs/{runId}.json + * + * 核心概念: + * Pipeline = 一条 SOP 流水线定义 (选题 → 写稿 → 审核 → 发布) + * Step = Pipeline 中的一个步骤 (分配给一个 Agent) + * Run = Pipeline 的一次实际执行 (含每步的输入/输出/状态) + */ + +const fs = require('fs'); +const path = require('path'); + +const DATA_DIR = path.resolve(__dirname, '../../data'); + +class PipelineService { + + constructor(companyId) { + if (!companyId) throw new Error('PipelineService 必须指定 companyId'); + this.companyId = companyId; + this.pipelinesDir = path.join(DATA_DIR, companyId, 'pipelines'); + this.runsDir = path.join(this.pipelinesDir, 'runs'); + + if (!fs.existsSync(this.pipelinesDir)) fs.mkdirSync(this.pipelinesDir, { recursive: true }); + if (!fs.existsSync(this.runsDir)) fs.mkdirSync(this.runsDir, { recursive: true }); + } + + // ═══════════════════════════════════════ + // Pipeline 定义 CRUD + // ═══════════════════════════════════════ + + /** + * 创建流水线定义 + * @param {object} opts - { id, name, description, steps: [ { name, agent, prompt, ... } ] } + */ + createPipeline({ id, name, description = '', steps = [] }) { + if (!id || !name) throw new Error('id 和 name 为必填项'); + + const fp = path.join(this.pipelinesDir, `${id}.json`); + if (fs.existsSync(fp)) throw new Error(`流水线 ${id} 已存在`); + + // 给每个 step 分配序号 + const numberedSteps = steps.map((step, idx) => ({ + index: idx, + name: step.name || `步骤${idx + 1}`, + agent: step.agent, // 负责的 Agent ID + prompt: step.prompt || '', // 发送给 Agent 的指令模板 + dependsOn: step.dependsOn ?? (idx > 0 ? idx - 1 : null), // 前置步骤 + timeout: step.timeout || 30, // 超时(分钟) + required: step.required !== false, // 是否必须 + })); + + const now = new Date().toISOString(); + const pipeline = { + id, + name, + description, + companyId: this.companyId, + steps: numberedSteps, + status: 'active', + createdAt: now, + updatedAt: now, + }; + + fs.writeFileSync(fp, JSON.stringify(pipeline, null, 2), 'utf-8'); + console.log(`[Pipeline] 📋 创建流水线: ${id} (${name}, ${steps.length} 步)`); + return pipeline; + } + + /** + * 获取流水线定义 + */ + getPipeline(pipelineId) { + const fp = path.join(this.pipelinesDir, `${pipelineId}.json`); + if (!fs.existsSync(fp)) return null; + return JSON.parse(fs.readFileSync(fp, 'utf-8')); + } + + /** + * 列出所有流水线 + */ + listPipelines() { + return fs.readdirSync(this.pipelinesDir) + .filter(f => f.endsWith('.json') && f !== 'runs') + .map(f => { + try { + const data = JSON.parse(fs.readFileSync(path.join(this.pipelinesDir, f), 'utf-8')); + return { + id: data.id, + name: data.name, + stepCount: data.steps?.length || 0, + status: data.status, + }; + } catch { return null; } + }) + .filter(Boolean); + } + + /** + * 更新流水线定义 + */ + updatePipeline(pipelineId, updates) { + const pipeline = this.getPipeline(pipelineId); + if (!pipeline) return null; + + 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.steps !== undefined) { + pipeline.steps = updates.steps.map((step, idx) => ({ + index: idx, + name: step.name || `步骤${idx + 1}`, + agent: step.agent, + prompt: step.prompt || '', + dependsOn: step.dependsOn ?? (idx > 0 ? idx - 1 : null), + timeout: step.timeout || 30, + required: step.required !== false, + })); + } + pipeline.updatedAt = new Date().toISOString(); + + fs.writeFileSync( + path.join(this.pipelinesDir, `${pipelineId}.json`), + JSON.stringify(pipeline, null, 2), 'utf-8' + ); + return pipeline; + } + + // ═══════════════════════════════════════ + // Pipeline 运行 (Run) + // ═══════════════════════════════════════ + + /** + * 启动一次流水线运行 + * @param {string} pipelineId + * @param {object} opts - { trigger, input } + * @returns {object} run + */ + startRun(pipelineId, { trigger = 'manual', input = {} } = {}) { + const pipeline = this.getPipeline(pipelineId); + if (!pipeline) throw new Error(`流水线 ${pipelineId} 不存在`); + if (pipeline.status !== 'active') throw new Error(`流水线 ${pipelineId} 未激活`); + + const runId = `R-${Date.now().toString(36)}`; + const now = new Date().toISOString(); + + const stepStates = pipeline.steps.map(step => ({ + index: step.index, + name: step.name, + agent: step.agent, + status: step.index === 0 ? 'ready' : 'waiting', // 第一步 ready，其余 waiting + input: step.index === 0 ? input : null, + output: null, + startedAt: null, + completedAt: null, + })); + + const run = { + runId, + pipelineId, + pipelineName: pipeline.name, + companyId: this.companyId, + trigger, + status: 'running', // running → completed / failed / cancelled + currentStep: 0, + steps: stepStates, + input, + output: null, + createdAt: now, + updatedAt: now, + }; + + this._saveRun(run); + console.log(`[Pipeline] 🚀 启动运行: ${runId} (${pipeline.name})`); + return run; + } + + /** + * 推进步骤 — 标记当前步骤完成，激活下一步 + * @param {string} runId + * @param {number} stepIndex + * @param {object} opts - { output, status? } + */ + advanceStep(runId, stepIndex, { output = null, status = 'done' } = {}) { + const run = this.getRun(runId); + if (!run) throw new Error(`Run ${runId} 不存在`); + + const step = run.steps[stepIndex]; + if (!step) throw new Error(`Step ${stepIndex} 不存在`); + + const now = new Date().toISOString(); + + // 更新当前步骤 + step.status = status; // done / skipped / failed + step.output = output; + step.completedAt = now; + + if (status === 'failed') { + run.status = 'failed'; + run.updatedAt = now; + this._saveRun(run); + console.log(`[Pipeline] ❌ 运行 ${runId} 步骤 ${stepIndex} 失败`); + return run; + } + + // 查找下一个步骤 + const nextIndex = stepIndex + 1; + if (nextIndex < run.steps.length) { + // 激活下一步 + run.steps[nextIndex].status = 'ready'; + run.steps[nextIndex].input = output; // 前一步的输出是下一步的输入 + run.currentStep = nextIndex; + console.log(`[Pipeline] ➡️ 运行 ${runId} 进入步骤 ${nextIndex}: ${run.steps[nextIndex].name}`); + } else { + // 所有步骤完成 + run.status = 'completed'; + run.output = output; + console.log(`[Pipeline] ✅ 运行 ${runId} 全部完成`); + } + + run.updatedAt = now; + this._saveRun(run); + return run; + } + + /** + * 标记步骤开始执行 + */ + markStepStarted(runId, stepIndex) { + const run = this.getRun(runId); + if (!run) return null; + + if (run.steps[stepIndex]) { + run.steps[stepIndex].status = 'in_progress'; + run.steps[stepIndex].startedAt = new Date().toISOString(); + run.updatedAt = new Date().toISOString(); + this._saveRun(run); + } + return run; + } + + /** + * 获取运行状态 + */ + getRun(runId) { + const fp = path.join(this.runsDir, `${runId}.json`); + if (!fs.existsSync(fp)) return null; + return JSON.parse(fs.readFileSync(fp, 'utf-8')); + } + + /** + * 列出所有运行记录 + */ + listRuns(pipelineId = null) { + if (!fs.existsSync(this.runsDir)) return []; + + return fs.readdirSync(this.runsDir) + .filter(f => f.endsWith('.json')) + .map(f => { + try { + const data = JSON.parse(fs.readFileSync(path.join(this.runsDir, f), 'utf-8')); + if (pipelineId && data.pipelineId !== pipelineId) return null; + return { + runId: data.runId, + pipelineId: data.pipelineId, + pipelineName: data.pipelineName, + status: data.status, + currentStep: data.currentStep, + totalSteps: data.steps.length, + trigger: data.trigger, + createdAt: data.createdAt, + }; + } catch { return null; } + }) + .filter(Boolean) + .sort((a, b) => b.createdAt.localeCompare(a.createdAt)); + } + + /** + * 获取当前待执行的步骤 (ready 状态) + */ + getCurrentTask(runId) { + const run = this.getRun(runId); + if (!run || run.status !== 'running') return null; + + const readyStep = run.steps.find(s => s.status === 'ready'); + if (!readyStep) return null; + + return { + runId, + pipelineId: run.pipelineId, + stepIndex: readyStep.index, + stepName: readyStep.name, + agent: readyStep.agent, + input: readyStep.input, + }; + } + + // ═══════════════════════════════════════ + // 私有方法 + // ═══════════════════════════════════════ + + _saveRun(run) { + fs.writeFileSync( + path.join(this.runsDir, `${run.runId}.json`), + JSON.stringify(run, null, 2), 'utf-8' + ); + } +} + +module.exports = { PipelineService }; -- libgit2 0.21.4