宜博 / lemonboss

const fs = require('fs');
const path = require('path');
const http = require('http');
const config = require('../config');

// Agent 目录寻址收编进 agent-locator（28 号设计 §2.5）。
const locator = require('./agent-locator');
const { listCompanyAgents } = require('./company-service');

/**
 * 心跳状态的归属是 `(agentId, companyId)`，所以它落**公司态目录**
 * （`system/agents/{aid}/{cid}/heartbeat.json`，与 company-context.md / user.md / memory/ 同级）。
 *
 * 改之前它写在本体的 config.json 里：A 公司把某角色调成 30 分钟，雇了同一个本体的
 * B 公司会跟着变 —— 本体只有一份，写进去就是全系统共享。而且危害不止「用户点了开关」：
 * 端点是**读的时候顺手补默认值再整份回写**，所以 28 号 E2E 全程没人碰过心跳设置，
 * `system/agents/orggraph-planning/config.json` 照样多出一行 `"intervalMinutes": 15`。
 * 这和第一轮修掉的「公司对话内容写进共享 SKILL.md」是同一类：公司级状态落在共享本体上。
 *
 * 本体的 `config.json.heartbeat` 保留，但降级成**只读缺省**——「这个角色建议多久心跳一次」
 * 是本体的合理内容，只是任何写入都不许回到那里去。
 */
const HEARTBEAT_STATE_FILE = 'heartbeat.json';

/** 没有任何配置时的兜底：默认不开，15 分钟一次 */
const HEARTBEAT_FALLBACK = { enabled: false, intervalMinutes: 15 };

/**
 * 寻址名 → 本体 id。
 *
 * 公司限定地址 `cid:role` 里冒号后面是**角色名**，它未必等于本体 id
 * （`company_agents` 的 role_name → system_agent_id 是一张映射）。改之前
 * heartbeat-service 按 role_name 推目录、company.js 的状态端点按 system_agent_id 推目录，
 * 两边算的不是同一个字符串 —— 那正是 agent-locator 顶部记过的那个教训。
 * 名册查不到（裸 agent id / 库还没建好）就把它当本体 id 用，与改前行为一致。
 */
function resolveBodyId(addr) {
  const { companyId, agentId } = locator.parseAddress(addr);
  if (!companyId) return agentId;
  try {
    const row = listCompanyAgents(companyId).find((r) => r.role_name === agentId);
    return row ? row.system_agent_id : agentId;
  } catch {
    return agentId;
  }
}

/** 本体目录（只读）——只用来读 config.json 里的缺省心跳 */
function getAgentDir(id) {
  return locator.bodyDir(resolveBodyId(id));
}

/** 这家公司的这个 agent 的心跳状态文件 —— 写侧与读侧唯一的那一份推导 */
function heartbeatStatePath(addr) {
  const { companyId } = locator.parseAddress(addr);
  return path.join(locator.companyDir(resolveBodyId(addr), companyId), HEARTBEAT_STATE_FILE);
}

/** 本体自带的缺省心跳（只读）。读不到 / 不是合法 JSON 一律退到兜底值 */
function readBodyHeartbeatDefault(addr) {
  try {
    const cfg = JSON.parse(fs.readFileSync(path.join(getAgentDir(addr), 'config.json'), 'utf8'));
    return { ...HEARTBEAT_FALLBACK, ...(cfg.heartbeat || {}) };
  } catch {
    return { ...HEARTBEAT_FALLBACK };
  }
}

class HeartbeatService {
  constructor() {
    this.timers = new Map(); // agentId → NodeJS.Timeout
    this.activeSessions = new Map(); // agentId → { convId, startedAt } — 防并发互斥
    this.port = config.port || 3000;
  }

