memind/scripts/check-agent-run-worker.mjs

#!/usr/bin/env node
import fs from 'node:fs';
import path from 'node:path';
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import mysql from 'mysql2/promise';

const execFileAsync = promisify(execFile);
const DEFAULT_LABEL = 'cn.tkmind.memind-agent-run-worker';

function loadEnvFile(filePath) {
  if (!fs.existsSync(filePath)) return;
  for (const line of fs.readFileSync(filePath, 'utf8').split('\n')) {
    const trimmed = line.trim();
    if (!trimmed || trimmed.startsWith('#')) continue;
    const idx = trimmed.indexOf('=');
    if (idx < 0) continue;
    const key = trimmed.slice(0, idx).trim();
    const value = trimmed.slice(idx + 1).trim();
    if (!process.env[key]) process.env[key] = value;
  }
}

function truthy(value) {
  return ['1', 'true', 'yes', 'on'].includes(String(value ?? '').trim().toLowerCase());
}

function parseMysqlConfig() {
  if (process.env.DATABASE_URL) {
    const url = new URL(process.env.DATABASE_URL);
    if (url.protocol !== 'mysql:') {
      throw new Error(`Unsupported DATABASE_URL scheme for agent run worker check: ${url.protocol}`);
    }
    return {
      host: url.hostname,
      port: Number(url.port || 3306),
      user: decodeURIComponent(url.username),
      password: decodeURIComponent(url.password),
      database: url.pathname.replace(/^\/+/, ''),
      charset: 'utf8mb4',
    };
  }
  return {
    host: process.env.MYSQL_HOST,
    port: Number(process.env.MYSQL_PORT || 3306),
    user: process.env.MYSQL_USER,
    password: process.env.MYSQL_PASSWORD,
    database: process.env.MYSQL_DATABASE,
    charset: 'utf8mb4',
  };
}

async function runCommand(command, args) {
  try {
    const { stdout, stderr } = await execFileAsync(command, args, {
      maxBuffer: 1024 * 1024,
    });
    return { ok: true, stdout, stderr };
  } catch (err) {
    return {
      ok: false,
      code: err?.code ?? null,
      stdout: err?.stdout ?? '',
      stderr: err?.stderr ?? '',
      message: err instanceof Error ? err.message : String(err),
    };
  }
}

function parseLaunchctlPrint(text) {
  const state = text.match(/\bstate = ([^\n]+)/)?.[1]?.trim() ?? null;
  const pid = text.match(/\bpid = ([0-9]+)/)?.[1] ?? null;
  const program = text.match(/\bprogram = ([^\n]+)/)?.[1]?.trim() ?? null;
  const pathValue = text.match(/\bpath = ([^\n]+)/)?.[1]?.trim() ?? null;
  return {
    loaded: Boolean(state || pid || program || pathValue),
    state,
    pid: pid == null ? null : Number(pid),
    program,
    path: pathValue,
  };
}

function parseDisabled(text, label) {
  const escaped = label.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
  const match = text.match(new RegExp(`"${escaped}"\\s*=>\\s*(disabled|enabled)`));
  return match?.[1] === 'disabled';
}

