feat(company): implement Phase 4 - SOP Pipeline execution engine

- PipelineService: define pipelines with ordered agent-assigned steps - Run lifecycle: start → markStepStarted → advanceStep → complete - Step output automatically flows as next step input - Pipeline CRUD + Run management REST API routes - E2E test: 106/106 passed (Phase 1 + 2 + 3 + 4)

feat(company): implement Phase 4 - SOP Pipeline execution engine
- PipelineService: define pipelines with ordered agent-assigned steps - Run lifecycle: start → markStepStarted → advanceStep → complete - Step output automatically flows as next step input - Pipeline CRUD + Run management REST API routes - E2E test: 106/106 passed (Phase 1 + 2 + 3 + 4)
yibo
1 parent 34584a52
Showing 4 changed files with 485 additions and 7 deletions
data/templates/_registry.json
server/routes/company.js
server/scripts/test-company-e2e.js
server/services/pipeline-service.js
@@ -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
@@ -192,4 +192,89 @@ router.get(&#39;/:companyId/goals/:goalId/progress&#39;, (req, res) =&gt; {
   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;
@@ -19,6 +19,7 @@ const fs = require(&#39;fs&#39;);
 // 直接引入 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);
+/**
+ * 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 };