宜博 / lemonboss

/**
 * 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, 'companies', 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 = [], cron = null }) {
    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,
      cron: cron || null,           // { intervalHours: 24 } — 自动触发间隔
      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.cron !== undefined) pipeline.cron = updates.cron;
    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;
      run.completedAt = now;
      console.log(`[Pipeline] ✅ 运行 ${runId} 全部完成`);

      // ── 循环执行: 如果 Pipeline 设置了 loop: true，自动启动下一轮 ──
      const pipeline = this.getPipeline(run.pipelineId);
      if (pipeline && pipeline.loop) {
        const loopCount = (run.loopCount || 0) + 1;
        const maxLoops = pipeline.maxLoops || 365; // 默认最多循环 365 次
        if (loopCount < maxLoops) {
          console.log(`[Pipeline] 🔄 Pipeline ${run.pipelineId} 循环运行第 ${loopCount + 1} 轮`);
          const newRun = this.startRun(run.pipelineId, {
            trigger: 'loop',
            input: { date: new Date().toISOString().slice(0, 10), previousRunId: runId, loopCount },
          });
          newRun.loopCount = loopCount;
          this._saveRun(newRun);
        } else {
          console.log(`[Pipeline] 🛑 Pipeline ${run.pipelineId} 已达到最大循环次数 ${maxLoops}`);
        }
      }
    }

    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,
    };
  }

  // ═══════════════════════════════════════
  //  Pipeline↔Task 桥接 — 通过 agentId 查找活跃步骤
  // ═══════════════════════════════════════

  /**
   * 根据 agentId 查找正在执行的 Pipeline Run 步骤
   * 用于 task/report 回调时自动推进 Pipeline
   *
   * @param {string} agentId - 执行 Agent 的 ID
   * @returns {{ runId, stepIndex, run }|null}
   */
  findActiveRunStep(agentId) {
    if (!fs.existsSync(this.runsDir)) return null;

    const runFiles = fs.readdirSync(this.runsDir).filter(f => f.endsWith('.json'));
    for (const f of runFiles) {
      try {
        const run = JSON.parse(fs.readFileSync(path.join(this.runsDir, f), 'utf-8'));
        if (run.status !== 'running') continue;

        for (const step of run.steps) {
          if (step.agent === agentId && step.status === 'in_progress') {
            return { runId: run.runId, stepIndex: step.index, run };
          }
        }
      } catch { /* skip corrupted files */ }
    }
    return null;
  }

  // ═══════════════════════════════════════
  //  COO 质量门控 — 驳回步骤
  // ═══════════════════════════════════════

  /**
   * 驳回步骤 — COO 审核不通过, 重置步骤为 ready 状态让 Agent 重做
   *
   * @param {string} runId
   * @param {number} stepIndex
   * @param {{ reason?: string, reviewedBy?: string }} opts
   * @returns {object} run
   */
  rejectStep(runId, stepIndex, { reason = '', reviewedBy = 'coo' } = {}) {
    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();

    // 记录驳回历史
    if (!step.rejections) step.rejections = [];
    step.rejections.push({
      reason,
      reviewedBy,
      rejectedOutput: step.output,
      rejectedAt: now,
    });

    // 重置步骤为 ready 状态
    step.status = 'ready';
    step.output = null;
    step.completedAt = null;
    step.startedAt = null;
    // 将驳回理由注入到 input 中，让 Agent 知道需要改进什么
    step.input = {
      ...(step.input || {}),
      __rejection: {
        reason,
        reviewedBy,
        attempt: step.rejections.length,
        instruction: `⚠️ 你的上次产出被 ${reviewedBy} 驳回。原因: ${reason}。请重新执行并改进。`,
      },
    };

    run.updatedAt = now;
    this._saveRun(run);

    console.log(`[Pipeline] 🔙 运行 ${runId} 步骤 ${stepIndex} 被 ${reviewedBy} 驳回: ${reason}`);
    return run;
  }

  // ═══════════════════════════════════════
  //  私有方法
  // ═══════════════════════════════════════

  _saveRun(run) {
    fs.writeFileSync(
      path.join(this.runsDir, `${run.runId}.json`),
      JSON.stringify(run, null, 2), 'utf-8'
    );
  }
}

module.exports = { PipelineService };