AgentHub/server/lib/scoped-stats.js

/**
 * @file scoped-stats.js
 * @description Source-scoped variants of the dashboard's aggregate queries
 * (stats + analytics). When the user restricts the "data scope" to a subset of
 * machines (see server/lib/source-filter.js), the routes call these instead of
 * the cached prepared statements in db.js so EVERY headline number — session /
 * agent / event counts, token totals, cost, daily charts, tool + type
 * distributions — reflects only the chosen sources.
 *
 * These build SQL dynamically (per request) and are used ONLY on the filtered
 * path; the unfiltered default keeps using db.js's prepared statements, so the
 * common zero-config case pays nothing for this feature.
 *
 * Every function takes a `scope` = `{where, params}`, a predicate over
 * `sessions` with unqualified columns. It is applied either directly (queries
 * over `sessions`) or as a `session_id IN (SELECT id FROM sessions WHERE …)`
 * subquery (queries over events / agents / token_usage), always via bound
 * parameters.
 *
 * The analytics ones also take an optional `range` (`parseRange` in
 * source-filter.js): sessions and agents count by `started_at`, events by
 * `created_at`. The daily charts keep their own 365-day window.
 *
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

/** Subquery restricting a `session_id` column to the scope's sessions. */
function sessionSubquery(scope) {
  return `SELECT id FROM sessions WHERE ${scope.where}`;
}

/** ` AND <col> >= ? AND <col> < ?` for a range (`col` is always a constant). */
function within(col, range) {
  const sql = [];
  const params = [];
  if (range?.start) {
    sql.push(` AND ${col} >= ?`);
    params.push(range.start);
  }
  if (range?.end) {
    sql.push(` AND ${col} < ?`);
    params.push(range.end);
  }
  return { sql: sql.join(""), params };
}

function statsOverview(db, scope, range) {
  const sq = sessionSubquery(scope);
  const p = scope.params;
  const s = within("started_at", range);
  const e = within("created_at", range);
  return db
    .prepare(
      `SELECT
        (SELECT COUNT(*) FROM sessions WHERE ${scope.where}${s.sql}) as total_sessions,
        (SELECT COUNT(*) FROM sessions WHERE status = 'active' AND ${scope.where}${s.sql}) as active_sessions,
        (SELECT COUNT(*) FROM agents WHERE status IN ('working','waiting') AND session_id IN (${sq})${s.sql}) as active_agents,
        (SELECT COUNT(*) FROM agents WHERE session_id IN (${sq})${s.sql}) as total_agents,
        (SELECT COUNT(*) FROM events WHERE session_id IN (${sq})${e.sql}) as total_events`
    )
    .get(
      ...p,
      ...s.params,
      ...p,
      ...s.params,
      ...p,
      ...s.params,
      ...p,
      ...s.params,
      ...p,
      ...e.params
    );
}

function agentStatusCounts(db, scope, range) {
  const s = within("started_at", range);
  return db
    .prepare(
      `SELECT status, COUNT(*) as count FROM agents WHERE session_id IN (${sessionSubquery(
        scope
      )})${s.sql} GROUP BY status`
    )
    .all(...scope.params, ...s.params);
}

function sessionStatusCounts(db, scope, range) {
  const s = within("started_at", range);
  return db
    .prepare(
      `SELECT status, COUNT(*) as count FROM sessions WHERE ${scope.where}${s.sql} GROUP BY status`
    )
    .all(...scope.params, ...s.params);
}

function countEventsToday(db, scope, toLocal, toUTC) {
  return db
    .prepare(
      `SELECT COUNT(*) as count FROM events
       WHERE created_at >= datetime('now', ?, 'start of day', ?)
         AND session_id IN (${sessionSubquery(scope)})`
    )
    .get(toLocal, toUTC, ...scope.params);
}

function tokenTotals(db, scope) {
  return db
    .prepare(
      `SELECT
        COALESCE(SUM(input_tokens + baseline_input), 0) as total_input,
        COALESCE(SUM(output_tokens + baseline_output), 0) as total_output,
        COALESCE(SUM(cache_read_tokens + baseline_cache_read), 0) as total_cache_read,
        COALESCE(SUM(cache_write_tokens + baseline_cache_write), 0) as total_cache_write,
        COALESCE(SUM(cache_write_1h_tokens + baseline_cache_write_1h), 0) as total_cache_write_1h,
        COALESCE(SUM(web_search_requests + baseline_web_search), 0) as total_web_search,
        COALESCE(SUM(web_fetch_requests + baseline_web_fetch), 0) as total_web_fetch,
        COALESCE(SUM(code_execution_requests + baseline_code_execution), 0) as total_code_execution
       FROM token_usage WHERE session_id IN (${sessionSubquery(scope)})`
    )
    .get(...scope.params);
}

