import * as fs from "node:fs/promises"; import * as path from "node:path"; import { type AssistantMessage, coerceServiceTierByFamily, getPriorityPremiumRequests, resolveModelServiceTier, type ServiceTierByFamily, type ToolCall, type ToolResultMessage, type Usage, } from "@oh-my-pi/pi-ai"; import { getSessionsDir, isEnoent, readLines } from "@oh-my-pi/pi-utils"; import type { AgentType, MessageStats, SessionEntry, SessionMessageEntry, SessionServiceTierChangeEntry, ToolCallStats, ToolResultLink, UserMessageLink, UserMessageStats, } from "./types"; import { computeUserMessageMetrics } from "./user-metrics"; /** Basename of an advisor agent's transcript inside a session artifacts dir. */ const ADVISOR_TRANSCRIPT_BASENAME = "__advisor.jsonl"; /** * Classify which agent produced a transcript from its path within the sessions * directory. Layout: `//.jsonl` is the `main` * agent; subagent and advisor transcripts live nested one level deeper inside * the session's artifacts dir (`//.jsonl`, * `//__advisor.jsonl`). Any advisor transcript * (`__advisor.jsonl` or `__advisor..jsonl`) — at any depth, including a * subagent's own advisor — counts as `advisor`; every other nested transcript * is a task `subagent`. */ export function classifyAgentType(sessionPath: string): AgentType { const base = path.basename(sessionPath); if (base === ADVISOR_TRANSCRIPT_BASENAME || (base.startsWith("__advisor.") && base.endsWith(".jsonl"))) { return "advisor"; } const rel = path.relative(getSessionsDir(), sessionPath); // `/.jsonl` -> 2 segments. Deeper nesting is a subagent. return rel.split(path.sep).length <= 2 ? "main" : "subagent"; } /** * Extract folder name from session filename. * Session files are named like: --work--pi--/timestamp_uuid.jsonl * The folder part uses -- as path separator. */ function extractFolderFromPath(sessionPath: string): string { const sessionsDir = getSessionsDir(); const rel = path.relative(sessionsDir, sessionPath); const projectDir = rel.split(path.sep)[0]; // Convert --work--pi-- to /work/pi return projectDir.replace(/^--/, "/").replace(/--/g, "/"); } /** * Check if an entry is an assistant message. */ function isAssistantMessage(entry: SessionEntry): entry is SessionMessageEntry { if (entry.type !== "message") return false; const msgEntry = entry as SessionMessageEntry; // Legacy sessions (pre-id tracking) recorded message entries without an `id`. // They're not linkable and would violate the messages.entry_id NOT NULL // constraint, so skip them at the parser boundary. if (typeof msgEntry.id !== "string" || msgEntry.id.length === 0) return false; return msgEntry.message?.role === "assistant"; } /** * Check if an entry is a user message (non-toolResult). */ function isUserMessage(entry: SessionEntry): entry is SessionMessageEntry { if (entry.type !== "message") return false; const msgEntry = entry as SessionMessageEntry; if (typeof msgEntry.id !== "string" || msgEntry.id.length === 0) return false; return msgEntry.message?.role === "user"; } /** * Check if an entry is a service-tier change. */ function isServiceTierChange(entry: SessionEntry): entry is SessionServiceTierChangeEntry { return entry.type === "service_tier_change"; } /** * Check if an entry is a tool-result message. */ function isToolResultMessage(entry: SessionEntry): entry is SessionMessageEntry { if (entry.type !== "message") return false; return (entry as SessionMessageEntry).message?.role === "toolResult"; } /** * Extract plain text from a user message content payload. */ function extractUserText(content: unknown): string { if (typeof content === "string") return content; if (!Array.isArray(content)) return ""; const parts: string[] = []; for (const block of content) { if (block && typeof block === "object" && (block as { type?: unknown }).type === "text") { const text = (block as { text?: unknown }).text; if (typeof text === "string") parts.push(text); } } return parts.join(""); } /** * Build user-message stats from an entry. Returns null for empty/synthetic content. */ function extractUserStats(sessionFile: string, folder: string, entry: SessionMessageEntry): UserMessageStats | null { const msg = entry.message as { role: "user"; content?: unknown; synthetic?: boolean }; if (msg.role !== "user" || msg.synthetic) return null; const text = extractUserText(msg.content); if (!text.trim()) return null; const metrics = computeUserMessageMetrics(text); const ts = Date.parse(entry.timestamp); return { sessionFile, entryId: entry.id, folder, timestamp: Number.isFinite(ts) ? ts : 0, model: null, provider: null, chars: metrics.chars, words: metrics.words, yelling: metrics.yelling, profanity: metrics.profanity, anguish: metrics.anguish, negation: metrics.negation, repetition: metrics.repetition, blame: metrics.blame, }; } /** * Extract stats from an assistant message entry. * * Session JSONL on disk is not guaranteed to match the current * `AssistantMessage` shape: crash-truncated turns, sessions written by older * versions, and foreign producers all flow through this parser. Every field * returned here feeds a NOT NULL column in stats.db, so malformed entries are * coerced (missing `stopReason`, token counts, `timestamp`) or skipped * (missing `model`/`provider`/`api`/`usage`) instead of crashing the whole * sync with a constraint violation. */ function extractStats( sessionFile: string, folder: string, entry: SessionMessageEntry, currentServiceTier: ServiceTierByFamily | undefined, agentType: AgentType, ): MessageStats | null { const msg = entry.message as AssistantMessage; if (msg?.role !== "assistant") return null; if (typeof msg.model !== "string" || typeof msg.provider !== "string" || typeof msg.api !== "string") return null; const rawUsage = msg.usage as Partial | undefined; if (!rawUsage || typeof rawUsage !== "object") return null; // Backfill: when the session recorded `priority` as the active service tier // at this point but the AI usage payload was captured before priority // requests were folded into `premiumRequests`, derive the count here so the // "Premium Reqs" stat aggregates priority traffic on re-sync. Trust any // non-zero value already in `usage.premiumRequests` (Copilot multipliers or // the new AI code path) and only synthesise when the field is missing/zero. const recorded = rawUsage.premiumRequests ?? 0; const model = { provider: msg.provider, api: msg.api, id: msg.model }; const tier = resolveModelServiceTier(currentServiceTier, model); const derived = recorded > 0 ? recorded : getPriorityPremiumRequests(tier, model); const wellFormed = typeof rawUsage.input === "number" && typeof rawUsage.output === "number" && typeof rawUsage.cacheRead === "number" && typeof rawUsage.cacheWrite === "number" && typeof rawUsage.totalTokens === "number"; const usage: Usage = wellFormed && derived === recorded ? (rawUsage as Usage) : { ...rawUsage, input: rawUsage.input ?? 0, output: rawUsage.output ?? 0, cacheRead: rawUsage.cacheRead ?? 0, cacheWrite: rawUsage.cacheWrite ?? 0, totalTokens: rawUsage.totalTokens ?? 0, cost: rawUsage.cost ?? { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, premiumRequests: derived, }; return { sessionFile, entryId: entry.id, folder, model: msg.model, provider: msg.provider, api: msg.api, timestamp: coerceEntryTimestamp(msg.timestamp, entry), duration: msg.duration ?? null, ttft: msg.ttft ?? null, // A message persisted without a terminal stop reason never completed // normally: classify by whether it carried an error. stopReason: msg.stopReason ?? (msg.errorMessage ? "error" : "aborted"), errorMessage: msg.errorMessage ?? null, usage, agentType, }; } /** Message timestamp, falling back to the entry's ISO timestamp, then 0. */ function coerceEntryTimestamp(timestamp: number | undefined, entry: SessionMessageEntry): number { if (typeof timestamp === "number" && Number.isFinite(timestamp)) return timestamp; const ts = Date.parse(entry.timestamp); return Number.isFinite(ts) ? ts : 0; } /** * Extract one {@link ToolCallStats} per `toolCall` content block of an * assistant message. Returns an empty array for turns without tool calls. */ function extractToolCalls( sessionFile: string, folder: string, entry: SessionMessageEntry, agentType: AgentType, ): ToolCallStats[] { const msg = entry.message as AssistantMessage; if (msg?.role !== "assistant" || !Array.isArray(msg.content)) return []; // `tool_calls` columns are NOT NULL: skip turns that can't be attributed // (malformed persisted entries — see extractStats) and blocks missing ids. if (typeof msg.model !== "string" || typeof msg.provider !== "string") return []; const blocks = msg.content.filter( (block): block is ToolCall => block.type === "toolCall" && typeof block.id === "string" && typeof block.name === "string", ); if (blocks.length === 0) return []; return blocks.map(block => { let argsChars = 0; try { argsChars = JSON.stringify(block.arguments ?? {}).length; } catch { // Non-serializable arguments (shouldn't happen in persisted JSONL); size unknown. } return { sessionFile, entryId: entry.id, toolCallId: block.id, folder, toolName: block.name, model: msg.model, provider: msg.provider, timestamp: coerceEntryTimestamp(msg.timestamp, entry), agentType, callsInTurn: blocks.length, argsChars, }; }); } /** * Build the result linkage for a `toolResult` entry: text characters fed back * into context plus the error flag, keyed to the originating call. */ function extractToolResultLink(sessionFile: string, entry: SessionMessageEntry): ToolResultLink | null { const msg = entry.message as ToolResultMessage; if (msg.role !== "toolResult" || typeof msg.toolCallId !== "string" || msg.toolCallId.length === 0) return null; let resultChars = 0; if (Array.isArray(msg.content)) { for (const block of msg.content) { if (block.type === "text" && typeof block.text === "string") resultChars += block.text.length; } } return { sessionFile, toolCallId: msg.toolCallId, resultChars, isError: msg.isError === true, }; } const LF = 0x0a; const CR = 0x0d; const jsonLineDecoder = new TextDecoder(); function parseJsonLine(bytes: Uint8Array, start: number, end: number): SessionEntry | null { while (end > start && bytes[end - 1] === CR) end--; if (end <= start) return null; try { return JSON.parse(jsonLineDecoder.decode(bytes.subarray(start, end))) as SessionEntry; } catch { return null; } } function visitSessionEntriesLenient(bytes: Uint8Array, visit: (entry: SessionEntry) => void): number { let cursor = 0; let read = 0; while (cursor < bytes.length) { const newline = bytes.indexOf(LF, cursor); const hasNewline = newline !== -1; const lineEnd = hasNewline ? newline : bytes.length; const entry = parseJsonLine(bytes, cursor, lineEnd); if (entry) { visit(entry); read = hasNewline ? newline + 1 : lineEnd; } else if (hasNewline) { read = newline + 1; } else { break; } cursor = hasNewline ? newline + 1 : lineEnd; } return read; } function parseSessionEntriesLenient(bytes: Uint8Array): { entries: SessionEntry[]; read: number } { const entries: SessionEntry[] = []; const read = visitSessionEntriesLenient(bytes, entry => entries.push(entry)); return { entries, read }; } function scanLastServiceTier(bytes: Uint8Array): ServiceTierByFamily | undefined { let currentServiceTier: ServiceTierByFamily | undefined; visitSessionEntriesLenient(bytes, entry => { if (isServiceTierChange(entry)) currentServiceTier = coerceServiceTierByFamily(entry.serviceTier); }); return currentServiceTier; } /** * Parse a session file and extract all assistant message stats. * Uses incremental reading with offset tracking. * * Service-tier carry-over: `currentServiceTier` is a session-scoped piece of * state derived from `service_tier_change` entries that affects whether * subsequent OpenAI assistant replies count as premium requests. Incremental * syncs that resume past the most-recent tier change would otherwise lose * that state and silently record `premiumRequests = 0` for priority traffic * (the coding-agent stopped folding the tier into `usage.premiumRequests` * after 13f59162e — the parser is now the sole source of truth). When * `fromOffset > 0` we therefore scan the bytes preceding `fromOffset` * for the latest service-tier value before parsing the unprocessed tail. * The scan only keeps the current tier and does not materialize prefix * entries, preserving offset-based memory behavior for large sessions. */ export interface ParseSessionResult { stats: MessageStats[]; userStats: UserMessageStats[]; userLinks: UserMessageLink[]; toolCalls: ToolCallStats[]; toolResults: ToolResultLink[]; newOffset: number; } export async function parseSessionFile(sessionPath: string, fromOffset = 0): Promise { let bytes: Uint8Array; try { bytes = await Bun.file(sessionPath).bytes(); } catch (err) { if (isEnoent(err)) return { stats: [], userStats: [], userLinks: [], toolCalls: [], toolResults: [], newOffset: fromOffset }; throw err; } const folder = extractFolderFromPath(sessionPath); const agentType = classifyAgentType(sessionPath); const stats: MessageStats[] = []; const userStats: UserMessageStats[] = []; const userLinks: UserMessageLink[] = []; const toolCalls: ToolCallStats[] = []; const toolResults: ToolResultLink[] = []; const userByEntryId = new Map(); const start = Math.max(0, Math.min(fromOffset, bytes.length)); const unprocessed = bytes.subarray(start); const { entries, read } = parseSessionEntriesLenient(unprocessed); let currentServiceTier: ServiceTierByFamily | undefined; if (start > 0) { currentServiceTier = scanLastServiceTier(bytes.subarray(0, start)); } for (const entry of entries) { if (isServiceTierChange(entry)) { currentServiceTier = coerceServiceTierByFamily(entry.serviceTier); continue; } if (isUserMessage(entry)) { const userMsg = extractUserStats(sessionPath, folder, entry); if (userMsg) { userStats.push(userMsg); userByEntryId.set(entry.id, userMsg); } continue; } if (isToolResultMessage(entry)) { const link = extractToolResultLink(sessionPath, entry); if (link) toolResults.push(link); continue; } if (isAssistantMessage(entry)) { const msgStats = extractStats(sessionPath, folder, entry, currentServiceTier, agentType); if (msgStats) stats.push(msgStats); toolCalls.push(...extractToolCalls(sessionPath, folder, entry, agentType)); // Link assistant's responding model back to the user message it answered. const parentId = (entry as SessionMessageEntry).parentId; if (parentId) { const msg = entry.message as AssistantMessage; if (msg.model && msg.provider) { // Emit unconditionally. The aggregator's UPDATE is guarded by // `model IS NULL` so this is idempotent: a no-op for already // linked rows, a fix-up for fresh inserts (which start NULL // because the user row is recorded before its reply lands) and // for cross-pass orphans whose parent was committed by an // earlier incremental sync. userLinks.push({ sessionFile: sessionPath, entryId: parentId, model: msg.model, provider: msg.provider, }); } } } } return { stats, userStats, userLinks, toolCalls, toolResults, newOffset: start + read }; } /** * List all session directories (folders). */ export async function listSessionFolders(): Promise { try { const sessionsDir = getSessionsDir(); const entries = await fs.readdir(sessionsDir, { withFileTypes: true }); return entries.filter(e => e.isDirectory()).map(e => path.join(sessionsDir, e.name)); } catch { return []; } } /** * List all session files in a folder. */ export async function listSessionFiles(folderPath: string): Promise { try { const entries = await fs.readdir(folderPath, { recursive: true, withFileTypes: true }); return entries.filter(e => e.isFile() && e.name.endsWith(".jsonl")).map(e => path.join(e.parentPath, e.name)); } catch { return []; } } /** * List all session files across all folders. */ export async function listAllSessionFiles(): Promise { const folders = await listSessionFolders(); const allFiles: string[] = []; for (const folder of folders) { const files = await listSessionFiles(folder); allFiles.push(...files); } return allFiles; } /** * Find a specific entry in a session file. */ export async function getSessionEntry(sessionPath: string, entryId: string): Promise { try { for await (const line of readLines(Bun.file(sessionPath).stream())) { const entry = parseJsonLine(line, 0, line.length); if (entry && "id" in entry && entry.id === entryId) { return entry; } } } catch (err) { if (isEnoent(err)) return null; throw err; } return null; }