  /**
   * 初始化并启动全局的心跳监控 (支持全球与所有公司级 Agent)
   */
  start() {
    console.log('[Heartbeat] 💓 开始初始化 Agent 心跳服务...');
    const allAgents = [];

    // 1. 扫描系统级 Agents（system/agents/*，只看深度 1）。
    //    `_` / `.` 开头是保留 id（`_base` 元 agent 模板、`_registry.json`），不是可调用 agent（28 §5.1）。
    const sysAgentsDir = locator.AGENTS_ROOT;
    if (fs.existsSync(sysAgentsDir)) {
      const globals = fs.readdirSync(sysAgentsDir, { withFileTypes: true })
        .filter(d => d.isDirectory() && !d.name.startsWith('_') && !d.name.startsWith('.'));
      for (const ag of globals) {
        if (fs.existsSync(path.join(sysAgentsDir, ag.name, 'config.json'))) allAgents.push(ag.name);
      }
    }

    // 2. 公司级 Agents：名册查 company_agents（28 批次 4-19）。
    //    原来是扫 `data/companies/*/agents/*` 的目录 —— 那些目录已经不存在了，
    //    继续扫的结果是一个空列表，也就是**所有公司 agent 的心跳静默不再挂载**。
    try {
      for (const row of listCompanyAgents()) {
        allAgents.push(`${row.company_id}:${row.role_name}`);
      }
    } catch (err) {
      // 库还没建好（首次启动 / 迁移中）不该拖垮整个服务启动，系统级心跳照常挂
      console.warn(`[Heartbeat] ⚠️ 读取公司名册失败，本轮只挂系统级心跳: ${err.message}`);
    }

    for (const agentId of allAgents) {
      this._setupAgentHeartbeat(agentId);
    }
  }

  /**
   * 这个地址对应的本体目录。**存在性检查必须用它**，不能自己再拼一次 ——
   * `cid:role` 里冒号后面是角色名，未必等于本体 id（company_agents 的映射）。
   * 端点用 `bodyDir(parseAddress(id).agentId)` 判 404 的话，凡是角色名 ≠ 本体 id 的
   * 都会被判成「agent 不存在」，而写侧其实找得到。
   */
  bodyDirOf(addr) {
    return getAgentDir(addr);
  }

  /**
   * 这个 `(agent, 公司)` 的有效心跳配置：公司态优先，缺字段回落本体缺省。
   * **读的一侧只有这一份实现** —— 端点和心跳挂载读的必须是同一个答案，
   * 否则会出现「控制台显示 30 分钟、实际按本体的 15 分钟在跑」这种不报错的错位。
   * @param {string} addr `cid:role` 或裸 agent id（后者落 `_global` 公司态）
   */
  readHeartbeatConfig(addr) {
    const base = readBodyHeartbeatDefault(addr);
    try {
      const own = JSON.parse(fs.readFileSync(heartbeatStatePath(addr), 'utf8'));
      return { ...base, ...own };
    } catch {
      return base; // 这家公司还没设置过 → 本体缺省
    }
  }

  /**
   * 改这个 `(agent, 公司)` 的心跳配置 —— **只写公司态，绝不回写本体**。
   * patch 里没给的字段保持原值（先读回有效配置再覆盖，所以第一次写会把本体缺省固化下来）。
   * @param {{enabled?: boolean, intervalMinutes?: number}} patch
   */
  writeHeartbeatConfig(addr, patch = {}) {
    const next = { ...this.readHeartbeatConfig(addr) };
    if (patch.intervalMinutes !== undefined) {
      // 间隔直接进 setInterval，NaN / 0 / 负数会让定时器退化成「每 tick 触发一次」——
      // 这是 HTTP 入参，拦在写盘之前，别等跑起来才发现。
      const n = Number(patch.intervalMinutes);
      if (!Number.isFinite(n) || n <= 0) throw new Error(`intervalMinutes 非法: ${patch.intervalMinutes}`);
      next.intervalMinutes = n;
    }
    if (patch.enabled !== undefined) next.enabled = Boolean(patch.enabled);

    const fp = heartbeatStatePath(addr);
    fs.mkdirSync(path.dirname(fp), { recursive: true });
    fs.writeFileSync(fp, JSON.stringify(next, null, 2), 'utf-8');
    return next;
  }

  /**
   * 停止所有心跳
   */
  stop() {
    for (const [agentId, timer] of this.timers.entries()) {
      clearInterval(timer);
    }
    this.timers.clear();
    console.log('[Heartbeat] 🛑 心跳服务已停止');
  }