function toolUsageCounts(db, scope, range) {
  const e = within("created_at", range);
  return db
    .prepare(
      `SELECT tool_name, COUNT(*) as count FROM events
       WHERE tool_name IS NOT NULL AND session_id IN (${sessionSubquery(scope)})${e.sql}
       GROUP BY tool_name ORDER BY count DESC LIMIT 20`
    )
    .all(...scope.params, ...e.params);
}

function dailyEventCounts(db, scope, tzModifier) {
  return db
    .prepare(
      `SELECT DATE(created_at, ?) as date, COUNT(*) as count FROM events
       WHERE created_at >= DATE('now', '-365 days') AND session_id IN (${sessionSubquery(scope)})
       GROUP BY 1 ORDER BY date ASC`
    )
    .all(tzModifier, ...scope.params);
}

function dailySessionCounts(db, scope, tzModifier) {
  return db
    .prepare(
      `SELECT DATE(started_at, ?) as date, COUNT(*) as count FROM sessions
       WHERE started_at >= DATE('now', '-365 days') AND ${scope.where}
       GROUP BY 1 ORDER BY date ASC`
    )
    .all(tzModifier, ...scope.params);
}

function agentTypeDistribution(db, scope, range) {
  const s = within("started_at", range);
  return db
    .prepare(
      `SELECT subagent_type, COUNT(*) as count FROM agents
       WHERE type = 'subagent' AND subagent_type IS NOT NULL AND session_id IN (${sessionSubquery(
         scope
       )})${s.sql}
       GROUP BY subagent_type ORDER BY count DESC`
    )
    .all(...scope.params, ...s.params);
}

function totalSubagentCount(db, scope, range) {
  const s = within("started_at", range);
  return db
    .prepare(
      `SELECT COUNT(*) as count FROM agents WHERE type = 'subagent' AND session_id IN (${sessionSubquery(
        scope
      )})${s.sql}`
    )
    .get(...scope.params, ...s.params);
}

function eventTypeCounts(db, scope, range) {
  const e = within("created_at", range);
  return db
    .prepare(
      `SELECT event_type, COUNT(*) as count FROM events
       WHERE session_id IN (${sessionSubquery(scope)})${e.sql}
       GROUP BY event_type ORDER BY count DESC`
    )
    .all(...scope.params, ...e.params);
}

function avgEventsPerSession(db, scope, range) {
  const s = within("started_at", range);
  const e = within("created_at", range);
  return db
    .prepare(
      `SELECT ROUND(CAST(COUNT(*) AS REAL) /
         MAX(1, (SELECT COUNT(*) FROM sessions WHERE ${scope.where}${s.sql})), 1) as avg
       FROM events WHERE session_id IN (${sessionSubquery(scope)})${e.sql}`
    )
    .get(...scope.params, ...s.params, ...scope.params, ...e.params);
}

/**
 * The analytics numbers that scan `events` (tool usage, daily events, event
 * types, events per session) for a scope or none. Takes the connection as an
 * argument so the read worker can run it (`analyticsEvents` in read-pool.js).
 */
function eventAggregates(db, scope, range, tzModifier) {
  if (scope) {
    return {
      toolUsage: toolUsageCounts(db, scope, range),
      dailyEvents: dailyEventCounts(db, scope, tzModifier),
      eventTypes: eventTypeCounts(db, scope, range),
      avgEvents: avgEventsPerSession(db, scope, range),
    };
  }
  return {
    toolUsage: db
      .prepare(
        `SELECT tool_name, COUNT(*) as count FROM events WHERE tool_name IS NOT NULL
         GROUP BY tool_name ORDER BY count DESC LIMIT 20`
      )
      .all(),
    dailyEvents: db
      .prepare(
        `SELECT DATE(created_at, ?) as date, COUNT(*) as count FROM events
         WHERE created_at >= DATE('now', '-365 days') GROUP BY 1 ORDER BY date ASC`
      )
      .all(tzModifier),
    eventTypes: db
      .prepare(
        "SELECT event_type, COUNT(*) as count FROM events GROUP BY event_type ORDER BY count DESC"
      )
      .all(),
    avgEvents: db
      .prepare(
        `SELECT ROUND(CAST(COUNT(*) AS REAL) / MAX(1, (SELECT COUNT(*) FROM sessions)), 1) as avg
         FROM events`
      )
      .get(),
  };
}

module.exports = {
  eventAggregates,
  statsOverview,
  agentStatusCounts,
  sessionStatusCounts,
  countEventsToday,
  tokenTotals,
  toolUsageCounts,
  dailyEventCounts,
  dailySessionCounts,
  agentTypeDistribution,
  totalSubagentCount,
  eventTypeCounts,
  avgEventsPerSession,
};