toknmtr-selfhost/agent/parse.ts

/**
 * toknmtr agent — JSONL parser.
 *
 * Walks ~/.claude/projects/<encoded-cwd>/<session-id>.jsonl and turns every line
 * into a typed event the server can ingest. NOT bundled by Vite — this runs
 * standalone via the capture hook + cron (e.g. `node --experimental-strip-types
 * agent/run.ts`).
 *
 * Real JSONL line shape (see CLAUDE.md / the shared agent contract):
 *  - Not every physical line is a trackable event: `mode`, `file-history-snapshot`,
 *    `attachment`, `ai-title`, `last-prompt`, etc. carry no top-level `uuid` and are
 *    skipped (parseLine returns null), but they (and every other line) may still carry
 *    `sessionId`/`cwd`/`gitBranch`/`version`/`entrypoint`/`timestamp`, which we scan for
 *    session metadata regardless of whether the line itself becomes an event.
 *  - Assistant turns STREAM: the same `message.id` repeats across several consecutive
 *    physical lines (each with its own unique top-level `uuid`, chained via `parentUuid`),
 *    each carrying a different slice of `message.content` (one block per line in
 *    practice: a `thinking` line, then a `text` line, then one `tool_use` line per tool
 *    call). Every physical line is still stored as its own `events` row.
 *  - `message.content` is either a plain string (typed user prompt) or an array of
 *    blocks: `{type:'text', text}`, `{type:'thinking', thinking}`,
 *    `{type:'tool_use', id, name, input}`, and on USER lines
 *    `{type:'tool_result', tool_use_id, content, is_error}` (content is a string or a
 *    content-block array).
 */
import { homedir } from 'node:os';
import { join } from 'node:path';

export const TRANSCRIPT_ROOT = join(homedir(), '.claude', 'projects');

export interface ParsedEvent {
	host: string;
	session_id: string;
	uuid: string;
	parent_uuid: string | null;
	ts_utc: string;
	type: string;
	role: string | null;
	model: string | null;
	request_id: string | null;
	message_id: string | null;
	is_sidechain: boolean;
	/** True for the one row per (session_id, message_id, request_id) group that should
	 *  count toward token totals — the final/max-output_tokens streamed line. */
	is_usage_canonical: boolean;
	stop_reason: string | null;
	/** ms between this line's timestamp and its parent line's timestamp (assistant lines
	 *  only, when the parent is known within the parsed chunk). */
	latency_ms: number | null;
	input_tokens: number | null;
	output_tokens: number | null;
	cache_creation_tokens: number | null;
	cache_read_tokens: number | null;
	web_search_requests: number | null;
	web_fetch_requests: number | null;
	/** Flattened visible text (joined `text` content blocks, or the raw string content of
	 *  a plain-string user message). Does NOT include `thinking` or `tool_result` content. */
	text: string | null;
}

/** Mirrors the `tool_calls` table. */
export interface ToolCall {
	host: string;
	session_id: string;
	tool_use_id: string;
	event_uuid: string | null;
	tool_name: string;
	input_json: string | null;
	is_error: boolean | null;
	result_bytes: number | null;
	duration_ms: number | null;
	ts_utc: string | null;
}

/** Mirrors the `sessions` table. */
export interface SessionMeta {
	host: string;
	session_id: string;
	project: string | null;
	git_branch: string | null;
	cc_version: string | null;
	entrypoint: string | null;
	started_at: string | null;
	ended_at: string | null;
}

interface ContentBlock {
	type?: string;
	text?: string;
	thinking?: string;
	id?: string;
	name?: string;
	input?: unknown;
	tool_use_id?: string;
	content?: unknown;
	is_error?: boolean;
}

function asContentBlocks(content: unknown): ContentBlock[] {
	return Array.isArray(content) ? (content as ContentBlock[]) : [];
}

/** Join all `text`-type content blocks, or return a plain-string message as-is. */
function flattenText(content: unknown): string | null {
	if (typeof content === 'string') return content.length > 0 ? content : null;
	const texts = asContentBlocks(content)
		.filter(
			(b): b is ContentBlock & { text: string } => b.type === 'text' && typeof b.text === 'string'
		)
		.map((b) => b.text);
	return texts.length > 0 ? texts.join('\n\n') : null;
}

/** Byte length of tool_result content, which may be a string or a content-block array. */
function byteLengthOf(content: unknown): number {
	if (typeof content === 'string') return Buffer.byteLength(content, 'utf8');
	if (content === undefined || content === null) return 0;
	try {
		return Buffer.byteLength(JSON.stringify(content), 'utf8');
	} catch {
		return 0;
	}
}

interface ParsedToolUse {
	tool_use_id: string;
	tool_name: string;
	input_json: string;
}

