swarm-master/openclaw/hooks/agentmon/handler.ts

import { randomUUID } from 'node:crypto';
import { hostname } from 'node:os';

type Dict = Record<string, any>;

const INGEST_URL = process.env.AGENTMON_INGEST_URL || 'http://192.168.122.1:8080';
const VM_NAME = process.env.AGENTMON_VM_NAME || hostname();
const BATCH_SIZE = 10;
const FLUSH_MS = 2000;
const FETCH_TIMEOUT_MS = 500;

let buffer: Dict[] = [];
let flushTimer: ReturnType<typeof setTimeout> | null = null;
let isFlushing = false;

const activeRuns = new Map<string, string>();

function isRecord(value: unknown): value is Dict {
  return value !== null && typeof value === 'object' && !Array.isArray(value);
}

function pickString(...values: unknown[]): string | undefined {
  for (const value of values) {
    if (typeof value === 'string' && value.trim() !== '') {
      return value;
    }
  }
  return undefined;
}

function pickNumber(...values: unknown[]): number | undefined {
  for (const value of values) {
    if (typeof value === 'number' && Number.isFinite(value)) {
      return value;
    }
  }
  return undefined;
}

function truncate(value: unknown, limit: number): string | undefined {
  if (value === undefined || value === null) {
    return undefined;
  }

  const text = typeof value === 'string' ? value : safeJSONStringify(value);
  if (!text) {
    return undefined;
  }

  if (text.length <= limit) {
    return text;
  }
  return text.slice(0, limit) + '...';
}

function safeJSONStringify(value: unknown): string {
  try {
    return JSON.stringify(value);
  } catch {
    return String(value);
  }
}

function getEventName(input: Dict): string {
  const direct = pickString(input.name, input.event);
  if (direct) {
    return direct;
  }

  if (typeof input.type === 'string' && input.type.includes(':')) {
    return input.type;
  }

  if (typeof input.type === 'string' && typeof input.action === 'string') {
    return `${input.type}:${input.action}`;
  }

  if (typeof input.type === 'string') {
    return input.type;
  }

  return '';
}

function getContext(input: Dict): Dict {
  return isRecord(input.context) ? input.context : {};
}

function getSessionKey(input: Dict, context: Dict): string | undefined {
  return pickString(
    input.sessionKey,
    context.sessionKey,
    context.session_id,
    input.session_id,
    isRecord(input.session) ? input.session.key : undefined,
    isRecord(context.session) ? context.session.key : undefined,
  );
}

function buildEnvelope(
  type: string,
  sessionKey?: string,
  opts: {
    runId?: string;
    spanId?: string;
    parentSpanId?: string;
    attributes?: Dict;
    payload?: Dict;
  } = {},
): Dict {
  const correlation: Dict = {};
  if (sessionKey) {
    correlation.session_id = sessionKey;
  }
  if (opts.runId) {
    correlation.run_id = opts.runId;
  }
  if (opts.spanId) {
    correlation.span_id = opts.spanId;
  }
  if (opts.parentSpanId) {
    correlation.parent_span_id = opts.parentSpanId;
  }

  const envelope: Dict = {
    schema: { name: 'agentmon.event', version: 1 },
    event: {
      id: randomUUID(),
      type,
      ts: new Date().toISOString(),
      source: {
        framework: 'openclaw',
        client_id: VM_NAME,
        host: VM_NAME,
      },
    },
  };

  if (Object.keys(correlation).length > 0) {
    envelope.correlation = correlation;
  }
  if (opts.attributes && Object.keys(opts.attributes).length > 0) {
    envelope.attributes = opts.attributes;
  }
  if (opts.payload && Object.keys(opts.payload).length > 0) {
    envelope.payload = opts.payload;
  }

  return envelope;
}

function scheduleFlush() {
  if (!flushTimer) {
    flushTimer = setTimeout(() => {
      void flush();
    }, FLUSH_MS);
  }
}

function enqueue(event: Dict) {
  buffer.push(event);
  if (buffer.length >= BATCH_SIZE) {
    void flush();
  } else {
    scheduleFlush();
  }
}

async function postBatch(batch: Dict[]) {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS);

  try {
    await fetch(`${INGEST_URL}/v1/events`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(batch),
      signal: controller.signal,
    });
  } finally {
    clearTimeout(timeout);
  }
}

