#!/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);