宜博 / lemonboss

/**
 * DevLoop · integrator-tools (6 个)
 *   port_picker       分配空闲 localhost 端口(避开 reserved)
 *   spawn_units       批量启动 be Unit 进程,每个独立端口
 *   health_wait       poll /api/health 直到 200 或超时
 *   run_case          执行 integration spec 单个 case(含 {{stepN.x}} 占位符)
 *   parse_placeholder 解析 {{stepN.field}} 形式占位符
 *   collect_stderr    从子进程取最近 stderr 行
 *
 * 规格: docs/prd/newfunction/_spec/agents-integrator.md · §7
 */

const net = require('net');
const fs = require('fs');
const path = require('path');
const { spawn } = require('child_process');

const pathResolver = require('../../../services/devloop/path-resolver');
const { ToolValidationError, ToolMissingDependencyError } = require('./_errors');

const RESERVED_PORTS = new Set([3000, 3010, 5173, 5432, 6379, 8080, 8443]);

// ──────────────────── port_picker ────────────────────

function isPortFree(port) {
  return new Promise(resolve => {
    const srv = net.createServer();
    srv.once('error', () => resolve(false));
    srv.once('listening', () => srv.close(() => resolve(true)));
    srv.listen(port, '127.0.0.1');
  });
}

const port_picker = {
  name: 'port_picker',
  description: '分配 N 个空闲 localhost 端口(18000-19000),避开系统保留',
  inputSchema: {
    type: 'object',
    required: ['count'],
    properties: {
      count: { type: 'integer', minimum: 1, maximum: 20 },
    },
  },
  async execute({ count }, _ctx) {
    const picked = [];
    let attempts = 0;
    while (picked.length < count && attempts < 100) {
      attempts++;
      const p = 18000 + Math.floor(Math.random() * 1000);
      if (RESERVED_PORTS.has(p) || picked.includes(p)) continue;
      if (await isPortFree(p)) picked.push(p);
    }
    if (picked.length < count) {
      throw new Error(`no free port after ${attempts} attempts`);
    }
    return { ok: true, ports: picked };
  },
};

// ──────────────────── spawn_units ────────────────────

const _activeHandles = new Map(); // pid → ProcessHandle[]

const spawn_units = {
  name: 'spawn_units',
  description: '批量启动 be Unit 真实进程(每个独立端口 + env 注入)',
  inputSchema: {
    type: 'object',
    required: ['be_units', 'ports', 'log_path'],
    properties: {
      be_units: { type: 'array', items: { type: 'string' } },
      ports: { type: 'object' },
      log_path: { type: 'string' },
    },
  },
  async execute({ be_units, ports, log_path }, ctx) {
    _req(ctx, ['pid']);
    fs.mkdirSync(path.dirname(log_path), { recursive: true });
    const logStream = fs.createWriteStream(log_path, { flags: 'a' });
    const handles = [];

    for (const unitId of be_units) {
      const port = ports[unitId];
      if (!port) throw new ToolValidationError('ports', `no port for ${unitId}`);
      const cwd = pathResolver.resolveUnitDir(ctx.pid, unitId);
      const entry = path.join(cwd, 'code', 'server.js');
      if (!fs.existsSync(entry)) {
        throw new ToolValidationError('be_units', `${unitId} has no code/server.js`);
      }
      const child = spawn('node', [entry], {
        cwd,
        env: { ...process.env, PORT: String(port), HOST: '127.0.0.1', UNIT_ID: unitId },
        detached: false,
        stdio: ['ignore', 'pipe', 'pipe'],
      });
      const prefix = `[${unitId}:${port}] `;
      child.stdout.on('data', c => logStream.write(prefix + c));
      child.stderr.on('data', c => logStream.write(prefix + c));
      handles.push({ unitId, port, pid: child.pid, proc: child });
    }
    _activeHandles.set(ctx.pid, handles);
    return { ok: true, handles: handles.map(h => ({ unitId: h.unitId, port: h.port, pid: h.pid })) };
  },
};

async function killAll(pid, graceMs = 5000) {
  const handles = _activeHandles.get(pid) || [];
  for (const h of handles) {
    try { h.proc.kill('SIGTERM'); } catch {}
  }
  await new Promise(r => setTimeout(r, graceMs));
  for (const h of handles) {
    try { if (!h.proc.killed) h.proc.kill('SIGKILL'); } catch {}
  }
  _activeHandles.delete(pid);
}

// ──────────────────── health_wait ────────────────────

const health_wait = {
  name: 'health_wait',
  description: 'poll /api/health (或指定 url) 直到 200 或超时',
  inputSchema: {
    type: 'object',
    required: ['url'],
    properties: {
      url: { type: 'string' },
      timeout_ms: { type: 'integer', default: 30_000 },
      interval_ms: { type: 'integer', default: 500 },
    },
  },
  async execute({ url, timeout_ms = 30_000, interval_ms = 500 }, _ctx) {
    const deadline = Date.now() + timeout_ms;
    while (Date.now() < deadline) {
      try {
        const r = await fetch(url);
        if (r.ok) return { ok: true, status: r.status, waited_ms: timeout_ms - (deadline - Date.now()) };
      } catch { /* not ready yet */ }
      await new Promise(r => setTimeout(r, interval_ms));
    }
    return { ok: false, timeout: true };
  },
};

