agentmon/hooks/codex/handler.js

#!/usr/bin/env node
import { randomUUID } from "node:crypto";
import { hostname } from "node:os";
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 = 2e3;
const FETCH_TIMEOUT_MS = 500;
let buffer = [];
let flushTimer = null;
let isFlushing = false;
const activeRuns = /* @__PURE__ */ new Map();
function isRecord(value) {
  return value !== null && typeof value === "object" && !Array.isArray(value);
}
function pickString(...values) {
  for (const value of values) {
    if (typeof value === "string" && value.trim() !== "") {
      return value;
    }
  }
  return void 0;
}
function pickNumber(...values) {
  for (const value of values) {
    if (typeof value === "number" && Number.isFinite(value)) {
      return value;
    }
  }
  return void 0;
}
function truncate(value, limit) {
  if (value === void 0 || value === null) {
    return void 0;
  }
  const text = typeof value === "string" ? value : safeJSONStringify(value);
  if (!text) {
    return void 0;
  }
  if (text.length <= limit) {
    return text;
  }
  return text.slice(0, limit) + "...";
}
function safeJSONStringify(value) {
  try {
    return JSON.stringify(value);
  } catch {
    return String(value);
  }
}
function getSessionKey(input) {
  return pickString(
    input.sessionId,
    input.session_id,
    input.conversationId,
    input.conversation_id
  );
}
function getUsage(input) {
  const usage = isRecord(input.usage) ? input.usage : isRecord(input.llm) ? input.llm : isRecord(input.tokens) ? input.tokens : void 0;
  if (!usage)
    return void 0;
  const result = {};
  if (usage.input !== void 0)
    result.input_tokens = usage.input;
  if (usage.output !== void 0)
    result.output_tokens = usage.output;
  if (usage.total !== void 0)
    result.total_tokens = usage.total;
  if (usage.cost !== void 0)
    result.total_cost = usage.cost;
  return Object.keys(result).length > 0 ? result : void 0;
}
function buildEnvelope(type, sessionKey, opts = {}) {
  const correlation = {};
  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 = {
    schema: { name: "agentmon.event", version: 1 },
    event: {
      id: randomUUID(),
      type,
      ts: (/* @__PURE__ */ 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) {
  buffer.push(event);
  if (buffer.length >= BATCH_SIZE) {
    void flush();
  } else {
    scheduleFlush();
  }
}
async function postBatch(batch) {
  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) {
  const sessionKey = getSessionKey(input) || randomUUID();
  const runId = randomUUID();
  activeRuns.set(sessionKey, runId);
  enqueue(buildEnvelope("session.start", sessionKey));
  enqueue(buildEnvelope("run.start", sessionKey, {
    runId,
    attributes: {
      trigger: pickString(input.trigger_type, input.trigger)
    },
    payload: {
      prompt_preview: truncate(input.prompt, 200)
    }
  }));
  await flush();
}
async function handleSessionEnd(input) {
  const sessionKey = getSessionKey(input);
  const runId = sessionKey ? activeRuns.get(sessionKey) : void 0;
  const usage = getUsage(input);
  const duration = pickNumber(input.duration_ms, input.elapsed_ms, input.duration);
  if (runId) {
    enqueue(buildEnvelope("run.end", sessionKey, {
      runId,
      payload: {
        status: "success",
        duration_ms: duration,
        ...usage && { usage }
      }
    }));
  }
  enqueue(buildEnvelope("session.end", sessionKey, {
    payload: usage ? { usage } : void 0
  }));
  activeRuns.delete(sessionKey || "");
  await flush();
}
async function handleNotification(input) {
  const sessionKey = getSessionKey(input);
  const notificationType = pickString(input.type, input.notification_type);
  const usage = getUsage(input);
  const duration = pickNumber(input.duration_ms, input.elapsed_ms);
  if (notificationType === "agent-turn-complete" || notificationType === "Done") {
    const runId = sessionKey ? activeRuns.get(sessionKey) : void 0;
    if (runId) {
      enqueue(buildEnvelope("run.end", sessionKey, {
        runId,
        payload: {
          status: "success",
          duration_ms: duration,
          ...usage && { usage }
        }
      }));
    }
    const newRunId = randomUUID();
    if (sessionKey) {
      activeRuns.set(sessionKey, newRunId);
    }
    enqueue(buildEnvelope("run.start", sessionKey, {
      runId: newRunId
    }));
  }
  await flush();
}
const handler = async () => {
  const args = process.argv.slice(2);
  const hookType = args[0] || "unknown";
  let input = {};
  try {
    const stdin = await readStdin();
    if (stdin) {
      input = JSON.parse(stdin);
    }
  } catch {
    input = {};
  }
  try {
    switch (hookType) {
      case "start":
        await handleSessionStart(input);
        break;
      case "stop":
        await handleSessionEnd(input);
        break;
      case "notification":
        await handleNotification(input);
        break;
      default:
        console.debug(`[agentmon] unknown hook type: ${hookType}`);
    }
  } catch (err) {
    console.debug("[agentmon] handler error:", err);
  }
};
async function readStdin() {
  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();