From f1fd4cc31666b33962e9a730cb7d925e860e6db5 Mon Sep 17 00:00:00 2001 From: Your Name Date: Thu, 2 Jul 2026 10:36:23 +0800 Subject: [PATCH] feat: add agent run worker status check --- scripts/build-portal-runtime.mjs | 6 + scripts/check-agent-run-worker.mjs | 177 +++++++++++++++++++++++++++++ 2 files changed, 183 insertions(+) create mode 100644 scripts/check-agent-run-worker.mjs diff --git a/scripts/build-portal-runtime.mjs b/scripts/build-portal-runtime.mjs index 7378c43..7e71b8b 100755 --- a/scripts/build-portal-runtime.mjs +++ b/scripts/build-portal-runtime.mjs @@ -331,6 +331,10 @@ async function writeMetadata() { path.join(root, 'scripts', 'check-agent-code-run-entry.mjs'), path.join(runtimeRoot, 'scripts', 'check-agent-code-run-entry.mjs'), ); + await fs.copyFile( + path.join(root, 'scripts', 'check-agent-run-worker.mjs'), + path.join(runtimeRoot, 'scripts', 'check-agent-run-worker.mjs'), + ); await writeFile( path.join(runtimeRoot, 'RUNBOOK.txt'), [ @@ -392,6 +396,7 @@ async function writeMetadata() { ' node scripts/agent-run-worker.mjs --status', ' node scripts/agent-run-worker.mjs --once', ' bash scripts/install-agent-run-worker-agent.sh # installs disabled by default', + ' node scripts/check-agent-run-worker.mjs # read-only LaunchAgent/queue check', '', ].join('\n'), ); @@ -421,6 +426,7 @@ async function main() { await fs.chmod(path.join(runtimeRoot, 'scripts', 'install-runtime-slo-report-agent.sh'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'install-agent-run-worker-agent.sh'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'check-tool-runtime.mjs'), 0o755); + await fs.chmod(path.join(runtimeRoot, 'scripts', 'check-agent-run-worker.mjs'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'memind-portal-tunnel.sh'), 0o755); console.log(''); console.log(`Portal runtime 已生成: ${runtimeRoot}`); diff --git a/scripts/check-agent-run-worker.mjs b/scripts/check-agent-run-worker.mjs new file mode 100644 index 0000000..8b63b2e --- /dev/null +++ b/scripts/check-agent-run-worker.mjs @@ -0,0 +1,177 @@ +#!/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 [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), + 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);