// ──────────────────── parse_placeholder ────────────────────

function parsePlaceholders(obj, prev) {
  const s = JSON.stringify(obj);
  const resolved = s.replace(/\{\{step(\d+)\.([\w.]+)\}\}/g, (_match, n, p) => {
    const r = prev[parseInt(n, 10)];
    if (!r) throw new Error(`placeholder step${n} has no result yet`);
    const parts = p.split('.');
    let v = r.body;
    for (const k of parts) v = v?.[k];
    return v != null ? String(v) : '';
  });
  return JSON.parse(resolved);
}

const parse_placeholder = {
  name: 'parse_placeholder',
  description: '解析 {{stepN.field}} 占位符,从前面 step 的 response.body 取值替换',
  inputSchema: {
    type: 'object',
    required: ['template', 'previous_steps'],
    properties: {
      template: {},
      previous_steps: { type: 'array' },
    },
  },
  async execute({ template, previous_steps }, _ctx) {
    return { ok: true, resolved: parsePlaceholders(template, previous_steps) };
  },
};

// ──────────────────── run_case ────────────────────

const run_case = {
  name: 'run_case',
  description: '执行 integration spec 单个 case (含 placeholder 解析 + step expect 校验)',
  inputSchema: {
    type: 'object',
    required: ['case_spec', 'ports'],
    properties: {
      case_spec: { type: 'object' },
      ports: { type: 'object' },
      timeout_ms: { type: 'integer', default: 30_000 },
    },
  },
  async execute({ case_spec, ports, timeout_ms = 30_000 }, _ctx) {
    const deadline = Date.now() + timeout_ms;
    const stepResults = [];
    const httpTrace = [];

    for (let i = 0; i < case_spec.steps.length; i++) {
      if (Date.now() > deadline) {
        return { ok: false, failed_step: i, reason: 'timeout', httpTrace };
      }
      const step = case_spec.steps[i];
      let body, headers;
      try {
        body = parsePlaceholders(step.body || {}, stepResults);
        headers = parsePlaceholders(step.headers || {}, stepResults);
      } catch (e) {
        return { ok: false, failed_step: i, reason: e.message, httpTrace };
      }

      const [method, p] = step.call.split(' ');
      const port = ports[step.unit];
      if (!port) {
        return { ok: false, failed_step: i, reason: `no port for unit ${step.unit}`, httpTrace };
      }
      const url = `http://127.0.0.1:${port}` + p;
      const startedAt = Date.now();
      let r, json = {};
      try {
        r = await fetch(url, {
          method,
          headers: { 'Content-Type': 'application/json', ...headers },
          body: method === 'GET' ? undefined : JSON.stringify(body),
        });
        try { json = await r.json(); } catch { /* maybe empty body */ }
      } catch (e) {
        return { ok: false, failed_step: i, got: { status: 0, error: e.message }, httpTrace };
      }

      httpTrace.push({
        ts: startedAt, step: i, unit: step.unit, method, url,
        status: r.status, body: json, duration_ms: Date.now() - startedAt,
      });
      stepResults.push({ status: r.status, body: json });

      if (!_checkExpect(step.expect, { status: r.status, body: json })) {
        return { ok: false, failed_step: i, got: { status: r.status, body: json }, httpTrace };
      }
    }
    return { ok: true, httpTrace, steps_ran: case_spec.steps.length };
  },
};

function _checkExpect(expect, got) {
  if (!expect) return true;
  if (expect.status && expect.status !== got.status) return false;
  // expect.json 简化检查:仅检查 key 存在性 + 类型;不做 JSONPath
  if (expect.json && typeof expect.json === 'object') {
    for (const [k, v] of Object.entries(expect.json)) {
      if (got.body?.[k] === undefined) return false;
      if (typeof v === 'string' && v.startsWith('$.')) continue; // JSONPath 简化:存在即可
    }
  }
  return true;
}

// ──────────────────── collect_stderr ────────────────────

const collect_stderr = {
  name: 'collect_stderr',
  description: '从 process.log 取最近 N 行 stderr 节选(供 BugTriager 用)',
  inputSchema: {
    type: 'object',
    required: ['log_path'],
    properties: {
      log_path: { type: 'string' },
      lines: { type: 'integer', default: 20 },
    },
  },
  async execute({ log_path, lines = 20 }, _ctx) {
    if (!fs.existsSync(log_path)) return { ok: false, reason: 'log not exist', excerpt: '' };
    const raw = fs.readFileSync(log_path, 'utf8');
    const all = raw.split('\n');
    const excerpt = all.slice(-lines).join('\n');
    return { ok: true, excerpt };
  },
};

function _req(ctx, keys) {
  for (const k of keys) {
    if (!ctx || ctx[k] == null) throw new ToolMissingDependencyError(`ctx.${k}`);
  }
}

module.exports = {
  port_picker,
  spawn_units,
  health_wait,
  run_case,
  parse_placeholder,
  collect_stderr,
  // 内部导出（kill / 测试用）
  killAll,
};