  /**
   * 为单个 Agent 根据 config 设置心跳
   */
  _setupAgentHeartbeat(agentId) {
    try {
      // 公司态优先、本体缺省兜底 —— 与端点读的是同一个函数（见 readHeartbeatConfig）
      const hb = this.readHeartbeatConfig(agentId);

      if (hb.enabled === true) {
        const intervalMs = (hb.intervalMinutes || 15) * 60 * 1000;

        // 确保不会重复启动
        if (this.timers.has(agentId)) {
          clearInterval(this.timers.get(agentId));
        }

        const timer = setInterval(() => this._triggerHeartbeat(agentId), intervalMs);
        this.timers.set(agentId, timer);
        console.log(`[Heartbeat] ✅ 已挂载 ${agentId} 的主动心跳 (间隔: ${hb.intervalMinutes} 分钟)`);
      }
    } catch (err) {
      console.warn(`[Heartbeat] ⚠️ 挂载 ${agentId} 心跳失败: ${err.message}`);
    }
  }

  /**
   * 内部方法：向 Agent 的 Chat API 发送一次隐式的系统心跳
   */
  _triggerHeartbeat(agentId) {
    // 🔒 互斥检查：如果该 Agent 已有活跃心跳会话，跳过本次
    const SESSION_TIMEOUT = 30 * 60 * 1000; // 30 分钟安全阀
    if (this.activeSessions.has(agentId)) {
      const active = this.activeSessions.get(agentId);
      const elapsed = Date.now() - active.startedAt;
      if (elapsed < SESSION_TIMEOUT) {
        console.log(`[Heartbeat] ⏭️ 跳过 ${agentId} — 已有活跃会话 ${active.convId} (${(elapsed / 1000).toFixed(0)}s)`);
        return;
      }
      // 超时强制释放（防死锁）
      console.warn(`[Heartbeat] ⚠️ 强制释放超时会话: ${agentId} / ${active.convId} (${(elapsed / 1000).toFixed(0)}s)`);
      this.activeSessions.delete(agentId);
    }

    const timestamp = new Date().toISOString().replace('T', ' ').substring(0, 19);
    console.log(`[Heartbeat] 💓 触发 ${agentId} 主动心跳 (${timestamp})...`);

    const convId = `heartbeat-${Date.now().toString(36)}`;

    // 🔒 注册活跃会话
    this.activeSessions.set(agentId, { convId, startedAt: Date.now() });

    const heartbeatMessage = [
      `[SYSTEM HEARTBEAT]`,
      `当前系统时间：${timestamp}`,
      `这是一个后台周期的主动心跳触发。`,
      `要求：请根据你的身份和 heartbeat.md 规则，扫描当前最新的全局 OKR 目标状态和节点。`,
      `1. 如果发现某个负责人超时、卡点或产出需要仲裁，请主动使用 dispatch 工具给他们发送通知或驳回。`,
      `2. 如果一切正常，无事发生，请直接回复"[心跳正常]"并且不要使用任何工具。`,
      `无需等待人类确认，请完全自主决策。`
    ].join('\n');

    const postData = JSON.stringify({
      message: heartbeatMessage,
      conversationId: convId,
      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 releaseSession = () => {
      this.activeSessions.delete(agentId);
      console.log(`[Heartbeat] 🔓 释放 ${agentId} 会话锁 (${convId})`);
    };

    const req = http.request(options, (res) => {
      let body = '';
      res.on('data', (chunk) => body += chunk);
      res.on('end', () => {
        releaseSession();
        if (res.statusCode >= 200 && res.statusCode < 300) {
          console.log(`[Heartbeat] 📨 ${agentId} 成功完成心跳推理会话 (${convId})`);
        } else {
          console.warn(`[Heartbeat] ⚠️ ${agentId} 心跳触发异常: ${res.statusCode} ${body}`);
        }
      });
    });

    req.on('error', (err) => {
      releaseSession();
      console.error(`[Heartbeat] ❌ ${agentId} 心跳请求失败: ${err.message}`);
    });

    req.write(postData);
    req.end();
  }
}

// 导出单例
const heartbeatService = new HeartbeatService();
module.exports = heartbeatService;