feat(hooks): consolidate shared transport helpers
This commit is contained in:
William Valentin
+16
-182
@@ -1,68 +1,25 @@
|
||||
#!/usr/bin/env node
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { hostname } from 'node:os';
|
||||
import {
|
||||
type Dict,
|
||||
isRecord,
|
||||
pickString,
|
||||
pickNumber,
|
||||
truncate,
|
||||
buildEnvelope,
|
||||
createTransport,
|
||||
readStdin,
|
||||
} from '../shared/lib';
|
||||
|
||||
const INGEST_URL = process.env.AGENTMON_INGEST_URL || 'http://localhost:8080';
|
||||
const FRAMEWORK = process.env.AGENTMON_FRAMEWORK || 'codex';
|
||||
const HOST = process.env.AGENTMON_HOST || hostname();
|
||||
const BATCH_SIZE = 10;
|
||||
const FLUSH_MS = 2000;
|
||||
const FETCH_TIMEOUT_MS = 500;
|
||||
|
||||
interface Dict { [key: string]: any }
|
||||
|
||||
let buffer: Dict[] = [];
|
||||
let flushTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let isFlushing = false;
|
||||
const { enqueue, flush } = createTransport(INGEST_URL);
|
||||
|
||||
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 getSessionKey(input: Dict): string | undefined {
|
||||
return pickString(
|
||||
input.id,
|
||||
@@ -108,75 +65,6 @@ function getModel(input: Dict): string | 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: FRAMEWORK,
|
||||
client_id: HOST,
|
||||
host: HOST,
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
function enqueueMetricSnapshot(sessionKey: string | undefined, runId: string | undefined, usage: Dict | undefined, input: Dict) {
|
||||
if (!usage) {
|
||||
return;
|
||||
@@ -188,7 +76,7 @@ function enqueueMetricSnapshot(sessionKey: string | undefined, runId: string | u
|
||||
metrics.model = model;
|
||||
}
|
||||
|
||||
enqueue(buildEnvelope('metric.snapshot', sessionKey, {
|
||||
enqueue(buildEnvelope(FRAMEWORK, HOST, 'metric.snapshot', sessionKey, {
|
||||
runId,
|
||||
payload: { metrics },
|
||||
}));
|
||||
@@ -200,7 +88,7 @@ function startRun(sessionKey: string | undefined, input: Dict): string {
|
||||
activeRuns.set(sessionKey, runId);
|
||||
}
|
||||
|
||||
enqueue(buildEnvelope('run.start', sessionKey, {
|
||||
enqueue(buildEnvelope(FRAMEWORK, HOST, 'run.start', sessionKey, {
|
||||
runId,
|
||||
attributes: {
|
||||
trigger: pickString(input.trigger_type, input.trigger, input.event, input.event_type),
|
||||
@@ -219,7 +107,7 @@ function endRun(sessionKey: string | undefined, runId: string | undefined, input
|
||||
}
|
||||
|
||||
const usage = getUsage(input);
|
||||
enqueue(buildEnvelope('run.end', sessionKey, {
|
||||
enqueue(buildEnvelope(FRAMEWORK, HOST, 'run.end', sessionKey, {
|
||||
runId,
|
||||
payload: {
|
||||
status: 'success',
|
||||
@@ -231,54 +119,10 @@ function endRun(sessionKey: string | undefined, runId: string | undefined, input
|
||||
enqueueMetricSnapshot(sessionKey, runId, usage, input);
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function handleSessionStart(input: Dict) {
|
||||
const sessionKey = getSessionKey(input) || randomUUID();
|
||||
|
||||
enqueue(buildEnvelope('session.start', sessionKey));
|
||||
enqueue(buildEnvelope(FRAMEWORK, HOST, 'session.start', sessionKey));
|
||||
startRun(sessionKey, input);
|
||||
|
||||
await flush();
|
||||
@@ -292,7 +136,7 @@ async function handleSessionEnd(input: Dict) {
|
||||
|
||||
endRun(sessionKey, runId, input, duration);
|
||||
|
||||
enqueue(buildEnvelope('session.end', sessionKey, {
|
||||
enqueue(buildEnvelope(FRAMEWORK, HOST, 'session.end', sessionKey, {
|
||||
payload: {
|
||||
model: getModel(input),
|
||||
...(usage && { usage }),
|
||||
@@ -376,14 +220,4 @@ const handler = async () => {
|
||||
}
|
||||
};
|
||||
|
||||
async function readStdin(): Promise<string> {
|
||||
return new Promise((resolve) => {
|
||||
let data = '';
|
||||
process.stdin.on('data', (chunk) => data += chunk);
|
||||
process.stdin.on('end', () => resolve(data));
|
||||
process.stdin.on('error', () => resolve(''));
|
||||
setTimeout(() => resolve(data), 100);
|
||||
});
|
||||
}
|
||||
|
||||
handler();
|
||||
|
||||
Reference in New Issue
Block a user