/** * @file Rules-based alerting engine. Evaluates user-defined alert rules against * live activity: event-driven rules (event_pattern, token_threshold, * cost_threshold) run on every hook ingest, time-based rules (inactivity, * status_duration, budget) run on a periodic sweep. Fired alerts are persisted * to alert_events with per-scope cooldown dedup — a budget instead fires once * per threshold per period — and broadcast to clients as `alert_triggered`. * `previewRule` replays a rule over recorded history without firing anything. * @author Nguyễn Ngọc Trí Vĩ */ const { db, stmts } = require("../db"); const { broadcast } = require("../websocket"); const RULE_TYPES = [ "event_pattern", "inactivity", "status_duration", "token_threshold", "cost_threshold", "budget", ]; const TOOL_RE = /^[a-z0-9][a-z0-9._-]{0,63}$/i; const AGENT_STATUSES = ["working", "waiting"]; // Enabled-rules cache. Hook ingest is hot — re-querying alert_rules on every // event would be wasted work since rules only change through the CRUD routes, // which call invalidateRuleCache(). let rulesCache = null; function invalidateRuleCache() { rulesCache = null; } function loadEnabledRules() { if (rulesCache) return rulesCache; rulesCache = stmts.listEnabledAlertRules.all().map((row) => { let config = {}; try { config = JSON.parse(row.config || "{}"); } catch { /* tolerate hand-edited bad JSON — rule simply never matches */ } return { ...row, config }; }); return rulesCache; } /** * Validate and normalize a rule config for its type. Returns * `{ ok: true, config }` with defaults applied, or `{ ok: false, error }`. */ function validateRuleConfig(ruleType, config) { if (!RULE_TYPES.includes(ruleType)) { return { ok: false, error: `rule_type must be one of: ${RULE_TYPES.join(", ")}` }; } const cfg = config && typeof config === "object" && !Array.isArray(config) ? config : null; if (!cfg) return { ok: false, error: "config must be an object" }; const num = (v) => (typeof v === "number" && Number.isFinite(v) && v > 0 ? v : null); switch (ruleType) { case "event_pattern": { const out = {}; for (const key of ["event_type", "tool_name", "summary_contains"]) { if (cfg[key] != null) { if (typeof cfg[key] !== "string" || !cfg[key].trim()) { return { ok: false, error: `${key} must be a non-empty string` }; } out[key] = cfg[key].trim(); } } if (!out.event_type && !out.tool_name && !out.summary_contains) { return { ok: false, error: "event_pattern needs at least one of event_type, tool_name, summary_contains", }; } const count = cfg.count == null ? 1 : num(cfg.count); if (!count || !Number.isInteger(count)) { return { ok: false, error: "count must be a positive integer" }; } out.count = count; if (count > 1) { const window = cfg.window_minutes == null ? 5 : num(cfg.window_minutes); if (!window) return { ok: false, error: "window_minutes must be a positive number" }; out.window_minutes = window; } return { ok: true, config: out }; } case "inactivity": { const minutes = num(cfg.minutes); if (!minutes) return { ok: false, error: "minutes must be a positive number" }; return { ok: true, config: { minutes } }; } case "status_duration": { if (!AGENT_STATUSES.includes(cfg.status)) { return { ok: false, error: `status must be one of: ${AGENT_STATUSES.join(", ")}` }; } const minutes = num(cfg.minutes); if (!minutes) return { ok: false, error: "minutes must be a positive number" }; return { ok: true, config: { status: cfg.status, minutes } }; } case "token_threshold": { const total = num(cfg.total_tokens); if (!total || !Number.isInteger(total)) { return { ok: false, error: "total_tokens must be a positive integer" }; } return { ok: true, config: { total_tokens: total } }; } case "cost_threshold": { const usd = num(cfg.usd); if (!usd) return { ok: false, error: "usd must be a positive number" }; return { ok: true, config: { usd } }; } case "budget": { const { PERIODS } = require("./budgets"); if (!PERIODS.includes(cfg.period)) { return { ok: false, error: `period must be one of: ${PERIODS.join(", ")}` }; } const limit = num(cfg.limit_usd); if (!limit) return { ok: false, error: "limit_usd must be a positive number" }; const thresholds = cfg.thresholds == null ? [80, 100] : cfg.thresholds; if ( !Array.isArray(thresholds) || thresholds.length === 0 || thresholds.length > 10 || !thresholds.every((t) => Number.isInteger(t) && t > 0 && t <= 1000) ) { return { ok: false, error: "thresholds must be 1–10 integer percents in 1..1000" }; } const out = { period: cfg.period, limit_usd: limit, thresholds: [...new Set(thresholds)] }; if (cfg.tool != null) { if (typeof cfg.tool !== "string" || !TOOL_RE.test(cfg.tool)) { return { ok: false, error: "tool must be an AI tool id" }; } out.tool = cfg.tool; } if (cfg.lane != null) { if (!Number.isInteger(cfg.lane) || cfg.lane <= 0) { return { ok: false, error: "lane must be a workspace id" }; } out.lane = cfg.lane; } const tz = cfg.tz_offset == null ? 0 : cfg.tz_offset; if (!Number.isInteger(tz) || Math.abs(tz) > 14 * 60) { return { ok: false, error: "tz_offset must be minutes within ±840" }; } out.tz_offset = tz; return { ok: true, config: out }; } default: return { ok: false, error: "unsupported rule_type" }; } } /** * Fire an alert unless the same rule already fired for the same scope inside * its cooldown window. Persists the alert row and broadcasts it. Returns the * inserted row, or null when suppressed by cooldown. */ function fireAlert( rule, { sessionId = null, agentId = null, message, details = null, ignoreCooldown = false } ) { const last = ignoreCooldown ? null : stmts.lastAlertFor.get(rule.id, sessionId, agentId); if (last) { const elapsedMs = Date.now() - new Date(last.triggered_at).getTime(); if (elapsedMs < rule.cooldown_seconds * 1000) return null; } const info = stmts.insertAlertEvent.run( rule.id, rule.name, rule.rule_type, sessionId, agentId, message, details ? JSON.stringify(details) : null ); const alert = stmts.getAlertEvent.get(info.lastInsertRowid); broadcast("alert_triggered", alert); // Fan out to configured webhook targets. Detached and fail-safe — webhook // delivery must never slow or break alert firing. Lazy-required to keep the // module graph acyclic and tolerate any load-order edge case. try { const { dispatchAlert } = require("./webhooks"); Promise.resolve(dispatchAlert(alert)).catch(() => {}); } catch (err) { console.warn("[ALERTS] webhook dispatch failed:", err?.message || err); } return alert; } // Dynamic count-in-window queries vary by which pattern fields a rule sets; // cache prepared statements by their SQL so hot rules don't re-prepare. const countStmtCache = new Map(); function countMatchingEvents(sessionId, cfg) { const where = ["session_id = ?", "created_at >= strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)"]; const params = [sessionId, `-${cfg.window_minutes * 60} seconds`]; if (cfg.event_type) { where.push("event_type = ?"); params.push(cfg.event_type); } if (cfg.tool_name) { where.push("tool_name = ?"); params.push(cfg.tool_name); } if (cfg.summary_contains) { where.push("LOWER(COALESCE(summary, '')) LIKE ?"); params.push(`%${cfg.summary_contains.toLowerCase()}%`); } const sql = `SELECT COUNT(*) as count FROM events WHERE ${where.join(" AND ")}`; let stmt = countStmtCache.get(sql); if (!stmt) { stmt = db.prepare(sql); countStmtCache.set(sql, stmt); } return stmt.get(...params).count; } function matchesPattern(event, cfg) { if (cfg.event_type && event.event_type !== cfg.event_type) return false; if (cfg.tool_name && event.tool_name !== cfg.tool_name) return false; if ( cfg.summary_contains && !(event.summary || "").toLowerCase().includes(cfg.summary_contains.toLowerCase()) ) { return false; } return true; } // Token totals only move on hooks that read the transcript — skip the SUM // query for the rest of the event stream. const TOKEN_BEARING_EVENTS = new Set(["PostToolUse", "Stop", "SubagentStop", "SessionEnd"]); // A budget threshold already fired in this period (details carry both). const budgetFiredStmt = db.prepare( `SELECT 1 FROM alert_events WHERE rule_id = ? AND json_extract(details, '$.period_start') = ? AND json_extract(details, '$.threshold') = ?` ); /** A session's cost, priced as of its start date like GET /api/pricing/cost/:id. */ function sessionCost(sessionId, rules) { const { calculateCost } = require("../routes/pricing"); const started = stmts.getSession.get(sessionId)?.started_at; return calculateCost( stmts.getTokensBySession.all(sessionId), rules || stmts.listPricing.all(), started ? String(started).slice(0, 10) : undefined ).total_cost; } // Sweep queries are static — prepare once at module load instead of on every // 60s tick. The time window arrives as a strftime modifier parameter. const staleSessionsStmt = db.prepare( `SELECT id, name FROM sessions WHERE status = 'active' AND updated_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)` ); const stuckAgentsStmt = db.prepare( `SELECT a.id, a.session_id, a.name FROM agents a JOIN sessions s ON s.id = a.session_id WHERE s.status = 'active' AND a.status = ? AND a.updated_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)` ); /** * Evaluate event-driven rules against one freshly ingested event. Must never * throw — hook ingestion stays fail-safe regardless of rule misconfiguration. */ function evaluateEvent(event) { if (!event || !event.session_id) return; let rules; try { rules = loadEnabledRules(); } catch (err) { console.warn("[ALERTS] rule load failed:", err?.message || err); return; } for (const rule of rules) { try { if (rule.rule_type === "event_pattern") { const cfg = rule.config; if (!matchesPattern(event, cfg)) continue; if (cfg.count > 1) { const seen = countMatchingEvents(event.session_id, cfg); if (seen < cfg.count) continue; fireAlert(rule, { sessionId: event.session_id, agentId: event.agent_id || null, message: `${rule.name}: ${seen} matching events in ${cfg.window_minutes} min (threshold ${cfg.count})`, details: { matched: cfg, observed_count: seen, last_event_type: event.event_type }, }); } else { fireAlert(rule, { sessionId: event.session_id, agentId: event.agent_id || null, message: `${rule.name}: event matched (${event.event_type}${event.tool_name ? ` · ${event.tool_name}` : ""})`, details: { matched: cfg, summary: event.summary || null }, }); } } else if (rule.rule_type === "token_threshold") { if (!TOKEN_BEARING_EVENTS.has(event.event_type)) continue; const totals = stmts.sessionTokenTotals.get(event.session_id); const total = totals.input_tokens + totals.output_tokens + totals.cache_read_tokens + totals.cache_write_tokens; if (total < rule.config.total_tokens) continue; fireAlert(rule, { sessionId: event.session_id, message: `${rule.name}: session used ${total.toLocaleString()} tokens (threshold ${rule.config.total_tokens.toLocaleString()})`, details: { total_tokens: total, threshold: rule.config.total_tokens }, }); } else if (rule.rule_type === "cost_threshold") { if (!TOKEN_BEARING_EVENTS.has(event.event_type)) continue; const cost = sessionCost(event.session_id); if (cost < rule.config.usd) continue; fireAlert(rule, { sessionId: event.session_id, message: `${rule.name}: session cost $${cost.toFixed(2)} (threshold $${rule.config.usd.toFixed(2)})`, details: { cost_usd: cost, threshold_usd: rule.config.usd }, }); } } catch (err) { console.warn(`[ALERTS] rule "${rule.name}" evaluation failed:`, err?.message || err); } } } /** * Evaluate time-based rules (inactivity, status_duration). Called by the * periodic sweep; exported so tests can invoke it deterministically. */ function sweepTimeRules() { let rules; try { rules = loadEnabledRules(); } catch (err) { console.warn("[ALERTS] rule load failed:", err?.message || err); return; } for (const rule of rules) { try { if (rule.rule_type === "inactivity") { // sessions.updated_at is bumped on every ingested event (touchSession), // so "stale updated_at on an active session" ≡ "no events for N min". const stale = staleSessionsStmt.all(`-${rule.config.minutes * 60} seconds`); for (const session of stale) { fireAlert(rule, { sessionId: session.id, message: `${rule.name}: no activity on "${session.name || session.id}" for ${rule.config.minutes} min`, details: { minutes: rule.config.minutes }, }); } } else if (rule.rule_type === "status_duration") { // agents.updated_at moves on any agent update (status flips, tool // changes), so this detects agents *stuck* in a status with no // activity — the hung-agent case the rule exists for. const stuck = stuckAgentsStmt.all( rule.config.status, `-${rule.config.minutes * 60} seconds` ); for (const agent of stuck) { fireAlert(rule, { sessionId: agent.session_id, agentId: agent.id, message: `${rule.name}: agent "${agent.name}" stuck in ${rule.config.status} for ${rule.config.minutes} min`, details: { status: rule.config.status, minutes: rule.config.minutes }, }); } } else if (rule.rule_type === "budget") { sweepBudget(rule); } } catch (err) { console.warn(`[ALERTS] rule "${rule.name}" sweep failed:`, err?.message || err); } } } /** Fires each newly crossed threshold of one budget rule, once per period. */ function sweepBudget(rule, now = Date.now()) { const { budgetSpend, crossed } = require("./budgets"); const s = budgetSpend(rule.config, { now }); for (const threshold of crossed(rule.config, s.pct)) { if (budgetFiredStmt.get(rule.id, s.period_start, threshold)) continue; fireAlert(rule, { ignoreCooldown: true, message: `${rule.name}: ${threshold}% of the ${rule.config.period} budget — $${s.spend.toFixed(2)} of $${rule.config.limit_usd.toFixed(2)}`, details: { period_start: s.period_start, threshold, spend: s.spend, limit_usd: rule.config.limit_usd, projected: s.projected, }, }); } } const PREVIEW_SAMPLES = 10; /** * Replays a rule over recorded history — what it WOULD have fired — without * writing or broadcasting anything. `days` bounds the window (default 7). * `status_duration` cannot be replayed (agent status history is not stored) * and answers `supported: false`. * @returns {{supported: boolean, window_days: number, fires: number, * sessions: number, samples: Array, periods?: Array}} */ function previewRule(ruleType, config, { days = 7, cooldownSeconds = 300, now = Date.now() } = {}) { const since = new Date(now - days * 86_400_000).toISOString(); const out = { supported: true, window_days: days, fires: 0, sessions: 0, samples: [] }; const fired = new Set(); const fire = (sessionId, at, detail) => { out.fires++; fired.add(sessionId); if (out.samples.length < PREVIEW_SAMPLES) out.samples.push({ session_id: sessionId, at, ...detail }); }; if (ruleType === "event_pattern") { const where = ["created_at >= ?"]; const params = [since]; if (config.event_type) { where.push("event_type = ?"); params.push(config.event_type); } if (config.tool_name) { where.push("tool_name = ?"); params.push(config.tool_name); } if (config.summary_contains) { where.push("LOWER(COALESCE(summary, '')) LIKE ?"); params.push(`%${config.summary_contains.toLowerCase()}%`); } const windowMs = (config.window_minutes || 0) * 60_000; const recent = new Map(); // session → match times inside the window const lastFire = new Map(); const rows = db .prepare( `SELECT session_id, created_at FROM events WHERE ${where.join(" AND ")} ORDER BY created_at, id` ) .iterate(...params); for (const r of rows) { const t = Date.parse(r.created_at); let seen = 1; if (config.count > 1) { const list = (recent.get(r.session_id) || []).filter((x) => t - x <= windowMs); list.push(t); recent.set(r.session_id, list); seen = list.length; if (seen < config.count) continue; } const last = lastFire.get(r.session_id); if (last !== undefined && t - last < cooldownSeconds * 1000) continue; lastFire.set(r.session_id, t); fire(r.session_id, r.created_at, { observed_count: seen }); } } else if (ruleType === "inactivity") { // A gap longer than `minutes` between two events of one session. const gapMs = config.minutes * 60_000; const rows = db .prepare( `SELECT session_id, created_at, LAG(created_at) OVER (PARTITION BY session_id ORDER BY created_at, id) AS prev FROM events WHERE created_at >= ?` ) .iterate(since); for (const r of rows) { if (!r.prev) continue; const gap = Date.parse(r.created_at) - Date.parse(r.prev); if (gap > gapMs) fire(r.session_id, r.prev, { gap_minutes: Math.round(gap / 60_000) }); } } else if (ruleType === "token_threshold" || ruleType === "cost_threshold") { const rules = stmts.listPricing.all(); const sessions = db .prepare("SELECT id, updated_at FROM sessions WHERE updated_at >= ? ORDER BY updated_at") .all(since); for (const s of sessions) { if (ruleType === "token_threshold") { const t = stmts.sessionTokenTotals.get(s.id); const total = t.input_tokens + t.output_tokens + t.cache_read_tokens + t.cache_write_tokens; if (total >= config.total_tokens) fire(s.id, s.updated_at, { total_tokens: total }); } else { const cost = sessionCost(s.id, rules); if (cost >= config.usd) fire(s.id, s.updated_at, { cost_usd: cost }); } } } else if (ruleType === "budget") { // The current period and the five before it. const { budgetSpend, crossed } = require("./budgets"); const rules = stmts.listPricing.all(); out.periods = []; for (let back = 5; back >= 0; back--) { const s = budgetSpend(config, { now, back, rules }); const hit = crossed(config, s.pct); out.fires += hit.length; out.periods.push({ ...s, crossed: hit }); } } else { out.supported = false; } out.sessions = fired.size; return out; } // Periodic sweep for the time-based rules. unref'd so it never keeps the // process (or the test runner) alive — same pattern as the hooks watchdog. const SWEEP_INTERVAL_MS = 60_000; const sweepTimer = setInterval(sweepTimeRules, SWEEP_INTERVAL_MS); if (sweepTimer.unref) sweepTimer.unref(); module.exports = { RULE_TYPES, validateRuleConfig, evaluateEvent, sweepTimeRules, fireAlert, invalidateRuleCache, previewRule, sweepBudget, };