AgentHub/server/lib/alerts.js

/**
 * @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.
 * `remote_unmonitored` (server mode, server/hub/) fires when a user's machine
 * uses the company account for a session it never registered with the hub.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

const { db, stmts } = require("../db");
const { broadcast } = require("../websocket");

const RULE_TYPES = [
  "event_pattern",
  "inactivity",
  "status_duration",
  "token_threshold",
  "cost_threshold",
  "budget",
  "remote_unmonitored",
];
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 };
    }
    case "remote_unmonitored":
      return { ok: true, config: {} };
    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<object>, periods?: Array<object>}}
 */
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;
}

/**
 * A remote client's company-account session that its device never registered
 * (server/hub/remote-sessions.js, after a grace period): every enabled
 * `remote_unmonitored` rule fires once for it (cooldown per session).
 */
function fireRemoteUnmonitored({ sessionId, username, device, deviceId, userId }) {
  for (const rule of loadEnabledRules()) {
    if (rule.rule_type !== "remote_unmonitored") continue;
    fireAlert(rule, {
      sessionId,
      message: `Company account used outside AgentHub: ${username} on ${device} (session ${String(sessionId).slice(0, 8)})`,
      details: { device_id: deviceId, device, user_id: userId, username },
    });
  }
}

// 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,
  fireRemoteUnmonitored,
};