`agents/onboard-agent/skills/monitor.js`

173 lines
/**
 * OnBoard Monitor Skill
 *
 * Watches pulse_topics for new high-urgency topics (last 30 minutes).
 * If a new topic doesn't already have discoveries, triggers a mini
 * discovery run for that topic.
 */

const { query } = require('../../shared/db');
const { logAction } = require('../../shared/audit-logger');
const { searchNonprofits } = require('../lib/propublica');
const { getIssueAreas, issueOverlapScore } = require('../lib/ntee-matcher');

const AGENT = 'onboard-agent';
const PLATFORM = 'pulse';

/**
 * Extract search keywords from a topic name + description.
 */
function extractKeywords(topic) {
  const text = `${topic.name || ''} ${topic.description || ''}`.toLowerCase();
  const stopWords = new Set(['the', 'a', 'an', 'is', 'are', 'was', 'were', 'be', 'been',
    'being', 'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could',
    'should', 'may', 'might', 'can', 'shall', 'to', 'of', 'in', 'for', 'on', 'with',
    'at', 'by', 'from', 'as', 'into', 'through', 'during', 'before', 'after',
    'and', 'but', 'or', 'nor', 'not', 'so', 'yet', 'both', 'either', 'neither',
    'this', 'that', 'these', 'those', 'it', 'its', 'they', 'them', 'their',
    'about', 'up', 'out', 'than', 'also', 'just', 'more', 'very']);

  const words = text.replace(/[^a-z0-9\s]/g, ' ').split(/\s+/).filter(w => w.length > 2 && !stopWords.has(w));
  return [...new Set(words)].slice(0, 5);
}

/**
 * @param {Object} body - Request body
 * @param {number} [body.minutes_back=30] - How many minutes back to check for new topics
 * @returns {Promise<Object>}
 */
module.exports = async function monitor(body = {}) {
  const minutesBack = body.minutes_back || 30;

  // 1. Query new high-urgency topics from the last N minutes
  const { rows: newTopics } = await query(
    `SELECT id, name, category, urgency, description, created_at
     FROM pulse_topics
     WHERE urgency IN ('critical', 'high')
       AND created_at > NOW() - INTERVAL '1 minute' * $1
     ORDER BY created_at DESC`,
    [minutesBack]
  );

  if (newTopics.length === 0) {
    return { new_topics: 0, triggered_discoveries: 0, message: 'No new high-urgency topics' };
  }

  let triggeredDiscoveries = 0;

  for (const topic of newTopics) {
    // 2. Check if we already have discoveries for this topic
    const { rows: existingDisc } = await query(
      `SELECT COUNT(*)::int AS count FROM onboard_discoveries
       WHERE trigger_topic_id = $1`,
      [topic.id]
    );

    if (existingDisc[0].count > 0) {
      continue; // Already have discoveries for this topic
    }

    // 3. Trigger mini discovery run
    const keywords = extractKeywords(topic);
    if (keywords.length === 0) continue;

    const searchQuery = keywords.join(' ');
    const result = await searchNonprofits(searchQuery);
    const orgs = result.organizations || [];

    for (const org of orgs.slice(0, 5)) { // Smaller batch for monitor
      const orgIssues = getIssueAreas(org.ntee_code);
      const issueScore = issueOverlapScore(orgIssues, keywords) * 40;
      const matchScore = Math.round(issueScore * 100) / 100;

      if (matchScore < 5) continue;

      // Insert org if new
      let orgId;
      try {
        if (org.ein) {
          const { rows: existingOrg } = await query(
            `SELECT id FROM organizations WHERE ein = $1 LIMIT 1`,
            [String(org.ein)]
          );
          if (existingOrg.length > 0) {
            orgId = existingOrg[0].id;
          }
        }

        if (!orgId) {
          const { rows: inserted } = await query(
            `INSERT INTO organizations (name, ein, type, ntee_code, city, state,
               annual_revenue, total_assets, source, source_id)
             VALUES ($1, $2, 'nonprofit', $3, $4, $5, $6, $7, 'onboard-agent', $8)
             ON CONFLICT (ein) DO UPDATE SET updated_at = NOW()
             RETURNING id`,
            [
              org.name,
              org.ein ? String(org.ein) : null,
              org.ntee_code || null,
              org.city || null,
              org.state || null,
              org.total_revenue || null,
              org.total_assets || null,
              org.ein ? `propublica-${org.ein}` : null,
            ]
          );
          orgId = inserted[0]?.id;
        }
      } catch (err) {
        console.error(`[${AGENT}] Monitor insert org error:`, err.message);
        continue;
      }

      if (!orgId) continue;

      // Insert discovery
      try {
        await query(
          `INSERT INTO onboard_discoveries
            (org_id, trigger_topic_id, trigger_type, trigger_label, match_score, match_details, status)
           VALUES ($1, $2, 'pulse_monitor', $3, $4, $5, 'discovered')
           ON CONFLICT DO NOTHING`,
          [
            orgId,
            topic.id,
            topic.name,
            matchScore,
            JSON.stringify({
              issue_score: matchScore,
              search_query: searchQuery,
              source: 'monitor_auto',
            }),
          ]
        );
        triggeredDiscoveries++;
      } catch (err) {
        // Ignore duplicate key errors
        if (!err.message.includes('duplicate') && !err.message.includes('unique')) {
          console.error(`[${AGENT}] Monitor insert discovery error:`, err.message);
        }
      }
    }
  }

  // Log the action
  await logAction({
    agent: AGENT,
    actionType: 'monitor',
    platform: PLATFORM,
    content: `Checked ${newTopics.length} new topics (last ${minutesBack}min)`,
    responseData: {
      new_topics: newTopics.length,
      triggered_discoveries: triggeredDiscoveries,
      minutes_back: minutesBack,
    },
    status: 'success',
  });

  return {
    new_topics: newTopics.length,
    triggered_discoveries: triggeredDiscoveries,
  };
};