宜博 / lemonboss

/**
 * RetrainWatchdog — 看门狗守护 retrain pipeline
 *
 * 定期扫描活跃 job，检测超时步骤，从 checkpoint 恢复。
 * 服务启动时扫描 conversations/retrain-* 恢复未完成 job。
 *
 * Phase 2: 支持 Master Agent 智能巡检 (每 5 分钟)
 */

const fs = require('fs');
const path = require('path');

const SCAN_INTERVAL = 60_000;            // 60s 基础扫描
const INSPECTION_INTERVAL = 5 * 60_000;  // 5 分钟 Master 巡检
const MAX_STEP_DURATION = 10 * 60_000;   // 单步最大 10 分钟
const MAX_RECOVERY_ATTEMPTS = 2;

class RetrainWatchdog {
  constructor() {
    this._timer = null;
    this._inspectionTimer = null;
    this._jobManager = null;
    this._getInnovator = null;
    this._getMasterSession = null;
    this._recovering = new Set();
    this._recoveryCount = new Map();
  }

  /**
   * @param {object} jobManager
   * @param {Function} getInnovator
   * @param {Function} [getMasterSession] - 可选，提供 MasterSession 进行智能巡检
   */
  start(jobManager, getInnovator, getMasterSession = null) {
    this._jobManager = jobManager;
    this._getInnovator = getInnovator;
    this._getMasterSession = getMasterSession;

    const mode = getMasterSession ? 'Master 智能巡检' : '规则引擎';
    console.log(`[Watchdog] 🐕 看门狗已启动 (${mode}, 每 60s 扫描)`);

    this._recoverOnStartup();

    this._timer = setInterval(() => this._scan(), SCAN_INTERVAL);
    this._timer.unref();

    if (getMasterSession) {
      this._inspectionTimer = setInterval(() => this._masterInspect(), INSPECTION_INTERVAL);
      this._inspectionTimer.unref();
    }
  }

  stop() {
    if (this._timer) { clearInterval(this._timer); this._timer = null; }
    if (this._inspectionTimer) { clearInterval(this._inspectionTimer); this._inspectionTimer = null; }
    console.log('[Watchdog] 看门狗已停止');
  }

  // ── Master 智能巡检 (每 5 分钟) ──

  async _masterInspect() {
    if (!this._getMasterSession) return;

    const activeStatuses = ['hunting', 'digesting', 'evolving', 'master-orchestrating'];
    const allJobs = this._jobManager.listJobs({ limit: 50 });
    const activeJobs = allJobs.filter(j => activeStatuses.includes(j.status));

    if (activeJobs.length === 0) return;

    try {
      const masterSession = this._getMasterSession();
      if (!masterSession) return;

      const jobSummary = activeJobs.map(j =>
        `- Job ${j.jobId}: skill=${j.skillName}, status=${j.status}, ` +
        `elapsed=${Math.round((Date.now() - new Date(j.updatedAt).getTime()) / 1000)}s`
      ).join('\n');

      const inspectPrompt = [
        `# 定期巡检 — Retrain 任务状态检查`,
        ``,
        `当前时间: ${new Date().toISOString()}`,
        `活跃任务数: ${activeJobs.length}`,
        ``,
        `## 活跃任务列表`,
        jobSummary,
        ``,
        `## 请执行以下检查`,
        `1. 使用 \`read_checkpoint\` 读取每个任务的 checkpoint`,
        `2. 如果任务超过 10 分钟无更新，使用 \`read_log\` 查看日志`,
        `3. 判断：任务是否正常？是否卡住？是否需要干预？`,
        `4. 输出简要巡检报告`,
      ].join('\n');

      console.log(`[Watchdog] 🧠 发送 Master 巡检 (${activeJobs.length} 个活跃任务)`);

      const result = await masterSession.handleSystemMessage(inspectPrompt, {
        maxToolRounds: 5,
      });

      if (result) {
        console.log(`[Watchdog] 🧠 巡检完成 (${result.rounds || 0} 轮, ${(result.toolCalls || []).length} 次工具调用)`);
        const logDir = path.resolve(__dirname, '../../conversations/.watchdog');
        if (!fs.existsSync(logDir)) fs.mkdirSync(logDir, { recursive: true });
        fs.writeFileSync(path.join(logDir, `inspect-${Date.now()}.json`), JSON.stringify({
          timestamp: new Date().toISOString(),
          activeJobs: activeJobs.length,
          output: result.output?.substring(0, 2000),
          rounds: result.rounds,
          toolCalls: (result.toolCalls || []).map(tc => ({ tool: tc.tool, params: tc.params })),
        }, null, 2));
      }
    } catch (err) {
      console.error('[Watchdog] 🧠❌ Master 巡检失败:', err.message);
    }
  }

  // ── 基础扫描 + 规则恢复 ──

