import { randomUUID } from 'node:crypto'; import { hostname } from 'node:os'; type Dict = Record; 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 | null = null; let isFlushing = false; const activeRuns = new Map(); const activeCompactions = new Map(); 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); activeCompactions.delete(sessionKey); } return; } if (eventName === 'command:reset') { enqueue(buildEnvelope('session.end', sessionKey)); enqueue(buildEnvelope('session.start', sessionKey)); if (sessionKey) { activeRuns.delete(sessionKey); activeCompactions.delete(sessionKey); } 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; } if (eventName === 'tool_result_persist') { const runId = sessionKey ? activeRuns.get(sessionKey) : undefined; const spanId = randomUUID(); const success = context.success !== false && !context.error; const toolName = pickString(context.toolName, context.tool_name, context.name) || 'unknown_tool'; 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 resultPreview = truncate(context.result ?? context.output, 500); if (resultPreview) { payload.result_preview = resultPreview; } enqueue(buildEnvelope('span.end', sessionKey, { runId, spanId, attributes: { span_kind: 'tool', name: toolName, }, payload, })); if (!success) { emitError(sessionKey, runId, spanId, context.error); } return; } if (eventName === 'session:compact:before') { const runId = sessionKey ? activeRuns.get(sessionKey) : undefined; const spanId = randomUUID(); if (sessionKey) { activeCompactions.set(sessionKey, spanId); } enqueue(buildEnvelope('span.start', sessionKey, { runId, spanId, attributes: { span_kind: 'internal', name: 'context_compaction', }, })); return; } if (eventName === 'session:compact:after') { const runId = sessionKey ? activeRuns.get(sessionKey) : undefined; const spanId = (sessionKey && activeCompactions.get(sessionKey)) || randomUUID(); if (sessionKey) { activeCompactions.delete(sessionKey); } enqueue(buildEnvelope('span.end', sessionKey, { runId, spanId, attributes: { span_kind: 'internal', name: 'context_compaction', }, payload: { status: 'success', duration_ms: pickNumber(context.duration_ms, context.durationMs, context.elapsed_ms), }, })); } } catch { console.debug('[agentmon] handler error'); } }; export default handler;