async function flush() {
  if (flushTimer) {
    clearTimeout(flushTimer);
    flushTimer = null;
  }
  if (isFlushing || buffer.length === 0) {
    return;
  }

  isFlushing = true;
  const batch = buffer.splice(0, BATCH_SIZE);

  try {
    await postBatch(batch);
  } catch {
    console.debug(`[agentmon] failed to flush ${batch.length} events`);
  } finally {
    isFlushing = false;
    if (buffer.length > 0) {
      if (buffer.length >= BATCH_SIZE) {
        void flush();
      } else {
        scheduleFlush();
      }
    }
  }
}

function emitError(sessionKey: string | undefined, runId: string | undefined, spanId: string | undefined, errorValue: unknown) {
  if (errorValue === undefined || errorValue === null || errorValue === false) {
    return;
  }

  const errorRecord = isRecord(errorValue) ? errorValue : {};
  const message = pickString(errorRecord.message, errorRecord.error, errorValue) || 'unknown';
  const errType = pickString(errorRecord.type, errorRecord.code) || 'openclaw';

  enqueue(buildEnvelope('error', sessionKey, {
    runId,
    spanId,
    payload: {
      error: {
        type: errType,
        message,
      },
    },
  }));
}

function buildRunPayload(context: Dict, success: boolean): Dict {
  const payload: Dict = {
    status: success ? 'success' : 'error',
  };

  const duration = pickNumber(context.duration_ms, context.durationMs, context.elapsed_ms);
  if (duration !== undefined) {
    payload.duration_ms = duration;
  }

  const usage = isRecord(context.usage) ? context.usage : undefined;
  if (usage) {
    payload.usage = usage;
  }

  const errorMessage = pickString(context.error, isRecord(context.result) ? context.result.error : undefined);
  if (errorMessage) {
    payload.error = errorMessage;
  }

  return payload;
}

const handler = async (rawEvent: unknown) => {
  if (!isRecord(rawEvent)) {
    return;
  }

  const context = getContext(rawEvent);
  const eventName = getEventName(rawEvent);
  const sessionKey = getSessionKey(rawEvent, context);

  try {
    if (eventName === 'command:new') {
      enqueue(buildEnvelope('session.start', sessionKey));
      return;
    }

    if (eventName === 'command:stop') {
      enqueue(buildEnvelope('session.end', sessionKey));
      if (sessionKey) {
        activeRuns.delete(sessionKey);
      }
      return;
    }

    if (eventName === 'command:reset') {
      enqueue(buildEnvelope('session.end', sessionKey));
      enqueue(buildEnvelope('session.start', sessionKey));
      if (sessionKey) {
        activeRuns.delete(sessionKey);
      }
      return;
    }

    if (eventName === 'agent:bootstrap') {
      // Only emit run.start if no run is already active for this session.
      // Interactive message-channel sessions already emit run.start via
      // message:received; this handler captures cron/automation embedded runs.
      const existingRunId = sessionKey ? activeRuns.get(sessionKey) : undefined;
      if (!existingRunId) {
        const runId = randomUUID();
        if (sessionKey) {
          activeRuns.set(sessionKey, runId);
        }

        enqueue(buildEnvelope('run.start', sessionKey, {
          runId,
          attributes: {
            agent_id: pickString(context.agentId as string | undefined),
            run_kind: 'embedded',
          },
        }));
      }
      return;
    }

    if (eventName === 'message:received') {
      const runId = randomUUID();
      if (sessionKey) {
        activeRuns.set(sessionKey, runId);
      }

      enqueue(buildEnvelope('run.start', sessionKey, {
        runId,
        attributes: {
          channel: pickString(context.channelId, context.channel_id),
          from: pickString(context.from, context.sender),
        },
        payload: {
          message_preview: truncate(
            pickString(context.content, context.message, context.text) || context.input,
            200,
          ),
        },
      }));
      return;
    }

    if (eventName === 'message:sent') {
      const runId = sessionKey ? activeRuns.get(sessionKey) : undefined;
      const success = context.success !== false && !context.error;

      enqueue(buildEnvelope('run.end', sessionKey, {
        runId,
        attributes: {
          channel: pickString(context.channelId, context.channel_id),
          to: pickString(context.to, context.recipient),
        },
        payload: buildRunPayload(context, success),
      }));

      if (!success) {
        emitError(sessionKey, runId, undefined, context.error);
      }
      return;
    }

  } catch {
    console.debug('[agentmon] handler error');
  }
};

export default handler;