← back to Animals
agents/animal-agent/server.js
265 lines
// Animal Agent — autonomous nightly builder.
//
// Runs as a long-lived pm2 process. Internal cron schedules tasks into
// `agent_tasks`; a worker loop pulls 1 task at a time and executes it.
// HTTP control surface for Steve to inspect / pause / kick:
//
// GET /healthz — liveness
// GET / — pretty-printed status dashboard
// GET /api/status — JSON snapshot of recent runs + queue
// POST /api/queue — manually queue { kind, payload_json }
// POST /api/pause — pause the worker loop
// POST /api/resume — resume
//
// Per Steve's no-cloud-agents rule, this lives entirely on Mac Studio 2.
// Per the "trust but verify" rule, every task records what it ran + the
// stdout/stderr tail in `agent_runs.notes` for audit.
import 'dotenv/config';
import express from 'express';
import { spawn } from 'node:child_process';
import path from 'node:path';
import fs from 'node:fs';
import { fileURLToPath } from 'node:url';
import { pool, many, one } from '../../src/lib/db.js';
import { log } from '../../src/lib/log.js';
const __dirname = path.dirname(fileURLToPath(import.meta.url));
const ROOT = path.resolve(__dirname, '..', '..');
let paused = false;
let currentTask = null;
const app = express();
app.use(express.json({ limit: '128kb' }));
// LAN/tailnet auth gate (env-only, no source fallback)
const BASIC_AUTH = process.env.BASIC_AUTH;
if (!BASIC_AUTH) {
console.error('[fatal] BASIC_AUTH env var unset — refusing to start');
process.exit(1);
}
const AUTH_HEADER = 'Basic ' + Buffer.from(BASIC_AUTH).toString('base64');
app.use((req, res, next) => {
if (req.path === '/health' || req.path === '/healthz') return next();
if (req.get('authorization') === AUTH_HEADER) return next();
res.set('WWW-Authenticate', 'Basic realm="mac2-pm2"');
res.status(401).send('auth required');
});
app.get('/healthz', (_req, res) => res.json({ ok: true, paused, currentTask: currentTask?.id || null }));
app.get('/', async (_req, res) => {
const queue = await many(`SELECT id, kind, status, scheduled_for, attempts FROM agent_tasks WHERE status IN ('queued','running') ORDER BY priority, scheduled_for LIMIT 30`);
const recent = await many(`SELECT id, kind, status, started_at, finished_at, error FROM agent_tasks WHERE finished_at IS NOT NULL ORDER BY finished_at DESC LIMIT 30`);
const runs = await many(`SELECT * FROM agent_runs ORDER BY started_at DESC LIMIT 10`);
res.send(`<!doctype html><html><head><meta charset=utf-8><title>Animal Agent</title>
<style>body{font:14px -apple-system,sans-serif;background:#0f1117;color:#e4e4e7;padding:20px;max-width:1100px;margin:0 auto}h1{color:#0f4d3a}h1 small{color:#888;font-size:.5em;font-weight:400}h2{margin-top:1.5em;color:#c97a3e}table{width:100%;border-collapse:collapse;background:#1a1d27;border:1px solid #2a2d3a;border-radius:8px;overflow:hidden}th,td{padding:8px 12px;text-align:left;border-bottom:1px solid #2a2d3a;font-size:13px}th{background:#22252f;color:#888;font-weight:600;text-transform:uppercase;letter-spacing:.05em;font-size:11px}.s-queued{color:#60a5fa}.s-running{color:#facc15}.s-completed{color:#4ade80}.s-failed{color:#f87171}form{display:inline}button{background:#c97a3e;color:#fff;border:0;padding:.5em 1em;border-radius:6px;font-weight:600;cursor:pointer;margin-right:.5em}button.pause{background:#f87171}.dot{display:inline-block;width:8px;height:8px;border-radius:50%;margin-right:6px}.dot.green{background:#4ade80}.dot.yellow{background:#facc15}</style>
</head><body>
<h1>🐾 Animal Agent <small>${paused ? '<span class=dot></span>PAUSED' : '<span class="dot green"></span>RUNNING'} · current: ${currentTask ? `${currentTask.kind} (#${currentTask.id})` : 'idle'}</small></h1>
<form onsubmit="event.preventDefault();fetch(this.action,{method:'POST'}).then(_=>location.reload())" action="${paused ? '/api/resume' : '/api/pause'}">
<button type=submit class="${paused ? '' : 'pause'}">${paused ? 'Resume' : 'Pause'}</button>
</form>
<form onsubmit="event.preventDefault();const k=prompt('kind?');if(k)fetch('/api/queue',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({kind:k})}).then(_=>location.reload())">
<button type=submit>+ Queue task</button>
</form>
<h2>Queue (next ${queue.length})</h2>
<table><tr><th>id</th><th>kind</th><th>status</th><th>scheduled</th><th>attempts</th></tr>
${queue.map(t => `<tr><td>${t.id}</td><td>${t.kind}</td><td class=s-${t.status}>${t.status}</td><td>${new Date(t.scheduled_for).toLocaleString()}</td><td>${t.attempts}</td></tr>`).join('')}
</table>
<h2>Recent (last ${recent.length})</h2>
<table><tr><th>id</th><th>kind</th><th>status</th><th>finished</th><th>err</th></tr>
${recent.map(t => `<tr><td>${t.id}</td><td>${t.kind}</td><td class=s-${t.status}>${t.status}</td><td>${t.finished_at?new Date(t.finished_at).toLocaleTimeString():''}</td><td>${(t.error||'').slice(0,80)}</td></tr>`).join('')}
</table>
<h2>Runs</h2>
<table><tr><th>id</th><th>started</th><th>ok</th><th>fail</th><th>notes</th></tr>
${runs.map(r => `<tr><td>${r.id}</td><td>${new Date(r.started_at).toLocaleString()}</td><td>${r.ok_count}</td><td>${r.fail_count}</td><td>${(r.notes||'').slice(0,120)}</td></tr>`).join('')}
</table>
</body></html>`);
});
app.get('/api/status', async (_req, res) => {
const queue = await many(`SELECT id, kind, status, scheduled_for FROM agent_tasks WHERE status IN ('queued','running') ORDER BY priority, scheduled_for LIMIT 50`);
const counts = await many(`SELECT status, COUNT(*)::int AS n FROM agent_tasks GROUP BY status`);
res.json({ paused, currentTask: currentTask?.id || null, queue, counts });
});
app.post('/api/queue', async (req, res) => {
const { kind, payload_json, priority, cron_label } = req.body || {};
if (!kind) return res.status(400).json({ error: 'kind required' });
const r = await one(`INSERT INTO agent_tasks (kind, payload_json, priority, cron_label) VALUES ($1, $2, $3, $4) RETURNING id`,
[kind, payload_json || {}, priority || 5, cron_label || 'manual']);
res.json({ ok: true, id: r.id });
});
app.post('/api/pause', (_req, res) => { paused = true; res.json({ ok: true, paused }); });
app.post('/api/resume', (_req, res) => { paused = false; res.json({ ok: true, paused }); });
const PORT = parseInt(process.env.PORT || '9725', 10);
app.listen(PORT, () => log.info(`animal-agent listening on :${PORT}`));
// ── cron rules (internal, no external scheduler needed) ───────────────────
// Each rule: { kind, cron_label, every_minutes, throttle_minutes, cron_dow?, cron_hour? }
// `throttle_minutes` = don't queue if a task with same cron_label was created
// in the last N minutes (keeps the queue from blowing up after a restart).
// `cron_dow` (0=Sun..6=Sat) and `cron_hour` (0..23) gate the rule to specific
// times — cronTick still runs every minute, but skips the rule unless local
// day-of-week and hour match. Useful for weekly digests that should fire on
// a specific day.
const CRON_RULES = [
{ kind: 'audit_websites', cron_label: 'cron.audit_15m', every_minutes: 15, throttle_minutes: 14, payload: { limit: 25 } },
{ kind: 'gen_mockups', cron_label: 'cron.mockups_30m', every_minutes: 30, throttle_minutes: 28, payload: { score_below: 65, limit: 15 } },
{ kind: 'enrich_emails', cron_label: 'cron.emails_60m', every_minutes: 60, throttle_minutes: 55, payload: { limit: 60, refresh_days: 30 } },
{ kind: 'wikimedia_breeds', cron_label: 'cron.wiki_180m', every_minutes: 180, throttle_minutes: 170, payload: { limit: 25 } },
{ kind: 'github_data', cron_label: 'cron.github_360m', every_minutes: 360, throttle_minutes: 340, payload: {} },
{ kind: 'ingest_osm_state', cron_label: 'cron.osm_240m', every_minutes: 240, throttle_minutes: 230, payload: { state: 'ROTATE' } },
{ kind: 'codex_debate', cron_label: 'cron.codex_360m', every_minutes: 360, throttle_minutes: 340, payload: { rounds: 4, budget: 4 } },
{ kind: 'build_subsite', cron_label: 'cron.subsite_60m', every_minutes: 60, throttle_minutes: 55, payload: { limit: 5 } },
// Weekly digest — Sundays at 08:00 local. Throttle far over a week so a
// restart late Sunday doesn't double-fire next Sunday.
{ kind: 'weekly_newsletter', cron_label: 'cron.newsletter_weekly', every_minutes: 60, throttle_minutes: 10080, cron_dow: 0, cron_hour: 8, payload: {} },
];
// Rotation pool for OSM state ingest
const STATE_ROTATION = ['CA','TX','FL','NY','IL','WA','CO','GA','MA','PA','OH','NC','MI','VA','AZ','NJ','TN','MO','MD','WI'];
// ── cron tick (every minute) ──────────────────────────────────────────────
async function cronTick() {
if (paused) return;
const now = new Date();
for (const rule of CRON_RULES) {
try {
// Day-of-week / hour gate (used by weekly_newsletter on Sundays at 08:00).
if (rule.cron_dow !== undefined && now.getDay() !== rule.cron_dow) continue;
if (rule.cron_hour !== undefined && now.getHours() !== rule.cron_hour) continue;
const recent = await one(
`SELECT 1 AS x FROM agent_tasks WHERE cron_label = $1 AND created_at > NOW() - ($2 || ' minutes')::interval LIMIT 1`,
[rule.cron_label, rule.throttle_minutes]);
if (recent) continue;
let payload = { ...rule.payload };
// Rotate state for OSM ingest
if (payload.state === 'ROTATE') {
const idx = (await one(`SELECT COUNT(*)::int AS n FROM agent_tasks WHERE cron_label = $1`, [rule.cron_label]))?.n || 0;
payload.state = STATE_ROTATION[idx % STATE_ROTATION.length];
}
await pool.query(
`INSERT INTO agent_tasks (kind, payload_json, priority, cron_label) VALUES ($1, $2, $3, $4)`,
[rule.kind, JSON.stringify(payload), 5, rule.cron_label]);
log.info(`cron queued ${rule.kind}`, { label: rule.cron_label, payload });
} catch (err) { log.error('cron rule err', { rule: rule.cron_label, err: err.message }); }
}
}
setInterval(cronTick, 60_000);
setTimeout(cronTick, 5000); // run once shortly after start
// ── worker loop ───────────────────────────────────────────────────────────
async function pickTask() {
// Atomic claim: SELECT … FOR UPDATE SKIP LOCKED
return await pool.connect().then(async client => {
try {
await client.query('BEGIN');
const r = await client.query(
`SELECT * FROM agent_tasks
WHERE status = 'queued' AND scheduled_for <= NOW()
ORDER BY priority, scheduled_for
FOR UPDATE SKIP LOCKED
LIMIT 1`);
if (!r.rows.length) { await client.query('COMMIT'); return null; }
const task = r.rows[0];
await client.query(
`UPDATE agent_tasks SET status='running', started_at=NOW(), attempts=attempts+1 WHERE id=$1`,
[task.id]);
await client.query('COMMIT');
return task;
} catch (e) { await client.query('ROLLBACK').catch(() => {}); throw e; }
finally { client.release(); }
});
}
async function workerLoop() {
while (true) {
try {
if (paused) { await sleep(5000); continue; }
const task = await pickTask();
if (!task) { await sleep(8000); continue; }
currentTask = task;
log.info(`▶ exec task ${task.id}`, { kind: task.kind });
const runId = (await one(`INSERT INTO agent_runs DEFAULT VALUES RETURNING id`))?.id;
let result; let err = null;
try {
result = await execTask(task);
await pool.query(`UPDATE agent_tasks SET status='completed', finished_at=NOW(), result_json=$2 WHERE id=$1`,
[task.id, JSON.stringify(result || {})]);
await pool.query(`UPDATE agent_runs SET finished_at=NOW(), ok_count=1, notes=$2 WHERE id=$1`,
[runId, `${task.kind}: ok ${JSON.stringify(result||{}).slice(0,200)}`]);
log.info(`✓ done task ${task.id}`, { kind: task.kind });
} catch (e) {
err = e.message;
const next = task.attempts >= task.max_attempts ? 'failed' : 'queued';
const delay = Math.min(60, 5 * Math.pow(2, task.attempts));
await pool.query(
`UPDATE agent_tasks SET status=$2, finished_at=CASE WHEN $2='failed' THEN NOW() ELSE NULL END,
scheduled_for = CASE WHEN $2='queued' THEN NOW() + ($3 || ' minutes')::interval ELSE scheduled_for END,
error=$4 WHERE id=$1`,
[task.id, next, delay, err]);
await pool.query(`UPDATE agent_runs SET finished_at=NOW(), fail_count=1, notes=$2 WHERE id=$1`,
[runId, `${task.kind}: ${err}`.slice(0,400)]);
log.error(`✗ fail task ${task.id}`, { kind: task.kind, err });
} finally {
currentTask = null;
}
} catch (e) {
log.error('workerLoop err', { err: e.message });
await sleep(5000);
}
}
}
workerLoop();
// ── task executors ───────────────────────────────────────────────────────
async function execTask(task) {
const p = task.payload_json || {};
const npm = (script, args = []) => runCmd('npm', ['run', script, '--', ...args], 30 * 60_000);
switch (task.kind) {
case 'audit_websites': return await npm('audit:websites', ['--limit', String(p.limit || 25)]);
case 'gen_mockups': return await npm('audit:mockups-unique', ['--score-below', String(p.score_below || 60), '--limit', String(p.limit || 10)]);
case 'enrich_emails': return await npm('enrich:emails', ['--limit', String(p.limit || 50), '--refresh-days', String(p.refresh_days || 30)]);
case 'wikimedia_breeds': return await npm('enrich:breed-gallery-v2', ['--limit', String(p.limit || 25)]);
case 'github_data': return await npm('ingest:github-data');
case 'ingest_osm_state': return await npm('ingest:osm-state', [p.state || 'CA']);
case 'ingest_reddit': return await npm('ingest:reddit');
case 'codex_debate': return await runCmd(path.resolve(process.env.HOME, '.claude/skills/claude-codex/scripts/start.sh'),
['--name', `animal-agent-${Date.now()}`, '--workdir', ROOT,
'--rounds', String(p.rounds || 4), '--budget', String(p.budget || 4),
'--apply'], 60_000); // APPLY MODE — codex actually edits the code
case 'build_subsite': return await runCmd('node', [path.join(ROOT, 'src/scripts/build_subsites.js'), '--limit', String(p.limit || 5)], 15 * 60_000);
case 'weekly_newsletter': {
const args = [];
if (p.limit) args.push('--limit', String(p.limit));
if (p.user_ids) args.push('--user-ids', Array.isArray(p.user_ids) ? p.user_ids.join(',') : String(p.user_ids));
if (p.dry_run) args.push('--dry-run');
return await runCmd('node', [path.join(ROOT, 'src/scripts/run_weekly_newsletter.js'), ...args], 30 * 60_000);
}
default: throw new Error(`unknown kind: ${task.kind}`);
}
}
function runCmd(cmd, args, timeoutMs = 600_000) {
return new Promise((resolve, reject) => {
const child = spawn(cmd, args, { cwd: ROOT, env: process.env });
let out = '', err = '';
const timer = setTimeout(() => { child.kill('SIGKILL'); reject(new Error(`timeout after ${timeoutMs}ms`)); }, timeoutMs);
child.stdout.on('data', d => { out += d.toString(); if (out.length > 8000) out = out.slice(-8000); });
child.stderr.on('data', d => { err += d.toString(); if (err.length > 8000) err = err.slice(-8000); });
child.on('close', (code) => {
clearTimeout(timer);
if (code === 0) resolve({ code, out_tail: out.split('\n').slice(-15).join('\n') });
else reject(new Error(`exit ${code}: ${err.split('\n').slice(-5).join(' | ')}`));
});
child.on('error', e => { clearTimeout(timer); reject(e); });
});
}
function sleep(ms) { return new Promise(r => setTimeout(r, ms)); }