#!/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'; import { buildPausePlan, evaluateHealth } from './agent-run-guard-lib.mjs'; const execFileAsync = promisify(execFile); const DEFAULT_WORKER_LABEL = 'cn.tkmind.memind-agent-run-worker'; const DEFAULT_PORTAL_LABEL = 'cn.tkmind.memind-portal'; 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(); let value = trimmed.slice(idx + 1).trim(); if ((value.startsWith('"') && value.endsWith('"')) || (value.startsWith("'") && value.endsWith("'"))) { value = value.slice(1, -1); } if (!process.env[key]) process.env[key] = value; } } function truthy(value) { return ['1', 'true', 'yes', 'on'].includes(String(value ?? '').trim().toLowerCase()); } function positiveInt(value, fallback) { const parsed = Number(value); return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : fallback; } function parseArgs(argv) { const args = { apply: argv.includes('--apply'), dryRun: argv.includes('--dry-run') || !argv.includes('--apply'), help: argv.includes('--help') || argv.includes('-h'), }; return args; } function printHelp() { console.log([ 'Usage:', ' node scripts/agent-run-guard.mjs [--dry-run]', ' node scripts/agent-run-guard.mjs --apply', '', 'Dry-run is the default. --apply stops the external worker and disables code-run gate in .env', 'on the first pause trigger only. Repeat checks while already paused are no-ops.', '', 'Recovery after manual intervention:', ' bash scripts/resume-agent-run-prod.sh', ].join('\n')); } 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 guard: ${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 replaceOrAppendEnv(raw, updates) { const pending = new Map(Object.entries(updates)); const lines = raw.split('\n'); const next = lines.map((line) => { const match = line.match(/^(\s*)([A-Za-z_][A-Za-z0-9_]*)(\s*=).*/); if (!match) return line; const key = match[2]; if (!pending.has(key)) return line; const value = pending.get(key); pending.delete(key); return `${key}=${value}`; }); if (pending.size > 0) { if (next.length > 0 && next[next.length - 1] !== '') next.push(''); next.push('## agent-run guard auto-disable'); for (const [key, value] of pending) next.push(`${key}=${value}`); } return next.join('\n').replace(/\n*$/, '\n'); } function timestamp() { const d = new Date(); const pad = (n) => String(n).padStart(2, '0'); return `${d.getFullYear()}${pad(d.getMonth() + 1)}${pad(d.getDate())}-${pad(d.getHours())}${pad(d.getMinutes())}${pad(d.getSeconds())}`; } async function readQueueHealth(now) { 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 [oldestPendingRows] = await conn.query( `SELECT MIN(updated_at) AS oldest_updated_at FROM h5_agent_runs WHERE status IN ('queued', 'retryable')`, ); const oldestPendingUpdatedAt = oldestPendingRows[0]?.oldest_updated_at == null ? null : Number(oldestPendingRows[0].oldest_updated_at); const [oldestRunningRows] = 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 = oldestRunningRows[0]?.started_at == null ? null : Number(oldestRunningRows[0].started_at); const oldestRunningHeartbeatAt = oldestRunningRows[0]?.latest_heartbeat_at == null ? null : Number(oldestRunningRows[0].latest_heartbeat_at); const oldestRunningHeartbeatAgeMs = oldestRunningRows[0] ? Math.max(0, now - Number(oldestRunningRows[0].latest_heartbeat_at ?? oldestRunningRows[0].started_at ?? 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 failedWindowMs = positiveInt(process.env.MEMIND_AGENT_RUN_GUARD_FAILED_WINDOW_MS, 10 * 60 * 1000); const since = now - failedWindowMs; const [failedRows] = await conn.query( `SELECT COUNT(*) AS count FROM h5_agent_runs WHERE status = 'failed' AND updated_at >= ?`, [since], ); const [latestFailedRows] = 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, queuedOrRetryable: Number(statusCounts.queued ?? 0) + Number(statusCounts.retryable ?? 0), running: Number(statusCounts.running ?? 0), oldestPendingUpdatedAt, oldestPendingAgeMs: oldestPendingUpdatedAt == null ? 0 : Math.max(0, now - oldestPendingUpdatedAt), oldestRunningStartedAt, oldestRunningAgeMs: oldestRunningStartedAt == null ? 0 : Math.max(0, now - oldestRunningStartedAt), oldestRunningHeartbeatAt, oldestRunningHeartbeatAgeMs, runningWithoutHeartbeatCount: Number(runningWithoutHeartbeatRows[0]?.count ?? 0), latestRunningRun: oldestRunningRows[0] ? { id: oldestRunningRows[0].id, requestId: oldestRunningRows[0].request_id, startedAt: oldestRunningStartedAt, updatedAt: oldestRunningRows[0].updated_at == null ? null : Number(oldestRunningRows[0].updated_at), heartbeatAt: oldestRunningHeartbeatAt, heartbeatAgeMs: oldestRunningHeartbeatAgeMs, } : null, failedWindowMs, failedRecentCount: Number(failedRows[0]?.count ?? 0), latestFailedRun: latestFailedRows[0] ? { id: latestFailedRows[0].id, requestId: latestFailedRows[0].request_id, error: latestFailedRows[0].error_message, updatedAt: Number(latestFailedRows[0].updated_at ?? 0), } : null, }; } finally { await conn.end(); } } async function disableCodeRuns({ root, envFile, workerLabel, portalLabel, gui, reasons, dryRun }) { const actions = []; const updates = { MEMIND_AGENT_CODE_RUNS_ENABLED: '0', MEMIND_AGENT_RUN_AUTODISPATCH: '1', }; if (dryRun) { return { applied: false, actions: [ `would backup ${envFile}`, `would set ${Object.entries(updates).map(([k, v]) => `${k}=${v}`).join(', ')}`, `would stop and disable ${workerLabel}`, `would kickstart ${portalLabel}`, ], }; } const backupRoot = process.env.MEMIND_AGENT_RUN_GUARD_BACKUP_DIR || path.join(path.dirname(root), 'memind_backups', `${timestamp()}-agent-run-guard-pause`); fs.mkdirSync(backupRoot, { recursive: true }); const envBackup = path.join(backupRoot, '.env.before'); fs.copyFileSync(envFile, envBackup); actions.push(`backed_up_env:${envBackup}`); const raw = fs.readFileSync(envFile, 'utf8'); const updated = replaceOrAppendEnv(raw, updates); const marker = [ '', `# agent-run guard pause at ${new Date().toISOString()}`, `# reasons: ${reasons.join('; ')}`, ].join('\n'); fs.writeFileSync(envFile, `${updated.replace(/\n*$/, '\n')}${marker}\n`, 'utf8'); actions.push(`updated_env:${envFile}`); await runCommand('launchctl', ['bootout', `${gui}/${workerLabel}`]); await runCommand('launchctl', ['disable', `${gui}/${workerLabel}`]); actions.push(`disabled_worker:${workerLabel}`); const kick = await runCommand('launchctl', ['kickstart', '-k', `${gui}/${portalLabel}`]); actions.push(kick.ok ? `kickstarted_portal:${portalLabel}` : `portal_kickstart_failed:${kick.message}`); return { applied: true, backupRoot, actions }; } const args = parseArgs(process.argv.slice(2)); if (args.help) { printHelp(); process.exit(0); } const root = path.join(path.dirname(new URL(import.meta.url).pathname), '..'); const envFile = process.env.MEMIND_ENV_FILE || path.join(root, '.env'); loadEnvFile(envFile); const envRaw = fs.existsSync(envFile) ? fs.readFileSync(envFile, 'utf8') : ''; const enabled = process.env.MEMIND_AGENT_RUN_GUARD_ENABLED == null ? true : truthy(process.env.MEMIND_AGENT_RUN_GUARD_ENABLED); const workerLabel = process.env.MEMIND_AGENT_RUN_WORKER_LABEL || DEFAULT_WORKER_LABEL; const portalLabel = process.env.MEMIND_PORTAL_LABEL || DEFAULT_PORTAL_LABEL; const gui = `gui/${process.getuid()}`; const now = Date.now(); const queue = await readQueueHealth(now).catch((err) => ({ error: err instanceof Error ? (err.message || err.code || err.name) : String(err), })); const evaluation = queue.error ? { thresholds: {}, reasons: [`queue_health_error ${queue.error}`], shouldPause: false } : evaluateHealth(queue); const pausePlan = buildPausePlan({ shouldPause: evaluation.shouldPause, envRaw, dryRun: args.dryRun, }); let pause = { applied: false, skipped: pausePlan.action === 'skip', alreadyPaused: pausePlan.alreadyPaused, actions: [...pausePlan.actions], }; if (enabled && pausePlan.action === 'apply') { const applied = await disableCodeRuns({ root, envFile, workerLabel, portalLabel, gui, reasons: evaluation.reasons, dryRun: args.dryRun, }); pause = { ...pause, ...applied, skipped: false, alreadyPaused: false, }; } const result = { ok: enabled ? !evaluation.shouldPause || pause.applied || pause.skipped || args.dryRun : true, checkedAt: new Date(now).toISOString(), enabled, mode: args.apply ? 'apply' : 'dry-run', queue, evaluation, pause, }; console.log(JSON.stringify(result, null, 2)); process.exit(result.ok ? 0 : 1);