`agents/shared/pulse-reporter.js`

139 lines
/**
 * Norma Pulse Reporter
 *
 * Utility to push data to the Pulse agent (port 9845).
 * All Norma agents report their activity and status here.
 */

const http = require('http');

const PULSE_HOST = '127.0.0.1';
const PULSE_PORT = 9845;
const AUTH_HEADER = 'Basic ' + Buffer.from(`${process.env.AUTH_USERNAME || 'admin'}:${process.env.AUTH_PASSWORD || ''}`).toString('base64');

/**
 * Make an HTTP POST request to the Pulse agent.
 *
 * @param {string} path - URL path (e.g. '/api/pulse/social-feed')
 * @param {Object} data - JSON payload
 * @returns {Promise<Object>} Parsed JSON response
 */
function _post(path, data) {
  return new Promise((resolve, reject) => {
    const payload = JSON.stringify(data);

    const options = {
      hostname: PULSE_HOST,
      port: PULSE_PORT,
      path: path,
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Content-Length': Buffer.byteLength(payload),
        'Authorization': AUTH_HEADER,
      },
      timeout: 10000,
    };

    const req = http.request(options, (res) => {
      let body = '';
      res.on('data', (chunk) => { body += chunk; });
      res.on('end', () => {
        try {
          const parsed = JSON.parse(body);
          resolve(parsed);
        } catch (e) {
          resolve({ raw: body, statusCode: res.statusCode });
        }
      });
    });

    req.on('error', (err) => {
      console.error(`[${new Date().toISOString()}] [pulse-reporter] Request error for ${path}:`, err.message);
      reject(err);
    });

    req.on('timeout', () => {
      req.destroy();
      reject(new Error(`Pulse request to ${path} timed out after 10s`));
    });

    req.write(payload);
    req.end();
  });
}

/**
 * Push social feed data to Pulse.
 * Used when an agent discovers new relevant posts, petitions, or discussions.
 *
 * @param {Object} data - Social feed data
 * @param {string} data.agent - Agent name
 * @param {string} data.platform - Platform name
 * @param {string} data.type - Content type ('post', 'petition', 'discussion', 'mention')
 * @param {string} data.title - Content title
 * @param {string} [data.url] - URL to the content
 * @param {string} [data.content] - Content body/snippet
 * @param {Object} [data.metadata] - Additional metadata
 * @returns {Promise<Object>}
 */
async function pushSocialFeed(data) {
  try {
    return await _post('/api/pulse/social-feed', data);
  } catch (err) {
    console.error(`[${new Date().toISOString()}] [pulse-reporter] Failed to push social feed:`, err.message);
    return null;
  }
}

/**
 * Push trending topics data to Pulse.
 * Used when an agent detects trending conversations or hashtags.
 *
 * @param {Array<Object>} topics - Array of trending topic objects
 * @param {string} topics[].topic - Topic name or hashtag
 * @param {string} topics[].platform - Platform where it's trending
 * @param {number} [topics[].volume] - Mention count or engagement score
 * @param {string} [topics[].sentiment] - Overall sentiment ('positive', 'negative', 'neutral')
 * @returns {Promise<Object>}
 */
async function pushTrending(topics) {
  try {
    return await _post('/api/pulse/trending', {
      topics: Array.isArray(topics) ? topics : [topics],
      reportedAt: new Date().toISOString(),
    });
  } catch (err) {
    console.error(`[${new Date().toISOString()}] [pulse-reporter] Failed to push trending:`, err.message);
    return null;
  }
}

/**
 * Report agent health/status to Pulse.
 * Called periodically by each agent to confirm it's alive and working.
 *
 * @param {string} agent - Agent name
 * @param {Object} status - Status data
 * @param {string} status.state - Agent state ('running', 'idle', 'error', 'rate_limited')
 * @param {number} [status.uptime] - Uptime in seconds
 * @param {Object} [status.rateLimits] - Current rate limit status
 * @param {Object} [status.lastAction] - Last action taken
 * @param {string} [status.errorMessage] - Current error, if any
 * @returns {Promise<Object>}
 */
async function reportAgentStatus(agent, status) {
  try {
    return await _post('/api/pulse/agent-status', {
      agent,
      ...status,
      reportedAt: new Date().toISOString(),
    });
  } catch (err) {
    console.error(`[${new Date().toISOString()}] [pulse-reporter] Failed to report status for ${agent}:`, err.message);
    return null;
  }
}

module.exports = { pushSocialFeed, pushTrending, reportAgentStatus };