async function readQueueSummary() {
  const conn = await mysql.createConnection(parseMysqlConfig());
  try {
    const [statusRows] = await conn.query(
      `SELECT status, COUNT(*) AS count
       FROM h5_agent_runs
       WHERE status IN ('queued', 'running', 'retryable')
       GROUP BY status`,
    );
    const statusCounts = {};
    for (const row of statusRows) statusCounts[row.status] = Number(row.count ?? 0);

    const [lagRows] = await conn.query(
      `SELECT MIN(updated_at) AS oldest_updated_at
       FROM h5_agent_runs
       WHERE status IN ('queued', 'retryable')`,
    );
    const oldestUpdatedAt = lagRows[0]?.oldest_updated_at == null
      ? null
      : Number(lagRows[0].oldest_updated_at);

    const [runningRows] = await conn.query(
      `SELECT
         r.id,
         r.request_id,
         r.started_at,
         r.updated_at,
         h.latest_heartbeat_at
       FROM h5_agent_runs r
       LEFT JOIN (
         SELECT run_id, MAX(created_at) AS latest_heartbeat_at
         FROM h5_agent_run_events
         WHERE event_type = 'worker_heartbeat'
         GROUP BY run_id
       ) h ON h.run_id = r.id
       WHERE r.status = 'running'
       ORDER BY COALESCE(h.latest_heartbeat_at, r.started_at) ASC
       LIMIT 1`,
    );
    const oldestRunningStartedAt = runningRows[0]?.started_at == null
      ? null
      : Number(runningRows[0].started_at);
    const oldestRunningHeartbeatAt = runningRows[0]?.latest_heartbeat_at == null
      ? null
      : Number(runningRows[0].latest_heartbeat_at);
    const oldestRunningHeartbeatAgeMs = runningRows[0]
      ? Math.max(0, Date.now() - Number(runningRows[0].latest_heartbeat_at ?? runningRows[0].started_at ?? Date.now()))
      : 0;
    const [runningWithoutHeartbeatRows] = await conn.query(
      `SELECT COUNT(*) AS count
       FROM h5_agent_runs r
       LEFT JOIN (
         SELECT run_id, MAX(created_at) AS latest_heartbeat_at
         FROM h5_agent_run_events
         WHERE event_type = 'worker_heartbeat'
         GROUP BY run_id
       ) h ON h.run_id = r.id
       WHERE r.status = 'running' AND h.latest_heartbeat_at IS NULL`,
    );

    const [failedRows] = await conn.query(
      `SELECT id, request_id, error_message, updated_at
       FROM h5_agent_runs
       WHERE status = 'failed'
       ORDER BY updated_at DESC
       LIMIT 1`,
    );

    return {
      statusCounts,
      oldestPendingUpdatedAt: oldestUpdatedAt,
      oldestPendingAgeMs: oldestUpdatedAt == null ? 0 : Math.max(0, Date.now() - oldestUpdatedAt),
      oldestRunningStartedAt,
      oldestRunningAgeMs: oldestRunningStartedAt == null ? 0 : Math.max(0, Date.now() - oldestRunningStartedAt),
      oldestRunningHeartbeatAt,
      oldestRunningHeartbeatAgeMs,
      runningWithoutHeartbeatCount: Number(runningWithoutHeartbeatRows[0]?.count ?? 0),
      latestRunningRun: runningRows[0] ? {
        id: runningRows[0].id,
        requestId: runningRows[0].request_id,
        startedAt: oldestRunningStartedAt,
        updatedAt: runningRows[0].updated_at == null ? null : Number(runningRows[0].updated_at),
        heartbeatAt: oldestRunningHeartbeatAt,
        heartbeatAgeMs: oldestRunningHeartbeatAgeMs,
      } : null,
      latestFailedRun: failedRows[0] ? {
        id: failedRows[0].id,
        requestId: failedRows[0].request_id,
        error: failedRows[0].error_message,
        updatedAt: Number(failedRows[0].updated_at ?? 0),
      } : null,
    };
  } finally {
    await conn.end();
  }
}

const root = path.join(path.dirname(new URL(import.meta.url).pathname), '..');
loadEnvFile(process.env.MEMIND_ENV_FILE || path.join(root, '.env'));

const label = process.env.MEMIND_AGENT_RUN_WORKER_LABEL || DEFAULT_LABEL;
const gui = `gui/${process.getuid()}`;
const plist = path.join(process.env.HOME || '', 'Library', 'LaunchAgents', `${label}.plist`);
const expectRunning = truthy(process.env.MEMIND_AGENT_RUN_WORKER_EXPECT_RUNNING);

const printResult = await runCommand('launchctl', ['print', `${gui}/${label}`]);
const disabledResult = await runCommand('launchctl', ['print-disabled', gui]);
const pgrepResult = await runCommand('pgrep', ['-fl', 'agent-run-worker.mjs']);
const launchd = parseLaunchctlPrint(`${printResult.stdout}\n${printResult.stderr}`);
const disabled = disabledResult.ok ? parseDisabled(disabledResult.stdout, label) : null;
const pgrepLines = pgrepResult.ok
  ? pgrepResult.stdout.split('\n').map((line) => line.trim()).filter(Boolean)
  : [];
const queue = await readQueueSummary().catch((err) => ({
  error: err instanceof Error ? (err.message || err.code || err.name) : String(err),
}));

const installed = fs.existsSync(plist);
const running = launchd.state === 'running' || pgrepLines.length > 0;
const ok = Boolean(
  installed &&
  !queue.error &&
  (expectRunning ? running : disabled === true && !running),
);

console.log(JSON.stringify({
  ok,
  checkedAt: new Date().toISOString(),
  expected: expectRunning ? 'running' : 'disabled',
  label,
  plist,
  installed,
  disabled,
  running,
  launchd,
  processes: pgrepLines,
  queue,
}, null, 2));

process.exit(ok ? 0 : 1);