interface ParsedToolResult {
	tool_use_id: string;
	is_error: boolean;
	result_bytes: number;
}

export interface LineParseResult {
	event: ParsedEvent;
	toolUses: ParsedToolUse[];
	toolResults: ParsedToolResult[];
}

/**
 * Parse a single JSONL line into a ParsedEvent (plus any tool_use/tool_result blocks it
 * carries), or null if the line isn't a trackable transcript event (no `type`/`uuid`) or
 * isn't valid JSON at all (workflow journal files, a truncated in-flight line, etc.).
 */
export function parseLine(host: string, line: string): LineParseResult | null {
	const trimmed = line.trim();
	if (!trimmed) return null;

	let d: Record<string, unknown>;
	try {
		d = JSON.parse(trimmed) as Record<string, unknown>;
	} catch {
		return null;
	}

	const type = d.type as string | undefined;
	if (!type || typeof d.uuid !== 'string') return null;

	const message = (d.message ?? {}) as Record<string, unknown>;
	const usage = (message.usage ?? {}) as Record<string, unknown>;
	const serverTool = (usage.server_tool_use ?? {}) as Record<string, unknown>;
	const content = message.content;

	const toolUses: ParsedToolUse[] = [];
	const toolResults: ParsedToolResult[] = [];
	for (const block of asContentBlocks(content)) {
		if (
			block.type === 'tool_use' &&
			typeof block.id === 'string' &&
			typeof block.name === 'string'
		) {
			toolUses.push({
				tool_use_id: block.id,
				tool_name: block.name,
				input_json: JSON.stringify(block.input ?? {})
			});
		} else if (block.type === 'tool_result' && typeof block.tool_use_id === 'string') {
			toolResults.push({
				tool_use_id: block.tool_use_id,
				is_error: Boolean(block.is_error),
				result_bytes: byteLengthOf(block.content)
			});
		}
	}

	const numOrNull = (v: unknown): number | null =>
		typeof v === 'number' && Number.isFinite(v) ? v : null;

	const event: ParsedEvent = {
		host,
		session_id: (d.sessionId as string) ?? '',
		uuid: d.uuid,
		parent_uuid: typeof d.parentUuid === 'string' ? d.parentUuid : null,
		ts_utc: typeof d.timestamp === 'string' ? d.timestamp : '',
		type,
		role: typeof message.role === 'string' ? message.role : null,
		model: typeof message.model === 'string' ? message.model : null,
		request_id: typeof d.requestId === 'string' ? d.requestId : null,
		message_id: typeof message.id === 'string' ? message.id : null,
		is_sidechain: Boolean(d.isSidechain),
		is_usage_canonical: false, // set by parseTranscript, which sees the whole batch
		stop_reason: typeof message.stop_reason === 'string' ? message.stop_reason : null,
		latency_ms: null, // set by parseTranscript, which has sibling-line context
		input_tokens: numOrNull(usage.input_tokens),
		output_tokens: numOrNull(usage.output_tokens),
		cache_creation_tokens: numOrNull(usage.cache_creation_input_tokens),
		cache_read_tokens: numOrNull(usage.cache_read_input_tokens),
		web_search_requests: numOrNull(serverTool.web_search_requests),
		web_fetch_requests: numOrNull(serverTool.web_fetch_requests),
		text: flattenText(content)
	};

	return { event, toolUses, toolResults };
}

export interface ParseResult {
	events: ParsedEvent[];
	toolCalls: ToolCall[];
	session: SessionMeta;
}

/**
 * Parse a chunk of JSONL text (a whole file, or just the bytes appended since the last
 * cursor position — see cursor.ts/run.ts) into events, tool calls, and session metadata.
 *
 * `sessionIdFallback` should be the session id derived from the filename (transcripts are
 * named `<session-id>.jsonl`); it's used for lines that for some reason omit `sessionId`,
 * and as the session's id before any line with `sessionId` has been seen.
 *
 * Session metadata (project/branch/version/entrypoint/started_at/ended_at) is scanned
 * across ALL lines in the chunk, including ones that aren't trackable events themselves
 * (e.g. the leading `mode`/`file-history-snapshot` lines carry `cwd`/`gitBranch`). On an
 * incremental (non-backfill) chunk that doesn't include the start of the file, some of
 * these fields may come back null — the ingest server COALESCEs session fields so that's
 * safe (an earlier sweep's non-null values are preserved).
 */
