import type { AgentMessage } from "@earendil-works/pi-agent-core"; import { type AssistantMessage, getCurrentSystemMessage, type ImageContent, type Message, type SystemMessage, type TextContent, type ToolResultMessage, type Usage, type UserMessage, uuidv7, } from "@earendil-works/pi-ai"; import { randomUUID } from "crypto"; import { appendFileSync, closeSync, createReadStream, existsSync, mkdirSync, openSync, readdirSync, readSync, type Stats, statSync, writeFileSync, } from "fs"; import { readdir, stat } from "fs/promises"; import { basename, join, resolve } from "path"; import { createInterface } from "readline"; import { StringDecoder } from "string_decoder"; import { APP_NAME, getAgentDir as getDefaultAgentDir, getSessionsDir } from "../config.ts"; import { normalizePath, resolvePath } from "../utils/paths.ts"; import { type BashExecutionMessage, type CustomMessage, createBranchSummaryMessage, createCompactionSummaryMessage, createCustomMessage, } from "./messages.ts"; export const CURRENT_SESSION_VERSION = 3; export interface SessionHeader { type: "session"; version?: number; // v1 sessions don't have this id: string; timestamp: string; cwd: string; parentSession?: string; } export interface NewSessionOptions { id?: string; parentSession?: string; } export interface SessionEntryBase { type: string; id: string; parentId: string | null; timestamp: string; } export interface SessionMessageEntry extends SessionEntryBase { type: "message"; message: AgentMessage; } export interface ThinkingLevelChangeEntry extends SessionEntryBase { type: "thinking_level_change"; thinkingLevel: string; } export interface ModelChangeEntry extends SessionEntryBase { type: "model_change"; provider: string; modelId: string; } export interface UsageEntry extends SessionEntryBase { type: "usage"; /** Arbitrary usage category, such as "cache_warm". */ kind: string; provider: string; model: string; usage: Usage; /** Optional human-readable qualifier for usage notices. */ note?: string; } export interface CompactionEntry extends SessionEntryBase { type: "compaction"; summary: string; firstKeptEntryId: string; tokensBefore: number; /** Extension-specific data (e.g., ArtifactIndex, version markers for structured compaction) */ details?: T; /** Usage from the LLM call(s) that generated this summary, if available */ usage?: Usage; /** True if generated by an extension, undefined/false if pi-generated (backward compatible) */ fromHook?: boolean; /** Complete prompt and tool state at this compaction boundary. */ systemMessage?: SystemMessage; } export interface BranchSummaryEntry extends SessionEntryBase { type: "branch_summary"; fromId: string; summary: string; /** Extension-specific data (not sent to LLM) */ details?: T; /** Usage from the LLM call that generated this summary, if available */ usage?: Usage; /** True if generated by an extension, false if pi-generated */ fromHook?: boolean; } /** * Custom entry for extensions to store extension-specific data in the session. * Use customType to identify your extension's entries. * * Purpose: Persist extension state across session reloads. On reload, extensions can * scan entries for their customType and reconstruct internal state. * * Does NOT participate in LLM context (ignored by buildSessionContext). * For injecting content into context, see CustomMessageEntry. */ export interface CustomEntry extends SessionEntryBase { type: "custom"; customType: string; data?: T; } /** Label entry for user-defined bookmarks/markers on entries. */ export interface LabelEntry extends SessionEntryBase { type: "label"; targetId: string; label: string | undefined; } /** Session metadata entry (e.g., user-defined display name). */ export interface SessionInfoEntry extends SessionEntryBase { type: "session_info"; name?: string; } /** * Custom message entry for extensions to inject messages into LLM context. * Use customType to identify your extension's entries. * * Unlike CustomEntry, this DOES participate in LLM context. * The content is converted to a user message in buildSessionContext(). * Use details for extension-specific metadata (not sent to LLM). * * display controls TUI rendering: * - false: hidden entirely * - true: rendered with distinct styling (different from user messages) */ export interface CustomMessageEntry extends SessionEntryBase { type: "custom_message"; customType: string; content: string | (TextContent | ImageContent)[]; details?: T; display: boolean; } /** Content that an append-only context edit may replace without changing message metadata. */ export type ContextEditableContent = | UserMessage["content"] | AssistantMessage["content"] | ToolResultMessage["content"] | CustomMessage["content"]; /** Append-only change to one earlier entry's contribution to model context. */ export interface ContextEditEntry extends SessionEntryBase { type: "context_edit"; targetId: string; /** Null omits the target from model context. A value replaces only its content. */ replacement: { content: ContextEditableContent } | null; } /** Session entry - has id/parentId for tree structure (returned by "read" methods in SessionManager) */ export type SessionEntry = | SessionMessageEntry | ThinkingLevelChangeEntry | ModelChangeEntry | UsageEntry | CompactionEntry | BranchSummaryEntry | CustomEntry | CustomMessageEntry | ContextEditEntry | LabelEntry | SessionInfoEntry; /** Raw file entry (includes header) */ export type FileEntry = SessionHeader | SessionEntry; /** Tree node for getTree() - defensive copy of session structure */ export interface SessionTreeNode { entry: SessionEntry; children: SessionTreeNode[]; /** Resolved label for this entry, if any */ label?: string; /** Timestamp of the latest label change for this entry, if any */ labelTimestamp?: string; } export interface ProjectedSessionEntry { /** Raw append-only entry that owns this projected contribution. */ sourceEntry: SessionEntry; /** Model-visible messages after context edits. Empty for state-only entries and omissions. */ messages: AgentMessage[]; } export interface SessionProjection { entries: ProjectedSessionEntry[]; messages: AgentMessage[]; thinkingLevel: string; model: { provider: string; modelId: string } | null; } export interface SessionContext { messages: AgentMessage[]; thinkingLevel: string; model: { provider: string; modelId: string } | null; } export interface SessionInfo { path: string; id: string; /** Working directory where the session was started. Empty string for old sessions. */ cwd: string; /** User-defined display name from session_info entries. */ name?: string; /** Path to the parent session (if this session was forked). */ parentSessionPath?: string; created: Date; modified: Date; messageCount: number; firstMessage: string; allMessagesText: string; } export type ReadonlySessionManager = Pick< SessionManager, | "getCwd" | "getSessionDir" | "getSessionId" | "getSessionFile" | "getLeafId" | "getLeafEntry" | "getEntry" | "getLabel" | "getBranch" | "buildContextEntries" | "buildSessionProjection" | "getHeader" | "getEntries" | "getTree" | "getSessionName" >; function createSessionId(): string { return uuidv7(); } export function assertValidSessionId(id: string): void { if (!/^[A-Za-z0-9](?:[A-Za-z0-9._-]*[A-Za-z0-9])?$/.test(id)) { throw new Error( "Session id must be non-empty, contain only alphanumeric characters, '-', '_', and '.', and start and end with an alphanumeric character", ); } } /** Generate a unique short ID (8 hex chars, collision-checked) */ function generateId(byId: { has(id: string): boolean }): string { for (let i = 0; i < 100; i++) { const id = randomUUID().slice(0, 8); if (!byId.has(id)) return id; } // Fallback to full UUID if somehow we have collisions return randomUUID(); } /** Migrate v1 → v2: add id/parentId tree structure. Mutates in place. */ function migrateV1ToV2(entries: FileEntry[]): void { const ids = new Set(); let prevId: string | null = null; for (const entry of entries) { if (entry.type === "session") { entry.version = 2; continue; } entry.id = generateId(ids); entry.parentId = prevId; prevId = entry.id; // Convert firstKeptEntryIndex to firstKeptEntryId for compaction if (entry.type === "compaction") { const comp = entry as CompactionEntry & { firstKeptEntryIndex?: number }; if (typeof comp.firstKeptEntryIndex === "number") { const targetEntry = entries[comp.firstKeptEntryIndex]; if (targetEntry && targetEntry.type !== "session") { comp.firstKeptEntryId = targetEntry.id; } delete comp.firstKeptEntryIndex; } } } } /** Migrate v2 → v3: rename hookMessage role to custom. Mutates in place. */ function migrateV2ToV3(entries: FileEntry[]): void { for (const entry of entries) { if (entry.type === "session") { entry.version = 3; continue; } // Update message entries with hookMessage role if (entry.type === "message") { const msgEntry = entry as SessionMessageEntry; if (msgEntry.message && (msgEntry.message as { role: string }).role === "hookMessage") { (msgEntry.message as { role: string }).role = "custom"; } } } } /** * Run all necessary migrations to bring entries to current version. * Mutates entries in place. Returns true if any migration was applied. */ function migrateToCurrentVersion(entries: FileEntry[]): boolean { const header = entries.find((e) => e.type === "session") as SessionHeader | undefined; const version = header?.version ?? 1; if (version >= CURRENT_SESSION_VERSION) return false; if (version < 2) migrateV1ToV2(entries); if (version < 3) migrateV2ToV3(entries); return true; } /** Exported for testing */ export function migrateSessionEntries(entries: FileEntry[]): void { migrateToCurrentVersion(entries); } /** Exported for compaction.test.ts */ export function parseSessionEntries(content: string): FileEntry[] { const entries: FileEntry[] = []; const lines = content.trim().split("\n"); for (const line of lines) { if (!line.trim()) continue; try { const entry = JSON.parse(line) as FileEntry; entries.push(entry); } catch { // Skip malformed lines } } return entries; } export function getLatestCompactionEntry(entries: SessionEntry[]): CompactionEntry | null { for (let i = entries.length - 1; i >= 0; i--) { if (entries[i].type === "compaction") { return entries[i] as CompactionEntry; } } return null; } function buildEntryIndex(entries: SessionEntry[], byId?: Map): Map { if (byId) return byId; const index = new Map(); for (const entry of entries) { index.set(entry.id, entry); } return index; } function buildSessionPath( entries: SessionEntry[], leafId?: string | null, byId?: Map, ): SessionEntry[] { const index = buildEntryIndex(entries, byId); let leaf: SessionEntry | undefined; if (leafId === null) { return []; } if (leafId) { leaf = index.get(leafId); } leaf ??= entries[entries.length - 1]; if (!leaf) { return []; } const path: SessionEntry[] = []; let current: SessionEntry | undefined = leaf; while (current) { path.push(current); current = current.parentId ? index.get(current.parentId) : undefined; } path.reverse(); return path; } function getSessionContextSettings(path: SessionEntry[]): Pick { let thinkingLevel = "off"; let model: { provider: string; modelId: string } | null = null; for (const entry of path) { if (entry.type === "thinking_level_change") { thinkingLevel = entry.thinkingLevel; } else if (entry.type === "model_change") { model = { provider: entry.provider, modelId: entry.modelId }; } else if (entry.type === "message" && entry.message.role === "assistant") { model = { provider: entry.message.provider, modelId: entry.message.model }; } } return { thinkingLevel, model }; } /** * Project one selected session entry into LLM/runtime messages. * Plain custom entries are display/state entries and do not participate in context. */ export function sessionEntryToContextMessages(entry: SessionEntry): AgentMessage[] { if (entry.type === "message") { const message = entry.message; // Session files are parsed without validation; old versions, forks, or // hand-edited files can contain messages with null/missing content. if (message.role === "system" && message.content == null) return [{ ...message, content: "" }]; if ( (message.role === "user" || message.role === "assistant" || message.role === "toolResult") && message.content == null ) { return [{ ...message, content: [] }]; } return [message]; } if (entry.type === "custom_message") { return [ createCustomMessage(entry.customType, entry.content ?? [], entry.display, entry.details, entry.timestamp), ]; } if (entry.type === "branch_summary" && entry.summary) { return [createBranchSummaryMessage(entry.summary, entry.fromId, entry.timestamp)]; } if (entry.type === "compaction") { const summary = createCompactionSummaryMessage(entry.summary, entry.tokensBefore, entry.timestamp); return entry.systemMessage ? [entry.systemMessage, summary] : [summary]; } return []; } /** * Build the active, compaction-aware session entry list. * * This follows the current leaf path. If the path contains compaction entries, * the latest compaction is represented by the compaction entry itself, followed * by the kept entries starting at firstKeptEntryId and all entries after the * compaction entry. Older summarized entries are omitted. */ export function buildContextEntries( entries: SessionEntry[], leafId?: string | null, byId?: Map, ): SessionEntry[] { const path = buildSessionPath(entries, leafId, byId); let compaction: CompactionEntry | null = null; for (const entry of path) { if (entry.type === "compaction") { compaction = entry; } } if (!compaction) { return path; } const compactionIdx = path.findIndex((entry) => entry.id === compaction.id); if (compactionIdx < 0) { return path; } const contextEntries: SessionEntry[] = [compaction]; let foundFirstKept = false; for (let i = 0; i < compactionIdx; i++) { const entry = path[i]; if (entry.id === compaction.firstKeptEntryId) { foundFirstKept = true; } if (foundFirstKept && !(entry.type === "message" && entry.message.role === "system")) { contextEntries.push(entry); } } contextEntries.push(...path.slice(compactionIdx + 1)); return contextEntries; } /** * Build the session context from entries using tree traversal. * If leafId is provided, walks from that entry to root. * Handles compaction and branch summaries along the path. */ function projectContextEntry(entry: SessionEntry, edit: ContextEditEntry | undefined): AgentMessage[] { const messages = sessionEntryToContextMessages(entry); if (!edit) return messages; const replacement = edit.replacement; if (replacement === null) return []; return messages.map((message) => { if ( message.role !== "user" && message.role !== "assistant" && message.role !== "toolResult" && message.role !== "custom" ) { return message; } const content = (message.role === "assistant" || message.role === "toolResult") && typeof replacement.content === "string" ? [{ type: "text" as const, text: replacement.content }] : replacement.content; return { ...message, content } as AgentMessage; }); } /** Build provenance-preserving, compaction-aware model context. */ export function buildSessionProjection( entries: SessionEntry[], leafId?: string | null, byId?: Map, ): SessionProjection { const path = buildSessionPath(entries, leafId, byId); const { thinkingLevel, model } = getSessionContextSettings(path); const contextEntries = buildContextEntries(entries, leafId, byId); const edits = new Map(); for (const entry of contextEntries) { if (entry.type === "context_edit") edits.set(entry.targetId, entry); } const projectedEntries = contextEntries.map( (sourceEntry, index): ProjectedSessionEntry => ({ sourceEntry, // buildContextEntries() may retain an older compaction entry because its // raw ID lies inside the newest retained range. Only the newest compaction // at index zero contributes a checkpoint and summary. messages: sourceEntry.type === "compaction" && index > 0 ? [] : projectContextEntry(sourceEntry, edits.get(sourceEntry.id)), }), ); return { entries: projectedEntries, messages: projectedEntries.flatMap((entry) => entry.messages), thinkingLevel, model, }; } /** Build the finalized model context from the canonical session projection. */ export function buildSessionContext( entries: SessionEntry[], leafId?: string | null, byId?: Map, ): SessionContext { const { messages, thinkingLevel, model } = buildSessionProjection(entries, leafId, byId); return { messages, thinkingLevel, model }; } /** * Compute the default session directory for a cwd. * Encodes cwd into a safe directory name under ~/.pi/agent/sessions/. */ function getDefaultSessionDirPath(cwd: string, agentDir: string = getDefaultAgentDir()): string { const resolvedCwd = resolvePath(cwd); const resolvedAgentDir = resolvePath(agentDir); const safePath = `--${resolvedCwd.replace(/^[/\\]/, "").replace(/[/\\:]/g, "-")}--`; return join(resolvedAgentDir, "sessions", safePath); } export function getDefaultSessionDir(cwd: string, agentDir: string = getDefaultAgentDir()): string { const sessionDir = getDefaultSessionDirPath(cwd, agentDir); if (!existsSync(sessionDir)) { mkdirSync(sessionDir, { recursive: true }); } return sessionDir; } const SESSION_READ_BUFFER_SIZE = 1024 * 1024; const SESSION_HEADER_READ_BUFFER_SIZE = 4096; /** Bound synchronous header discovery while allowing large cwd and custom metadata fields. */ const MAX_SESSION_HEADER_SCAN_BYTES = 1024 * 1024; class SessionHeaderScanLimitError extends Error { constructor(filePath: string) { super(`Session header exceeds ${MAX_SESSION_HEADER_SCAN_BYTES}-byte scan limit: ${filePath}`); this.name = "SessionHeaderScanLimitError"; } } function parseSessionEntryLine(line: string): FileEntry | null { if (!line.trim()) return null; try { return JSON.parse(line) as FileEntry; } catch { // Skip malformed lines return null; } } /** Exported for testing */ export function loadEntriesFromFile(filePath: string): FileEntry[] { const resolvedFilePath = normalizePath(filePath); if (!existsSync(resolvedFilePath)) return []; const entries: FileEntry[] = []; let pending = ""; const fd = openSync(resolvedFilePath, "r"); try { const decoder = new StringDecoder("utf8"); const buffer = Buffer.allocUnsafe(SESSION_READ_BUFFER_SIZE); while (true) { const bytesRead = readSync(fd, buffer, 0, buffer.length, null); if (bytesRead === 0) break; pending += decoder.write(buffer.subarray(0, bytesRead)); let lineStart = 0; let newlineIndex = pending.indexOf("\n", lineStart); while (newlineIndex !== -1) { const entry = parseSessionEntryLine(pending.slice(lineStart, newlineIndex)); if (entry) entries.push(entry); lineStart = newlineIndex + 1; newlineIndex = pending.indexOf("\n", lineStart); } pending = pending.slice(lineStart); } pending += decoder.end(); const finalEntry = parseSessionEntryLine(pending); if (finalEntry) entries.push(finalEntry); } finally { closeSync(fd); } // Validate session header before repairing the file. if (entries.length === 0) return entries; const header = entries[0]; if (header.type !== "session" || typeof (header as { id?: unknown }).id !== "string") { return []; } if (pending) appendFileSync(resolvedFilePath, "\n"); return entries; } /** * Inspect a physical line while searching for the first parsed session entry. * Blank and malformed lines are skipped to match loadEntriesFromFile(). * Returns undefined to keep scanning, null for a parsed non-header entry, or the header. */ function parseSessionHeaderCandidate(line: string): SessionHeader | null | undefined { if (!line.trim()) return undefined; const entry = parseSessionEntryLine(line); if (!entry) return undefined; if (entry.type !== "session" || typeof (entry as { id?: unknown }).id !== "string") return null; return entry; } function readSessionHeader(filePath: string): SessionHeader | null { const fd = openSync(filePath, "r"); try { const decoder = new StringDecoder("utf8"); const buffer = Buffer.allocUnsafe(SESSION_HEADER_READ_BUFFER_SIZE); const lineChunks: string[] = []; let scannedBytes = 0; while (scannedBytes < MAX_SESSION_HEADER_SCAN_BYTES) { const readLength = Math.min(buffer.length, MAX_SESSION_HEADER_SCAN_BYTES - scannedBytes); const bytesRead = readSync(fd, buffer, 0, readLength, null); if (bytesRead === 0) { lineChunks.push(decoder.end()); return parseSessionHeaderCandidate(lineChunks.join("")) ?? null; } scannedBytes += bytesRead; const chunk = decoder.write(buffer.subarray(0, bytesRead)); let lineStart = 0; let newlineIndex = chunk.indexOf("\n", lineStart); while (newlineIndex !== -1) { lineChunks.push(chunk.slice(lineStart, newlineIndex)); const header = parseSessionHeaderCandidate(lineChunks.join("")); if (header !== undefined) return header; lineChunks.length = 0; lineStart = newlineIndex + 1; newlineIndex = chunk.indexOf("\n", lineStart); } lineChunks.push(chunk.slice(lineStart)); } // Probe for EOF so a final header without a newline is allowed when it ends // exactly at the scan limit. Any additional byte exceeds the bounded scan. const probe = Buffer.allocUnsafe(1); if (readSync(fd, probe, 0, probe.length, null) === 0) { lineChunks.push(decoder.end()); return parseSessionHeaderCandidate(lineChunks.join("")) ?? null; } throw new SessionHeaderScanLimitError(filePath); } finally { closeSync(fd); } } function readSessionHeaderForDiscovery(filePath: string): SessionHeader | null { try { return readSessionHeader(filePath); } catch { // Discovery is best-effort: unreadable or oversized files are not sessions, // and one corrupt file must not prevent other sessions from being found. return null; } } function getSessionHeaderCwd(header: SessionHeader): string | undefined { const cwd = (header as { cwd?: unknown }).cwd; return typeof cwd === "string" ? cwd : undefined; } function sessionCwdMatches(cwd: string | undefined, resolvedCwd: string): boolean { return cwd !== undefined && cwd !== "" && resolvePath(cwd) === resolvedCwd; } /** Exported for testing */ export function findMostRecentSession(sessionDir: string, cwd?: string): string | null { const resolvedSessionDir = normalizePath(sessionDir); const resolvedCwd = cwd ? resolvePath(cwd) : undefined; try { const files = readdirSync(resolvedSessionDir) .filter((file) => file.endsWith(".jsonl")) .map((file) => join(resolvedSessionDir, file)) .map((path) => ({ path, mtime: statSync(path).mtimeMs })) .sort((a, b) => b.mtime - a.mtime); for (const { path } of files) { const header = readSessionHeaderForDiscovery(path); if (header && (!resolvedCwd || sessionCwdMatches(getSessionHeaderCwd(header), resolvedCwd))) return path; } return null; } catch { // Directory access and stat races make recent-session discovery unavailable. return null; } } function isMessageWithContent(message: AgentMessage): message is Message { return typeof (message as Message).role === "string" && "content" in message; } function extractTextContent(message: Message): string { const content = message.content; if (typeof content === "string") { return content; } return content .filter((block): block is TextContent => block.type === "text") .map((block) => block.text) .join(" "); } function getMessageActivityTime(entry: SessionMessageEntry): number | undefined { const message = entry.message; if (!isMessageWithContent(message)) return undefined; if (message.role !== "user" && message.role !== "assistant") return undefined; const msgTimestamp = (message as { timestamp?: number }).timestamp; if (typeof msgTimestamp === "number") { return msgTimestamp; } const t = new Date(entry.timestamp).getTime(); return Number.isNaN(t) ? undefined : t; } async function buildSessionInfo( filePath: string, signal?: AbortSignal, fileStats?: Stats, ): Promise { try { const stats = fileStats ?? (await stat(filePath)); let header: SessionHeader | null = null; let messageCount = 0; let firstMessage = ""; const allMessages: string[] = []; let name: string | undefined; let lastActivityTime: number | undefined; const rl = createInterface({ input: createReadStream(filePath, { encoding: "utf8", signal }), crlfDelay: Infinity, }); for await (const line of rl) { const entry = parseSessionEntryLine(line); if (!entry) continue; if (!header) { if (entry.type !== "session") return null; header = entry; continue; } // Extract session name (use latest, including explicit clears) if (entry.type === "session_info") { name = entry.name?.trim() || undefined; } if (entry.type !== "message") continue; messageCount++; const activityTime = getMessageActivityTime(entry); if (typeof activityTime === "number") { lastActivityTime = Math.max(lastActivityTime ?? 0, activityTime); } const message = entry.message; if (!isMessageWithContent(message)) continue; if (message.role !== "user" && message.role !== "assistant") continue; const textContent = extractTextContent(message); if (!textContent) continue; allMessages.push(textContent); if (!firstMessage && message.role === "user") { firstMessage = textContent; } } if (!header) return null; const cwd = typeof header.cwd === "string" ? header.cwd : ""; const parentSessionPath = header.parentSession; const headerTime = typeof header.timestamp === "string" ? new Date(header.timestamp).getTime() : NaN; const modified = typeof lastActivityTime === "number" && lastActivityTime > 0 ? new Date(lastActivityTime) : !Number.isNaN(headerTime) ? new Date(headerTime) : stats.mtime; return { path: filePath, id: header.id, cwd, name, parentSessionPath, created: new Date(header.timestamp), modified, messageCount, firstMessage: firstMessage || "(no messages)", allMessagesText: allMessages.join(" "), }; } catch { signal?.throwIfAborted(); return null; } } export type SessionListProgress = ( loaded: number, total: number, /** Sessions loaded so far, sorted by activity. Present on periodic updates. */ partialSessions?: readonly SessionInfo[], ) => void; const MAX_CONCURRENT_SESSION_INFO_LOADS = 10; const MAX_CONCURRENT_SESSION_DISCOVERY_LOADS = 64; const CURRENT_SESSION_LIST_PUBLISH_INTERVAL = 10; const ALL_SESSION_LIST_PUBLISH_INTERVAL = 100; interface SessionFileCandidate { path: string; stats?: Stats; } async function mapWithConcurrency( items: T[], limit: number, map: (item: T, index: number) => Promise, signal?: AbortSignal, ): Promise { const results = new Array(items.length); let nextIndex = 0; const worker = async (): Promise => { while (nextIndex < items.length) { signal?.throwIfAborted(); const index = nextIndex++; results[index] = await map(items[index]!, index); } }; await Promise.all(Array.from({ length: Math.min(limit, items.length) }, () => worker())); return results; } function sortSessionInfos(sessions: SessionInfo[]): SessionInfo[] { return sessions.sort((a, b) => b.modified.getTime() - a.modified.getTime()); } function buildSessionInfosWithConcurrency( files: SessionFileCandidate[], onLoaded: (info: SessionInfo | null, index: number) => void, signal?: AbortSignal, ): Promise<(SessionInfo | null)[]> { return mapWithConcurrency( files, MAX_CONCURRENT_SESSION_INFO_LOADS, async (file, index) => { const info = await buildSessionInfo(file.path, signal, file.stats); onLoaded(info, index); return info; }, signal, ); } async function listSessionsFromDir( dir: string, onProgress?: SessionListProgress, signal?: AbortSignal, ): Promise { signal?.throwIfAborted(); if (!existsSync(dir)) return []; try { const dirEntries = await readdir(dir); const files = dirEntries .filter((file) => file.endsWith(".jsonl")) .sort((a, b) => b.localeCompare(a)) .map((file) => ({ path: join(dir, file) })); const total = files.length; const partialSessions: SessionInfo[] = []; let loaded = 0; const results = await buildSessionInfosWithConcurrency( files, (info) => { loaded++; if (info) partialSessions.push(info); const publishPartial = loaded === 1 || loaded % CURRENT_SESSION_LIST_PUBLISH_INTERVAL === 0 || loaded === files.length; onProgress?.(loaded, total, publishPartial ? sortSessionInfos([...partialSessions]) : undefined); }, signal, ); return results.filter((info): info is SessionInfo => info !== null); } catch { signal?.throwIfAborted(); return []; } } /** * Manages conversation sessions as append-only trees stored in JSONL files. * * Each session entry has an id and parentId forming a tree structure. The "leaf" * pointer tracks the current position. Appending creates a child of the current leaf. * Branching moves the leaf to an earlier entry, allowing new branches without * modifying history. * * Use buildSessionContext() to get the resolved message list for the LLM, which * handles compaction summaries and follows the path from root to current leaf. */ export class SessionManager { private sessionId: string = ""; private sessionFile: string | undefined; private sessionDir: string; private cwd: string; private persist: boolean; private flushed: boolean = false; private fileEntries: FileEntry[] = []; private byId: Map = new Map(); private labelsById: Map = new Map(); private labelTimestampsById: Map = new Map(); private leafId: string | null = null; private constructor( cwd: string, sessionDir: string, sessionFile: string | undefined, persist: boolean, newSessionOptions?: NewSessionOptions, preloadedFileEntries?: FileEntry[], ) { this.cwd = resolvePath(cwd); this.sessionDir = normalizePath(sessionDir); this.persist = persist; if (persist && this.sessionDir && !existsSync(this.sessionDir)) { mkdirSync(this.sessionDir, { recursive: true }); } if (sessionFile) { this._setSessionFile(sessionFile, preloadedFileEntries); } else if (preloadedFileEntries?.length) { this._loadEntries(preloadedFileEntries, newSessionOptions); } else { this.newSession(newSessionOptions); } } /** Switch to a different session file (used for resume and branching) */ setSessionFile(sessionFile: string): void { this._setSessionFile(sessionFile); } private _setSessionFile(sessionFile: string, preloadedFileEntries?: FileEntry[]): void { this.sessionFile = resolvePath(sessionFile); if (existsSync(this.sessionFile)) { const entries = preloadedFileEntries ?? loadEntriesFromFile(this.sessionFile); // If file was empty, initialize it with a valid session header. If it was // non-empty but did not parse as a pi session, fail without modifying it. if (entries.length === 0) { const explicitPath = this.sessionFile; if (statSync(explicitPath).size > 0) { throw new Error(`Session file is not a valid ${APP_NAME} session: ${explicitPath}`); } this.newSession(); this.sessionFile = explicitPath; this._rewriteFile(); this.flushed = true; return; } this._loadEntries(entries); this.flushed = true; } else { const explicitPath = this.sessionFile; this.newSession(); this.sessionFile = explicitPath; // preserve explicit path from --session flag } } newSession(options?: NewSessionOptions): string | undefined { if (options?.id !== undefined) { assertValidSessionId(options.id); } this.sessionId = options?.id ?? createSessionId(); const timestamp = new Date().toISOString(); const header: SessionHeader = { type: "session", version: CURRENT_SESSION_VERSION, id: this.sessionId, timestamp, cwd: this.cwd, parentSession: options?.parentSession, }; this.fileEntries = [header]; this.byId.clear(); this.labelsById.clear(); this.labelTimestampsById.clear(); this.leafId = null; this.flushed = false; if (this.persist) { const fileTimestamp = timestamp.replace(/[:.]/g, "-"); this.sessionFile = join(this.getSessionDir(), `${fileTimestamp}_${this.sessionId}.jsonl`); } return this.sessionFile; } private _loadEntries(entries: FileEntry[], options?: NewSessionOptions): void { const header = entries.find((e) => e.type === "session") as SessionHeader | undefined; if (header) { this.fileEntries = entries; this.sessionId = header.id; if (migrateToCurrentVersion(this.fileEntries)) { this._rewriteFile(); } } else { this.newSession(options); this.fileEntries = this.fileEntries.concat(entries); } this._buildIndex(); } private _buildIndex(): void { this.byId.clear(); this.labelsById.clear(); this.labelTimestampsById.clear(); this.leafId = null; for (const entry of this.fileEntries) { if (entry.type === "session") continue; this.byId.set(entry.id, entry); this.leafId = entry.id; if (entry.type === "label") { if (entry.label) { this.labelsById.set(entry.targetId, entry.label); this.labelTimestampsById.set(entry.targetId, entry.timestamp); } else { this.labelsById.delete(entry.targetId); this.labelTimestampsById.delete(entry.targetId); } } } } private _rewriteFile(): void { if (!this.persist || !this.sessionFile) return; const fd = openSync(this.sessionFile, "w"); try { for (const entry of this.fileEntries) { writeFileSync(fd, `${JSON.stringify(entry)}\n`); } } finally { closeSync(fd); } } isPersisted(): boolean { return this.persist; } getCwd(): string { return this.cwd; } getSessionDir(): string { return this.sessionDir; } usesDefaultSessionDir(): boolean { return this.sessionDir === getDefaultSessionDirPath(this.cwd); } getSessionId(): string { return this.sessionId; } getSessionFile(): string | undefined { return this.sessionFile; } /** * A new session file is created only once the session contains a user or assistant message. * Setup entries alone (model, thinking level, system prompt) stay in memory so opening and * closing pi without chatting leaves no file behind. Starting at the user message (not the * first assistant reply) keeps the prompt on disk if the first turn never completes (#10000). */ private _hasConversation(): boolean { return this.fileEntries.some( (e) => e.type === "message" && (e.message.role === "user" || e.message.role === "assistant"), ); } _persist(entry: SessionEntry): void { if (!this.persist || !this.sessionFile) return; if (!this.flushed) { if (!this._hasConversation()) return; const fd = openSync(this.sessionFile, "wx"); try { for (const e of this.fileEntries) { writeFileSync(fd, `${JSON.stringify(e)}\n`); } } finally { closeSync(fd); } this.flushed = true; } else { appendFileSync(this.sessionFile, `${JSON.stringify(entry)}\n`); } } private _appendEntry(entry: SessionEntry): void { this.fileEntries.push(entry); this.byId.set(entry.id, entry); this.leafId = entry.id; this._persist(entry); } /** Append a message as child of current leaf, then advance leaf. Returns entry id. * Does not allow writing CompactionSummaryMessage and BranchSummaryMessage directly. * Reason: we want these to be top-level entries in the session, not message session entries, * so it is easier to find them. * These need to be appended via appendCompaction() and appendBranchSummary() methods. */ appendMessage(message: Message | CustomMessage | BashExecutionMessage): string { const entry: SessionMessageEntry = { type: "message", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), message, }; this._appendEntry(entry); return entry.id; } /** Append a thinking level change as child of current leaf, then advance leaf. Returns entry id. */ appendThinkingLevelChange(thinkingLevel: string): string { const entry: ThinkingLevelChangeEntry = { type: "thinking_level_change", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), thinkingLevel, }; this._appendEntry(entry); return entry.id; } /** Append a model change as child of current leaf, then advance leaf. Returns entry id. */ appendModelChange(provider: string, modelId: string): string { const entry: ModelChangeEntry = { type: "model_change", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), provider, modelId, }; this._appendEntry(entry); return entry.id; } /** Append model-attributed usage that does not participate in LLM context. Returns the appended entry. */ appendUsage(kind: string, provider: string, model: string, usage: Usage, note?: string): UsageEntry { const entry: UsageEntry = { type: "usage", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), kind, provider, model, usage, ...(note ? { note } : {}), }; this._appendEntry(entry); return entry; } /** Append a compaction summary as child of current leaf, then advance leaf. Returns entry id. */ appendCompaction( summary: string, firstKeptEntryId: string | null, tokensBefore: number, details?: T, fromHook?: boolean, usage?: Usage, ): string { const timestamp = new Date().toISOString(); const systemMessage = getCurrentSystemMessage(this.buildSessionProjection().messages); const id = generateId(this.byId); const entry: CompactionEntry = { type: "compaction", id, parentId: this.leafId, timestamp, summary, firstKeptEntryId: firstKeptEntryId ?? id, tokensBefore, details, usage, fromHook, ...(systemMessage ? { systemMessage: { ...systemMessage, timestamp: new Date(timestamp).getTime() } } : {}), }; this._appendEntry(entry); return entry.id; } /** Append a custom entry (for extensions) as child of current leaf, then advance leaf. Returns entry id. */ appendCustomEntry(customType: string, data?: unknown): string { const entry: CustomEntry = { type: "custom", customType, data, id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), }; this._appendEntry(entry); return entry.id; } /** Append a session info entry (e.g., display name). Returns entry id. */ appendSessionInfo(name: string): string { const sanitizedName = name.replace(/[\r\n]+/g, " ").trim(); const entry: SessionInfoEntry = { type: "session_info", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), name: sanitizedName, }; this._appendEntry(entry); return entry.id; } /** Get the current session name from the latest session_info entry, if any. */ getSessionName(): string | undefined { // Walk entries in reverse to find the latest session_info entry. // Empty names explicitly clear the session title. Reads fileEntries directly: the footer // calls this on every frame, and getEntries() copies the whole session. for (let i = this.fileEntries.length - 1; i >= 0; i--) { const entry = this.fileEntries[i]; if (entry.type === "session_info") { return entry.name?.trim() || undefined; } } return undefined; } /** * Append a custom message entry (for extensions) that participates in LLM context. * @param customType Extension identifier for filtering on reload * @param content Message content (string or TextContent/ImageContent array) * @param display Whether to show in TUI (true = styled display, false = hidden) * @param details Optional extension-specific metadata (not sent to LLM) * @returns Entry id */ appendCustomMessageEntry( customType: string, content: string | (TextContent | ImageContent)[], display: boolean, details?: T, ): string { const entry: CustomMessageEntry = { type: "custom_message", customType, content, display, details, id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), }; this._appendEntry(entry); return entry.id; } /** Append a branch-local edit to an earlier model-visible entry. */ appendContextEdit(targetId: string, replacement: ContextEditEntry["replacement"]): string { if ( replacement !== null && (typeof replacement !== "object" || !("content" in replacement) || (typeof replacement.content !== "string" && !Array.isArray(replacement.content))) ) { throw new Error("Context edit replacement must be null or contain string/array content"); } const target = this.byId.get(targetId); if (!target) throw new Error(`Entry ${targetId} not found`); if (!this.getBranch().some((entry) => entry.id === targetId)) { throw new Error(`Entry ${targetId} is not on the active branch`); } const editable = target.type === "custom_message" || (target.type === "message" && (target.message.role === "user" || target.message.role === "assistant" || target.message.role === "toolResult")); if (!editable) throw new Error(`Entry ${targetId} does not contribute editable model content`); const targetRole = target.type === "message" ? target.message.role : "custom"; const normalizedReplacement = replacement !== null && (targetRole === "assistant" || targetRole === "toolResult") && typeof replacement.content === "string" ? { content: [{ type: "text" as const, text: replacement.content }] } : replacement; const entry: ContextEditEntry = { type: "context_edit", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), targetId, replacement: normalizedReplacement, }; this._appendEntry(entry); return entry.id; } // ========================================================================= // Tree Traversal // ========================================================================= getLeafId(): string | null { return this.leafId; } getLeafEntry(): SessionEntry | undefined { return this.leafId ? this.byId.get(this.leafId) : undefined; } getEntry(id: string): SessionEntry | undefined { return this.byId.get(id); } /** * Get all direct children of an entry. */ getChildren(parentId: string): SessionEntry[] { const children: SessionEntry[] = []; for (const entry of this.byId.values()) { if (entry.parentId === parentId) { children.push(entry); } } return children; } /** * Get the label for an entry, if any. */ getLabel(id: string): string | undefined { return this.labelsById.get(id); } /** * Set or clear a label on an entry. * Labels are user-defined markers for bookmarking/navigation. * Pass undefined or empty string to clear the label. */ appendLabelChange(targetId: string, label: string | undefined): string { if (!this.byId.has(targetId)) { throw new Error(`Entry ${targetId} not found`); } const entry: LabelEntry = { type: "label", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), targetId, label, }; this._appendEntry(entry); if (label) { this.labelsById.set(targetId, label); this.labelTimestampsById.set(targetId, entry.timestamp); } else { this.labelsById.delete(targetId); this.labelTimestampsById.delete(targetId); } return entry.id; } /** * Walk from entry to root, returning all entries in path order. * Includes all entry types (messages, compaction, model changes, etc.). * Use buildSessionContext() to get the resolved messages for the LLM. */ getBranch(fromId?: string): SessionEntry[] { const path: SessionEntry[] = []; const startId = fromId ?? this.leafId; let current = startId ? this.byId.get(startId) : undefined; while (current) { path.push(current); current = current.parentId ? this.byId.get(current.parentId) : undefined; } path.reverse(); return path; } /** * Build the active, compaction-aware entry list for context/rendering. * Uses tree traversal from current leaf. */ buildContextEntries(): SessionEntry[] { return buildContextEntries(this.getEntries(), this.leafId, this.byId); } /** * Build the session context (what gets sent to the LLM). * Uses tree traversal from current leaf. */ buildSessionProjection(): SessionProjection { return buildSessionProjection(this.getEntries(), this.leafId, this.byId); } buildSessionContext(): SessionContext { const { messages, thinkingLevel, model } = this.buildSessionProjection(); return { messages, thinkingLevel, model }; } /** * Get session header. */ getHeader(): SessionHeader | null { const h = this.fileEntries.find((e) => e.type === "session"); return h ? (h as SessionHeader) : null; } /** Number of session entries (excludes header), without copying them like `getEntries()`. */ getEntryCount(): number { return this.byId.size; } /** * Get all session entries (excludes header). Returns a shallow copy. * The session is append-only: use appendXXX() to add entries, branch() to * change the leaf pointer. Entries cannot be modified or deleted. */ getEntries(): SessionEntry[] { return this.fileEntries.filter((e): e is SessionEntry => e.type !== "session"); } /** * Get the session as a tree structure. Returns a shallow defensive copy of all entries. * A well-formed session has exactly one root (first entry with parentId === null). * Orphaned entries (broken parent chain) are also returned as roots. */ getTree(): SessionTreeNode[] { const entries = this.getEntries(); const nodeMap = new Map(); const roots: SessionTreeNode[] = []; // Create nodes with resolved labels for (const entry of entries) { const label = this.labelsById.get(entry.id); const labelTimestamp = this.labelTimestampsById.get(entry.id); nodeMap.set(entry.id, { entry, children: [], label, labelTimestamp }); } // Build tree for (const entry of entries) { const node = nodeMap.get(entry.id)!; if (entry.parentId === null || entry.parentId === entry.id) { roots.push(node); } else { const parent = nodeMap.get(entry.parentId); if (parent) { parent.children.push(node); } else { // Orphan - treat as root roots.push(node); } } } // Sort children by timestamp (oldest first, newest at bottom) // Use iterative approach to avoid stack overflow on deep trees const stack: SessionTreeNode[] = [...roots]; while (stack.length > 0) { const node = stack.pop()!; node.children.sort((a, b) => new Date(a.entry.timestamp).getTime() - new Date(b.entry.timestamp).getTime()); stack.push(...node.children); } return roots; } // ========================================================================= // Branching // ========================================================================= /** * Start a new branch from an earlier entry. * Moves the leaf pointer to the specified entry. The next appendXXX() call * will create a child of that entry, forming a new branch. Existing entries * are not modified or deleted. */ branch(branchFromId: string): void { if (!this.byId.has(branchFromId)) { throw new Error(`Entry ${branchFromId} not found`); } this.leafId = branchFromId; } /** * Reset the leaf pointer to null (before any entries). * The next appendXXX() call will create a new root entry (parentId = null). * Use this when navigating to re-edit the first user message. */ resetLeaf(): void { this.leafId = null; } /** * Start a new branch with a summary of the abandoned path. * Same as branch(), but also appends a branch_summary entry that captures * context from the abandoned conversation path. */ branchWithSummary( branchFromId: string | null, summary: string, details?: unknown, fromHook?: boolean, usage?: Usage, ): string { if (branchFromId !== null && !this.byId.has(branchFromId)) { throw new Error(`Entry ${branchFromId} not found`); } const fromId = this.leafId ?? "root"; this.leafId = branchFromId; const entry: BranchSummaryEntry = { type: "branch_summary", id: generateId(this.byId), parentId: branchFromId, timestamp: new Date().toISOString(), fromId, summary, details, usage, fromHook, }; this._appendEntry(entry); return entry.id; } /** * Create a new session file containing only the path from root to the specified leaf. * Useful for extracting a single conversation path from a branched session. * Returns the new session file path, or undefined if not persisting. */ createBranchedSession(leafId: string): string | undefined { const previousSessionFile = this.sessionFile; const path = this.getBranch(leafId); if (path.length === 0) { throw new Error(`Entry ${leafId} not found`); } // Filter out LabelEntry from path - we'll recreate them from the resolved map. // Because labels are real tree entries, later entries can be children of labels; // removing labels requires re-chaining the retained path to avoid orphaned subtrees. const pathWithoutLabels: SessionEntry[] = []; const replacementByLabelId = new Map(); const pendingLabelIds: string[] = []; let pathParentId: string | null = null; for (const entry of path) { if (entry.type === "label") { pendingLabelIds.push(entry.id); continue; } for (const labelId of pendingLabelIds) { replacementByLabelId.set(labelId, entry.id); } pendingLabelIds.length = 0; pathWithoutLabels.push( entry.type === "compaction" ? { ...entry, parentId: pathParentId, firstKeptEntryId: entry.firstKeptEntryId === entry.id ? entry.id : (replacementByLabelId.get(entry.firstKeptEntryId) ?? entry.firstKeptEntryId), } : { ...entry, parentId: pathParentId }, ); pathParentId = entry.id; } const newSessionId = createSessionId(); const timestamp = new Date().toISOString(); const fileTimestamp = timestamp.replace(/[:.]/g, "-"); const newSessionFile = join(this.getSessionDir(), `${fileTimestamp}_${newSessionId}.jsonl`); const header: SessionHeader = { type: "session", version: CURRENT_SESSION_VERSION, id: newSessionId, timestamp, cwd: this.cwd, parentSession: this.persist ? previousSessionFile : undefined, }; // Collect labels for entries in the path const pathEntryIds = new Set(pathWithoutLabels.map((e) => e.id)); const labelsToWrite: Array<{ targetId: string; label: string; timestamp: string }> = []; for (const [targetId, label] of this.labelsById) { if (pathEntryIds.has(targetId)) { labelsToWrite.push({ targetId, label, timestamp: this.labelTimestampsById.get(targetId)! }); } } if (this.persist) { // Build label entries const lastEntryId = pathWithoutLabels[pathWithoutLabels.length - 1]?.id || null; let parentId = lastEntryId; const labelEntries: LabelEntry[] = []; for (const { targetId, label, timestamp: labelTimestamp } of labelsToWrite) { const labelEntry: LabelEntry = { type: "label", id: generateId(new Set(pathEntryIds)), parentId, timestamp: labelTimestamp, targetId, label, }; pathEntryIds.add(labelEntry.id); labelEntries.push(labelEntry); parentId = labelEntry.id; } this.fileEntries = [header, ...pathWithoutLabels, ...labelEntries]; this.sessionId = newSessionId; this.sessionFile = newSessionFile; this._buildIndex(); // Use the same rule as _persist(): write now if the branched path already // has a conversation, otherwise let _persist() create the file later. if (this._hasConversation()) { this._rewriteFile(); this.flushed = true; } else { this.flushed = false; } return newSessionFile; } // In-memory mode: replace current session with the path + labels const labelEntries: LabelEntry[] = []; let parentId = pathWithoutLabels[pathWithoutLabels.length - 1]?.id || null; for (const { targetId, label, timestamp: labelTimestamp } of labelsToWrite) { const labelEntry: LabelEntry = { type: "label", id: generateId(new Set([...pathEntryIds, ...labelEntries.map((e) => e.id)])), parentId, timestamp: labelTimestamp, targetId, label, }; labelEntries.push(labelEntry); parentId = labelEntry.id; } this.fileEntries = [header, ...pathWithoutLabels, ...labelEntries]; this.sessionId = newSessionId; this._buildIndex(); return undefined; } /** * Create a new session. * @param cwd Working directory (stored in session header) * @param sessionDir Optional session directory. If omitted, uses default (~/.pi/agent/sessions//). */ static create(cwd: string, sessionDir?: string, options?: NewSessionOptions): SessionManager { const dir = sessionDir ? normalizePath(sessionDir) : getDefaultSessionDir(cwd); return new SessionManager(cwd, dir, undefined, true, options); } /** * Open a specific session file. * @param path Path to session file * @param sessionDir Optional session directory for /new or /branch. If omitted, derives from file's parent. * @param cwdOverride Optional cwd override instead of the session header cwd. */ static open(path: string, sessionDir?: string, cwdOverride?: string): SessionManager { const resolvedPath = resolvePath(path); let header: SessionHeader | null = null; let preloadedFileEntries: FileEntry[] | undefined; if (cwdOverride === undefined && existsSync(resolvedPath)) { try { header = readSessionHeader(resolvedPath); } catch (error) { if (!(error instanceof SessionHeaderScanLimitError)) throw error; // The bounded scan is only a discovery optimization. A full load remains // authoritative for legacy files with very large headers or prefixes. preloadedFileEntries = loadEntriesFromFile(resolvedPath); const firstEntry = preloadedFileEntries[0]; header = firstEntry?.type === "session" ? firstEntry : null; } } const cwd = cwdOverride ?? (header ? getSessionHeaderCwd(header) : undefined) ?? process.cwd(); // If no sessionDir provided, derive from file's parent directory const dir = sessionDir ? normalizePath(sessionDir) : resolve(resolvedPath, ".."); return new SessionManager(cwd, dir, resolvedPath, true, undefined, preloadedFileEntries); } /** * Continue the most recent session, or create new if none. * @param cwd Working directory * @param sessionDir Optional session directory. If omitted, uses default (~/.pi/agent/sessions//). */ static continueRecent(cwd: string, sessionDir?: string): SessionManager { const dir = sessionDir ? normalizePath(sessionDir) : getDefaultSessionDir(cwd); const filterCwd = sessionDir !== undefined && dir !== getDefaultSessionDirPath(cwd); const mostRecent = findMostRecentSession(dir, filterCwd ? cwd : undefined); if (mostRecent) { return new SessionManager(cwd, dir, mostRecent, true); } return new SessionManager(cwd, dir, undefined, true); } /** Create an in-memory session (no file persistence), optionally from entries held outside the filesystem. */ static inMemory(cwd: string = process.cwd(), options?: NewSessionOptions, entries?: FileEntry[]): SessionManager { return new SessionManager(cwd, "", undefined, false, options, entries); } /** * Fork a session from another project directory into the current project. * Creates a new session in the target cwd with the full history from the source session. * @param sourcePath Path to the source session file * @param targetCwd Target working directory (where the new session will be stored) * @param sessionDir Optional session directory. If omitted, uses default for targetCwd. */ static forkFrom( sourcePath: string, targetCwd: string, sessionDir?: string, options?: NewSessionOptions, ): SessionManager { const resolvedSourcePath = resolvePath(sourcePath); const resolvedTargetCwd = resolvePath(targetCwd); const sourceEntries = loadEntriesFromFile(resolvedSourcePath); if (sourceEntries.length === 0) { throw new Error(`Cannot fork: source session file is empty or invalid: ${resolvedSourcePath}`); } const sourceHeader = sourceEntries.find((e) => e.type === "session") as SessionHeader | undefined; if (!sourceHeader) { throw new Error(`Cannot fork: source session has no header: ${resolvedSourcePath}`); } const dir = sessionDir ? normalizePath(sessionDir) : getDefaultSessionDir(resolvedTargetCwd); if (!existsSync(dir)) { mkdirSync(dir, { recursive: true }); } // Create new session file with new ID but forked content if (options?.id !== undefined) { assertValidSessionId(options.id); } const newSessionId = options?.id ?? createSessionId(); const timestamp = new Date().toISOString(); const fileTimestamp = timestamp.replace(/[:.]/g, "-"); const newSessionFile = join(dir, `${fileTimestamp}_${newSessionId}.jsonl`); // Write new header pointing to source as parent, with updated cwd const newHeader: SessionHeader = { type: "session", version: CURRENT_SESSION_VERSION, id: newSessionId, timestamp, cwd: resolvedTargetCwd, parentSession: resolvedSourcePath, }; writeFileSync(newSessionFile, `${JSON.stringify(newHeader)}\n`, { flag: "wx" }); // Copy all non-header entries from source for (const entry of sourceEntries) { if (entry.type !== "session") { appendFileSync(newSessionFile, `${JSON.stringify(entry)}\n`); } } return new SessionManager(resolvedTargetCwd, dir, newSessionFile, true); } /** * Find an exact session ID without loading transcript bodies. * @param cwd Working directory (used to compute default session directory) * @param id Exact session ID * @param sessionDir Optional session directory. If omitted, uses default (~/.pi/agent/sessions//). */ static findById(cwd: string, id: string, sessionDir?: string): string | undefined { const dir = sessionDir ? normalizePath(sessionDir) : getDefaultSessionDir(cwd); const filterCwd = sessionDir !== undefined && dir !== getDefaultSessionDirPath(cwd); const resolvedCwd = resolvePath(cwd); try { for (const file of readdirSync(dir)) { if (!file.endsWith(".jsonl")) continue; const path = join(dir, file); const header = readSessionHeaderForDiscovery(path); if (header?.id !== id) continue; if (filterCwd && !sessionCwdMatches(getSessionHeaderCwd(header), resolvedCwd)) continue; return path; } } catch { // Exact session discovery is best-effort, matching list(). } return undefined; } /** * List all sessions for a directory. * @param cwd Working directory (used to compute default session directory) * @param sessionDir Optional session directory. If omitted, uses default (~/.pi/agent/sessions//). * @param onProgress Optional callback for progress updates (loaded, total) */ static async list( cwd: string, sessionDir?: string, onProgress?: SessionListProgress, signal?: AbortSignal, ): Promise { const dir = sessionDir ? normalizePath(sessionDir) : getDefaultSessionDir(cwd); const filterCwd = sessionDir !== undefined && dir !== getDefaultSessionDirPath(cwd); const resolvedCwd = resolvePath(cwd); const includeSession = (session: SessionInfo) => !filterCwd || sessionCwdMatches(session.cwd, resolvedCwd); const progress: SessionListProgress | undefined = onProgress ? (loaded, total, partialSessions) => onProgress(loaded, total, partialSessions?.filter(includeSession)) : undefined; const sessions = (await listSessionsFromDir(dir, progress, signal)).filter(includeSession); return sortSessionInfos(sessions); } /** * List all sessions across all project directories. * @param onProgress Optional callback for progress updates (loaded, total) */ static async listAll(onProgress?: SessionListProgress, signal?: AbortSignal): Promise; static async listAll( sessionDir?: string, onProgress?: SessionListProgress, signal?: AbortSignal, ): Promise; static async listAll( sessionDirOrOnProgress?: string | SessionListProgress, onProgressOrSignal?: SessionListProgress | AbortSignal, signal?: AbortSignal, ): Promise { const customSessionDir = typeof sessionDirOrOnProgress === "string" ? normalizePath(sessionDirOrOnProgress) : undefined; const progress = typeof sessionDirOrOnProgress === "function" ? sessionDirOrOnProgress : typeof onProgressOrSignal === "function" ? onProgressOrSignal : undefined; const abortSignal = typeof sessionDirOrOnProgress === "string" || typeof onProgressOrSignal === "function" ? signal : (onProgressOrSignal ?? signal); abortSignal?.throwIfAborted(); if (customSessionDir) { return sortSessionInfos(await listSessionsFromDir(customSessionDir, progress, abortSignal)); } const sessionsDir = getSessionsDir(); try { if (!existsSync(sessionsDir)) return []; const entries = await readdir(sessionsDir, { withFileTypes: true }); const dirs = entries .filter((entry) => entry.isDirectory() || entry.isSymbolicLink()) .map((entry) => join(sessionsDir, entry.name)); const dirFiles = await mapWithConcurrency( dirs, MAX_CONCURRENT_SESSION_DISCOVERY_LOADS, async (dir) => { try { return (await readdir(dir)).filter((file) => file.endsWith(".jsonl")).map((file) => join(dir, file)); } catch { return []; } }, abortSignal, ); const allFiles = dirFiles.flat(); const candidates = await mapWithConcurrency( allFiles, MAX_CONCURRENT_SESSION_DISCOVERY_LOADS, async (path): Promise => { try { return { path, stats: await stat(path) }; } catch { return { path }; } }, abortSignal, ); candidates.sort( (a, b) => (b.stats?.mtimeMs ?? Number.NEGATIVE_INFINITY) - (a.stats?.mtimeMs ?? Number.NEGATIVE_INFINITY) || basename(b.path).localeCompare(basename(a.path)), ); const totalFiles = candidates.length; let loaded = 0; let firstCandidateLoaded = false; const partialSessions: SessionInfo[] = []; const results = await buildSessionInfosWithConcurrency( candidates, (info, index) => { loaded++; if (index === 0) firstCandidateLoaded = true; if (info) partialSessions.push(info); const publishPartial = firstCandidateLoaded && (index === 0 || loaded % ALL_SESSION_LIST_PUBLISH_INTERVAL === 0 || loaded === totalFiles); progress?.(loaded, totalFiles, publishPartial ? sortSessionInfos([...partialSessions]) : undefined); }, abortSignal, ); return sortSessionInfos(results.filter((info): info is SessionInfo => info !== null)); } catch { abortSignal?.throwIfAborted(); return []; } } }