will/agentmon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(codex): recover session lifecycle from hooks

Browse Source
This commit is contained in:
William Valentin
2026-04-21 13:02:58 -07:00
parent 8b6ce8e628
commit d5154b8eec
4 changed files with 526 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hooks/codex/handler.ts
+209 -28
View File
@@ -1,6 +1,8 @@
#!/usr/bin/env node
import { randomUUID } from 'node:crypto';
import { hostname } from 'node:os';
import { mkdirSync, readFileSync, unlinkSync, writeFileSync } from 'node:fs';
import { homedir, hostname } from 'node:os';
import { join } from 'node:path';
import {
type Dict,
isRecord,
@@ -18,28 +20,108 @@ const HOST = process.env.AGENTMON_HOST || hostname();
const { enqueue, flush } = createTransport(INGEST_URL);
const activeRuns = new Map<string, string>();
interface SessionState {
sessionStarted?: boolean;
runId?: string;
}
const STATE_DIR = join(homedir(), '.agentmon-state', 'codex');
function ensureStateDir() {
try {
mkdirSync(STATE_DIR, { recursive: true });
} catch {
// Ignore state directory creation failures so hooks stay best-effort.
}
}
function loadState(sessionKey: string): SessionState {
try {
const raw = readFileSync(join(STATE_DIR, sessionKey + '.json'), 'utf8');
return JSON.parse(raw) as SessionState;
} catch {
return {};
}
}
function saveState(sessionKey: string, state: SessionState) {
ensureStateDir();
try {
writeFileSync(join(STATE_DIR, sessionKey + '.json'), JSON.stringify(state), 'utf8');
} catch {
// Ignore state write failures so event emission can continue.
}
}
function clearState(sessionKey: string) {
try {
unlinkSync(join(STATE_DIR, sessionKey + '.json'));
} catch {
// Ignore state cleanup failures.
}
}
function getContext(input: Dict): Dict {
return isRecord(input.context) ? input.context : {};
}
function getSessionRecord(input: Dict): Dict {
return isRecord(input.session) ? input.session : {};
}
function getConversationRecord(input: Dict): Dict {
return isRecord(input.conversation) ? input.conversation : {};
}
function getSessionKey(input: Dict): string | undefined {
const context = getContext(input);
const session = getSessionRecord(input);
const conversation = getConversationRecord(input);
return pickString(
input.id,
input.session,
input.sessionId,
input.session_id,
input.sessionID,
input.threadId,
input.thread_id,
input.threadID,
input.chatId,
input.chat_id,
input.conversationId,
input.conversation_id,
context.id,
context.session,
context.sessionKey,
context.sessionId,
context.session_id,
context.threadId,
context.thread_id,
context.conversationId,
context.conversation_id,
session.id,
session.key,
session.sessionId,
session.session_id,
conversation.id,
conversation.sessionId,
conversation.session_id,
conversation.conversationId,
conversation.conversation_id,
);
}
function getUsage(input: Dict): Dict | undefined {
const context = getContext(input);
const usage = isRecord(input.usage) ? input.usage :
isRecord(input.llm) ? input.llm :
isRecord(input.tokens) ? input.tokens :
isRecord(input.llm_usage) ? input.llm_usage : undefined;
isRecord(input.llm_usage) ? input.llm_usage :
isRecord(context.usage) ? context.usage :
isRecord(context.llm) ? context.llm :
isRecord(context.tokens) ? context.tokens :
isRecord(context.llm_usage) ? context.llm_usage : undefined;
if (!usage) return undefined;
const result: Dict = {};
@@ -58,13 +140,79 @@ function getUsage(input: Dict): Dict | undefined {
}
function getModel(input: Dict): string | undefined {
const context = getContext(input);
return pickString(
input.model,
isRecord(input.llm) ? input.llm.model : undefined,
isRecord(input.usage) ? input.usage.model : undefined,
context.model,
isRecord(context.llm) ? context.llm.model : undefined,
isRecord(context.usage) ? context.usage.model : undefined,
);
}
function getPrompt(input: Dict): string | undefined {
const context = getContext(input);
return pickString(
input.prompt,
input.text,
input.message,
context.prompt,
context.text,
context.message,
);
}
function getDuration(input: Dict): number | undefined {
const context = getContext(input);
return pickNumber(
input.duration_ms,
input.elapsed_ms,
input.duration,
context.duration_ms,
context.elapsed_ms,
context.duration,
);
}
function getNotificationType(input: Dict): string | undefined {
const context = getContext(input);
return pickString(
input.type,
input.notification_type,
input.event,
input.event_type,
context.type,
context.notification_type,
context.event,
context.event_type,
);
}
function saveSessionState(sessionKey: string | undefined, state: SessionState) {
if (!sessionKey) {
return;
}
saveState(sessionKey, state);
}
function ensureSessionStarted(sessionKey: string | undefined, input: Dict, state: SessionState): SessionState {
if (!sessionKey || state.sessionStarted) {
return state;
}
enqueue(buildEnvelope(FRAMEWORK, HOST, 'session.start', sessionKey, {
attributes: {
synthetic: true,
recovered_from: 'codex-hook',
},
}));
state.sessionStarted = true;
saveSessionState(sessionKey, state);
return state;
}
function enqueueMetricSnapshot(sessionKey: string | undefined, runId: string | undefined, usage: Dict | undefined, input: Dict) {
if (!usage) {
return;
@@ -82,26 +230,41 @@ function enqueueMetricSnapshot(sessionKey: string | undefined, runId: string | u
}));
}
function startRun(sessionKey: string | undefined, input: Dict): string {
function startRun(sessionKey: string | undefined, input: Dict, state: SessionState, synthetic = false): string {
ensureSessionStarted(sessionKey, input, state);
const runId = randomUUID();
if (sessionKey) {
activeRuns.set(sessionKey, runId);
state.runId = runId;
saveSessionState(sessionKey, state);
}
enqueue(buildEnvelope(FRAMEWORK, HOST, 'run.start', sessionKey, {
runId,
attributes: {
trigger: pickString(input.trigger_type, input.trigger, input.event, input.event_type),
...(synthetic && { synthetic: true }),
},
payload: {
prompt_preview: truncate(pickString(input.prompt, input.message, input.text), 200),
prompt_preview: truncate(getPrompt(input), 200),
},
}));
return runId;
}
function endRun(sessionKey: string | undefined, runId: string | undefined, input: Dict, duration?: number) {
function ensureRunStarted(sessionKey: string | undefined, input: Dict, state: SessionState): string | undefined {
if (!sessionKey) {
return undefined;
}
if (state.runId) {
return state.runId;
}
return startRun(sessionKey, input, state, true);
}
function endRun(sessionKey: string | undefined, state: SessionState, input: Dict, duration?: number) {
const runId = state.runId;
if (!runId) {
return;
}
@@ -117,24 +280,35 @@ function endRun(sessionKey: string | undefined, runId: string | undefined, input
},
}));
enqueueMetricSnapshot(sessionKey, runId, usage, input);
state.runId = undefined;
saveSessionState(sessionKey, state);
}
async function handleSessionStart(input: Dict) {
const sessionKey = getSessionKey(input) || randomUUID();
const state = loadState(sessionKey);
enqueue(buildEnvelope(FRAMEWORK, HOST, 'session.start', sessionKey));
startRun(sessionKey, input);
if (!state.sessionStarted) {
enqueue(buildEnvelope(FRAMEWORK, HOST, 'session.start', sessionKey));
state.sessionStarted = true;
}
startRun(sessionKey, input, state);
await flush();
}
async function handleSessionEnd(input: Dict) {
const sessionKey = getSessionKey(input);
const runId = sessionKey ? activeRuns.get(sessionKey) : undefined;
const state = sessionKey ? loadState(sessionKey) : {};
const usage = getUsage(input);
const duration = pickNumber(input.duration_ms, input.elapsed_ms, input.duration);
const duration = getDuration(input);
endRun(sessionKey, runId, input, duration);
ensureSessionStarted(sessionKey, input, state);
if (!state.runId && sessionKey) {
ensureRunStarted(sessionKey, input, state);
}
endRun(sessionKey, state, input, duration);
enqueue(buildEnvelope(FRAMEWORK, HOST, 'session.end', sessionKey, {
payload: {
@@ -143,41 +317,48 @@ async function handleSessionEnd(input: Dict) {
},
}));
enqueueMetricSnapshot(sessionKey, runId, usage, input);
activeRuns.delete(sessionKey || '');
if (sessionKey) {
clearState(sessionKey);
}
await flush();
}
async function handlePromptSubmit(input: Dict) {
const sessionKey = getSessionKey(input);
const runId = sessionKey ? activeRuns.get(sessionKey) : undefined;
const duration = pickNumber(input.elapsed_ms, input.duration_ms, input.duration);
const prompt = pickString(input.prompt, input.text, input.message);
const state = sessionKey ? loadState(sessionKey) : {};
const duration = getDuration(input);
const prompt = getPrompt(input);
if (runId && prompt) {
endRun(sessionKey, runId, input, duration);
ensureSessionStarted(sessionKey, input, state);
if (state.runId && prompt) {
endRun(sessionKey, state, input, duration);
}
startRun(sessionKey, input);
startRun(sessionKey, input, state, !state.sessionStarted);
await flush();
}
async function handleNotification(input: Dict) {
const sessionKey = getSessionKey(input);
const notificationType = pickString(input.type, input.notification_type, input.event, input.event_type);
const state = sessionKey ? loadState(sessionKey) : {};
const notificationType = getNotificationType(input);
const usage = getUsage(input);
const duration = pickNumber(input.duration_ms, input.elapsed_ms);
const duration = getDuration(input);
ensureSessionStarted(sessionKey, input, state);
if (notificationType === 'agent-turn-complete' || notificationType === 'Done' || notificationType === 'turn.complete') {
const runId = sessionKey ? activeRuns.get(sessionKey) : undefined;
endRun(sessionKey, runId, input, duration);
if (!state.runId && sessionKey) {
ensureRunStarted(sessionKey, input, state);
}
endRun(sessionKey, state, input, duration);
if (pickString(input.prompt, input.message, input.text)) {
startRun(sessionKey, input);
if (getPrompt(input)) {
startRun(sessionKey, input, state);
}
} else if (usage) {
enqueueMetricSnapshot(sessionKey, sessionKey ? activeRuns.get(sessionKey) : undefined, usage, input);
enqueueMetricSnapshot(sessionKey, state.runId, usage, input);
}
await flush();