export function parseTranscript(
	host: string,
	sessionIdFallback: string,
	text: string
): ParseResult {
	const events: ParsedEvent[] = [];
	const toolCalls: ToolCall[] = [];

	const tsByUuid = new Map<string, string>();
	const pendingToolCalls = new Map<string, ToolCall>();
	const usageGroups = new Map<string, ParsedEvent[]>();

	let sessionId = sessionIdFallback;
	let project: string | null = null;
	let gitBranch: string | null = null;
	let ccVersion: string | null = null;
	let entrypoint: string | null = null;
	let minTs: string | null = null;
	let maxTs: string | null = null;

	for (const rawLine of text.split('\n')) {
		const trimmed = rawLine.trim();
		if (!trimmed) continue;

		let d: Record<string, unknown>;
		try {
			d = JSON.parse(trimmed) as Record<string, unknown>;
		} catch {
			continue; // malformed/truncated line — skip gracefully
		}

		// Pull session metadata from every line that has it, not just trackable events.
		if (typeof d.sessionId === 'string' && d.sessionId.length > 0) sessionId = d.sessionId;
		if (typeof d.cwd === 'string') project = d.cwd;
		if (typeof d.gitBranch === 'string') gitBranch = d.gitBranch;
		if (typeof d.version === 'string') ccVersion = d.version;
		if (typeof d.entrypoint === 'string') entrypoint = d.entrypoint;
		if (typeof d.timestamp === 'string' && d.timestamp.length > 0) {
			if (!minTs || d.timestamp < minTs) minTs = d.timestamp;
			if (!maxTs || d.timestamp > maxTs) maxTs = d.timestamp;
		}

		const parsed = parseLine(host, trimmed);
		if (!parsed) continue;
		const { event, toolUses, toolResults } = parsed;
		if (!event.session_id) event.session_id = sessionId;

		tsByUuid.set(event.uuid, event.ts_utc);
		if (event.type === 'assistant' && event.parent_uuid && event.ts_utc) {
			const parentTs = tsByUuid.get(event.parent_uuid);
			if (parentTs) {
				const dt = Date.parse(event.ts_utc) - Date.parse(parentTs);
				if (Number.isFinite(dt) && dt >= 0) event.latency_ms = dt;
			}
		}

		events.push(event);

		if (event.message_id) {
			const key = `${event.session_id}${event.message_id}${event.request_id ?? ''}`;
			const group = usageGroups.get(key);
			if (group) group.push(event);
			else usageGroups.set(key, [event]);
		}

		for (const tu of toolUses) {
			pendingToolCalls.set(tu.tool_use_id, {
				host,
				session_id: event.session_id,
				tool_use_id: tu.tool_use_id,
				event_uuid: event.uuid,
				tool_name: tu.tool_name,
				input_json: tu.input_json,
				is_error: null,
				result_bytes: null,
				duration_ms: null,
				ts_utc: event.ts_utc || null
			});
		}

		for (const tr of toolResults) {
			const call = pendingToolCalls.get(tr.tool_use_id);
			if (call) {
				call.is_error = tr.is_error;
				call.result_bytes = tr.result_bytes;
				if (call.ts_utc && event.ts_utc) {
					const dt = Date.parse(event.ts_utc) - Date.parse(call.ts_utc);
					if (Number.isFinite(dt) && dt >= 0) call.duration_ms = dt;
				}
				pendingToolCalls.delete(tr.tool_use_id);
				toolCalls.push(call);
			} else {
				// The matching tool_use was outside this chunk (e.g. parsed in a previous
				// incremental sweep). Still record what we know — the server merges
				// tool_calls rows by (host, session_id, tool_use_id) via COALESCE, so this
				// won't blank out the tool_name/input_json captured earlier.
				toolCalls.push({
					host,
					session_id: event.session_id,
					tool_use_id: tr.tool_use_id,
					event_uuid: null,
					tool_name: 'unknown',
					input_json: null,
					is_error: tr.is_error,
					result_bytes: tr.result_bytes,
					duration_ms: null,
					ts_utc: event.ts_utc || null
				});
			}
		}
	}

	// tool_use blocks still awaiting their result at the end of this chunk — record them
	// now (result_bytes/is_error/duration_ms stay null; a later sweep fills them in).
	for (const call of pendingToolCalls.values()) toolCalls.push(call);

	// Usage dedup: exactly one row per (session_id, message_id, request_id) group is
	// canonical — the max-output_tokens row, tie-broken toward the later (final) line.
	for (const group of usageGroups.values()) {
		let best: ParsedEvent | null = null;
		for (const ev of group) {
			if (
				!best ||
				(ev.output_tokens ?? -1) > (best.output_tokens ?? -1) ||
				((ev.output_tokens ?? -1) === (best.output_tokens ?? -1) && ev.ts_utc >= best.ts_utc)
			) {
				best = ev;
			}
		}
		if (best) best.is_usage_canonical = true;
	}

	const session: SessionMeta = {
		host,
		session_id: sessionId,
		project,
		git_branch: gitBranch,
		cc_version: ccVersion,
		entrypoint,
		started_at: minTs,
		ended_at: maxTs
	};

	return { events, toolCalls, session };
}