  async _recoverOnStartup() {
    const conversationsDir = path.resolve(__dirname, '../../conversations');
    if (!fs.existsSync(conversationsDir)) return;

    const dirs = fs.readdirSync(conversationsDir)
      .filter(d => d.startsWith('retrain-') && fs.statSync(path.join(conversationsDir, d)).isDirectory());

    let recovered = 0;
    for (const dir of dirs) {
      const checkpointPath = path.join(conversationsDir, dir, 'checkpoint.json');
      if (!fs.existsSync(checkpointPath)) continue;

      try {
        const checkpoint = JSON.parse(fs.readFileSync(checkpointPath, 'utf-8'));
        if (checkpoint.step === 'publish' && checkpoint.status === 'done') continue;
        if (checkpoint.status === 'done' || checkpoint.status === 'failed') {
          const existingJob = this._jobManager.getJob(checkpoint.jobId);
          if (existingJob && (existingJob.status === 'completed' || existingJob.status === 'failed')) continue;

          console.log(`[Watchdog] ♻️ 发现未完成 job: ${checkpoint.jobId} (step=${checkpoint.step}, status=${checkpoint.status})`);
          await this._recoverJob(checkpoint);
          recovered++;
        }
      } catch (err) {
        console.warn(`[Watchdog] 读取 checkpoint 失败: ${dir}`, err.message);
      }
    }
    if (recovered > 0) console.log(`[Watchdog] 启动恢复: ${recovered} 个 job`);
  }

  _scan() {
    if (!this._jobManager) return;

    const activeStatuses = ['hunting', 'digesting', 'evolving'];
    const allJobs = this._jobManager.listJobs({ limit: 50 });

    for (const job of allJobs) {
      if (!activeStatuses.includes(job.status)) continue;
      if (this._recovering.has(job.jobId)) continue;

      const elapsed = Date.now() - new Date(job.updatedAt).getTime();
      if (elapsed > MAX_STEP_DURATION) {
        console.log(`[Watchdog] ⏰ Job ${job.jobId} 超时 (${Math.round(elapsed / 1000)}s > ${MAX_STEP_DURATION / 1000}s)`);

        const conversationsDir = path.resolve(__dirname, '../../conversations');
        const checkpointPath = path.join(conversationsDir, `retrain-${job.jobId}`, 'checkpoint.json');

        if (fs.existsSync(checkpointPath)) {
          try {
            const checkpoint = JSON.parse(fs.readFileSync(checkpointPath, 'utf-8'));
            this._recoverJob(checkpoint).catch(err => {
              console.error(`[Watchdog] 恢复失败: ${job.jobId}`, err.message);
            });
          } catch (err) {
            this._markFailed(job.jobId, '无法读取 checkpoint');
          }
        } else {
          this._markFailed(job.jobId, '超时且无 checkpoint 可恢复');
        }
      }
    }
  }

  async _recoverJob(checkpoint) {
    const { jobId, skillName } = checkpoint;

    const count = this._recoveryCount.get(jobId) || 0;
    if (count >= MAX_RECOVERY_ATTEMPTS) {
      this._markFailed(jobId, `超过最大恢复次数 (${MAX_RECOVERY_ATTEMPTS})`);
      return;
    }

    this._recovering.add(jobId);
    this._recoveryCount.set(jobId, count + 1);

    try {
      const innovator = this._getInnovator();
      if (!innovator) throw new Error('InnovatorService 未就绪');

      const skill = innovator.registry.getSkill(skillName);
      if (!skill) throw new Error(`Skill ${skillName} 不存在`);

      let job = this._jobManager.getJob(jobId);
      if (!job) {
        job = this._jobManager.createJob({
          type: 'retrain', skillName,
          status: 'recovering', progress: '从 checkpoint 恢复中...',
        });
        job.jobId = jobId;
        this._jobManager.jobs.set(jobId, job);
      } else {
        this._jobManager.updateJob(jobId, {
          status: 'recovering',
          progress: `从 checkpoint 恢复中 (第 ${count + 1} 次)...`,
        });
      }

      innovator.registry.updateSkill(skillName, { status: 'training' });
      console.log(`[Watchdog] 🔧 开始恢复 job ${jobId} (第 ${count + 1} 次)`);

      const { RetrainPipeline } = require('./retrain-pipeline');
      const pipeline = new RetrainPipeline({
        job, skill, innovator, jobManager: this._jobManager,
      });

      // 优先 Master 编排模式
      const masterSession = this._getMasterSession?.();
      if (masterSession) {
        await pipeline.runWithMaster(masterSession);
      } else {
        await pipeline.run();
      }
    } catch (err) {
      console.error(`[Watchdog] ❌ 恢复 job ${jobId} 失败:`, err.message);
      this._markFailed(jobId, `恢复失败: ${err.message}`);
      try {
        const innovator = this._getInnovator();
        if (innovator && skillName) innovator.registry.updateSkill(skillName, { status: 'enable' });
      } catch {}
    } finally {
      this._recovering.delete(jobId);
    }
  }

  _markFailed(jobId, reason) {
    this._jobManager.updateJob(jobId, { status: 'failed', progress: `失败: ${reason}` });
  }
}

const watchdog = new RetrainWatchdog();
module.exports = { RetrainWatchdog, watchdog };