Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
108 kB · 3309 lines
TypeScript
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310import type { AgentMessage } from "@earendil-works/pi-agent-core";import type { AssistantMessage, ImageContent, Message, ServiceTier, TextContent, Usage } from "@earendil-works/pi-ai";import { randomUUID } from "crypto";import { appendFileSync, chownSync, closeSync, existsSync, fstatSync, mkdirSync, openSync, readdirSync, readFileSync, readSync, statSync, writeSync,} from "fs";import { readdir, readFile, stat } from "fs/promises";import { basename, dirname, join, resolve } from "path";import { v7 as uuidv7 } from "uuid";import { getAgentDir as getDefaultAgentDir, getSessionsDir } from "../config.js";import { realpathIfPresentSync, writeFileAtomicSync } from "../utils/atomic-file.js";import { readBytesSync, readFirstLineSync, readLinesAsBuffers } from "../utils/file-lines.js";import { captureGitContext, type GitContext, gitContextsEqual } from "../utils/git.js";import { Semaphore } from "../utils/semaphore.js";import { type BashExecutionMessage, type CustomMessage, createBranchSummaryMessage, createCompactionSummaryMessage, createCustomMessage,} from "./messages.js";import { canonicalSessionCatalogDir, isSessionCatalogStoreEnabled, openSessionCatalogStore, type SessionCatalogRow, type SessionCatalogStore,} from "./session-catalog-db.js";import { readSessionEntryStructure, type SessionEntryStructure } from "./session-entry-structure.js";import { matchesSavedSessionSelector, normalizeSessionId } from "./session-id.js";import { addAssistantUsage, cloneUsage, emptyUsage, type SessionUsageSummary, sessionUsageSummaryFrom, subtractAssistantUsage,} from "./usage.js";
export const CURRENT_SESSION_VERSION = 3;const SESSION_LIST_SEARCH_TEXT_MAX_CHARS = 64 * 1024;const SESSION_LIST_PARSE_MAX_LINE_CHARS = 1024 * 1024;const SESSION_LIST_LARGE_MESSAGE_PREVIEW_MAX_CHARS = 256;const SESSION_STREAMING_LOAD_THRESHOLD_BYTES = 128 * 1024 * 1024;const SESSION_ASYNC_PARSE_YIELD_BYTES = 4 * 1024 * 1024;const SESSION_SCAN_CONCURRENCY = 16;
// Entry types that can represent user intent (vs. daemon bookkeeping like// session_state/agent_status/git_state/child_usage_attributed). Used by// hasUserContent to decide whether a message-less draft is safe to discard.const CONTENT_ENTRY_TYPES = new Set([ "message", "custom_message", "custom", "model_change", "thinking_level_change", "service_tier_change", "session_info", "label", "compaction", "branch_summary",]);
function statMetadataIfPresent(path: string): { mode: number; uid: number; gid: number } | undefined { try { const { mode, uid, gid } = statSync(path); return { mode: mode & 0o777, uid, gid }; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; throw error; }}
export interface SessionHeader { type: "session"; version?: number; // v1 sessions don't have this id: string; timestamp: string; cwd: string; parentSession?: string; rlmDepth?: number; git?: GitContext;}
export interface NewSessionOptions { id?: string; parentSession?: string; rlmDepth?: number;}
export type SessionPersistListener = (sessionFile: string) => void;
export interface SessionEntryBase { type: string; id: string; parentId: string | null; timestamp: string;}
export interface SessionMessageEntry extends SessionEntryBase { type: "message"; message: AgentMessage;}
type AssistantSessionMessageEntry = SessionMessageEntry & { message: AssistantMessage };
export interface ThinkingLevelChangeEntry extends SessionEntryBase { type: "thinking_level_change"; thinkingLevel: string;}
export interface ServiceTierChangeEntry extends SessionEntryBase { type: "service_tier_change"; serviceTier: ServiceTier;}
export interface ModelChangeEntry extends SessionEntryBase { type: "model_change"; provider: string; modelId: string;}
export interface CompactionEntry<T = unknown> extends SessionEntryBase { type: "compaction"; summary: string; firstKeptEntryId: string; tokensBefore: number; details?: T; fromHook?: boolean; customInstructions?: string; usage?: Usage; /** Harness digest snapshot taken at compaction time; rendered before the summary in LLM context. */ harnessDigest?: string;}
export interface BranchSummaryEntry<T = unknown> extends SessionEntryBase { type: "branch_summary"; fromId: string; summary: string; details?: T; fromHook?: boolean; usage?: Usage;}
export interface CustomEntry<T = unknown> extends SessionEntryBase { type: "custom"; customType: string; data?: T;}
/** * Records usage folded into a parent assistant message after an RLM child run. * The child usage is kept separately so audit/UI code can explain why the * parent turn's aggregate usage exceeds the parent model response itself. */export interface ChildUsageAttributionEntry extends SessionEntryBase { type: "child_usage_attributed"; targetId: string; childUsage: Usage; aggregateUsage: Usage; origin?: "spawn_task" | "agent_message" | "direct_user";}
export interface LabelEntry extends SessionEntryBase { type: "label"; targetId: string; label: string | undefined;}
export interface SessionInfoEntry extends SessionEntryBase { type: "session_info"; name?: string;}
export type SessionStateStatus = "active" | "archived" | "crash";
export interface SessionState { status: SessionStateStatus;}
export interface SessionStateEntry extends SessionEntryBase { type: "session_state"; state: SessionState;}
export type AgentTaskState = "needs_input" | "completed" | "error";
export interface AgentStatus { summary: string; taskState?: AgentTaskState; basedOnMessageCount: number;}
export interface AgentStatusEntry extends SessionEntryBase { type: "agent_status"; status: AgentStatus;}
export interface GitStateEntry extends SessionEntryBase { type: "git_state"; git: GitContext;}
export interface CustomMessageEntry<T = unknown> extends SessionEntryBase { type: "custom_message"; customType: string; content: string | (TextContent | ImageContent)[]; details?: T; display: boolean;}
export type SessionEntry = | SessionMessageEntry | ThinkingLevelChangeEntry | ServiceTierChangeEntry | ModelChangeEntry | CompactionEntry | BranchSummaryEntry | CustomEntry | ChildUsageAttributionEntry | CustomMessageEntry | LabelEntry | SessionInfoEntry | SessionStateEntry | AgentStatusEntry | GitStateEntry;
export type FileEntry = SessionHeader | SessionEntry;
export interface SessionTreeFlatNode { entry: SessionEntry; label?: string; labelTimestamp?: string;}
export interface SessionTreeNode extends SessionTreeFlatNode { children: SessionTreeNode[];}
export interface SessionContext { messages: AgentMessage[]; thinkingLevel: string; serviceTier: ServiceTier; model: { provider: string; modelId: string } | null;}
export interface SessionModelRef { provider: string; modelId: string;}
export interface SessionInfo { path: string; id: string; cwd: string; name?: string; state?: SessionState; /** Last model the session ran with, from model_change entries and assistant messages. */ model?: SessionModelRef; parentSessionPath?: string; rlmDepth: number; created: Date; modified: Date; messageCount: number; firstMessage: string; allMessagesText: string; agentStatus?: AgentStatus; usage?: SessionUsageSummary;}
export type ReadonlySessionManager = Pick< SessionManager, | "getCwd" | "getSessionDir" | "getSessionId" | "getSessionFile" | "getLeafId" | "getLeafEntry" | "getEntry" | "getLabel" | "getBranch" | "getHeader" | "getEntries" | "getTree" | "getSessionName">;
function createSessionId(): string { return uuidv7();}
function getSessionFilePath(sessionDir: string, sessionId: string): string { return join(sessionDir, `${sessionId}.jsonl`);}
function createUniqueSessionFileTarget(sessionDir: string): { sessionId: string; sessionFile: string } { for (let i = 0; i < 100; i++) { const sessionId = createSessionId(); const sessionFile = getSessionFilePath(sessionDir, sessionId); if (!existsSync(sessionFile)) { return { sessionId, sessionFile }; } } throw new Error("Unable to create a unique session file");}
// writeSync may return a short count without throwing; a torn line would// corrupt the forked JSONL on reopen, so loop until the whole buffer lands.function writeAllSync(descriptor: number, data: string, path: string): void { const bytes = Buffer.from(data, "utf8"); let offset = 0; while (offset < bytes.length) { const written = writeSync(descriptor, bytes, offset, bytes.length - offset); if (written <= 0) throw new Error(`Short write persisting ${path}`); offset += written; }}
export function getSessionArtifactsRoot(sessionDir: string): string { return join(dirname(sessionDir), "session-artifacts");}
export function getSessionArtifactPath(sessionDir: string, sessionId: string): string { return join(getSessionArtifactsRoot(sessionDir), sessionId);}
export function getSessionArtifactPathForFile(sessionFile: string, sessionId?: string): string { return getSessionArtifactPath(dirname(sessionFile), sessionId ?? basename(sessionFile).replace(/\.jsonl$/, ""));}
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; } return randomUUID();}
function migrateV1ToV2(entries: FileEntry[]): void { const ids = new Set<string>(); 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;
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; } } }}
function migrateV2ToV3(entries: FileEntry[]): void { for (const entry of entries) { if (entry.type === "session") { entry.version = 3; continue; }
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"; } } }}
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;}
export function migrateSessionEntries(entries: FileEntry[]): void { migrateToCurrentVersion(entries);}
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. } }
applyChildUsageAttributions(entries); return entries;}
function applyChildUsageAttributions(entries: FileEntry[]): void { const assistantEntriesById = new Map<string, AssistantSessionMessageEntry>(); for (const entry of entries) { if (entry.type === "message" && entry.message.role === "assistant") { assistantEntriesById.set(entry.id, entry as AssistantSessionMessageEntry); } }
for (const entry of entries) { if (entry.type !== "child_usage_attributed") continue; const target = assistantEntriesById.get(entry.targetId); if (!target) continue; target.message.usage = cloneUsage(entry.aggregateUsage); }}
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;}
export function buildSessionContext( entries: SessionEntry[], leafId?: string | null, byId?: Map<string, SessionEntry>,): SessionContext { if (!byId) { byId = new Map<string, SessionEntry>(); for (const entry of entries) { byId.set(entry.id, entry); } }
let leaf: SessionEntry | undefined; if (leafId === null) { return { messages: [], thinkingLevel: "off", serviceTier: "default", model: null }; } if (leafId) { leaf = byId.get(leafId); } if (!leaf) { leaf = entries[entries.length - 1]; }
if (!leaf) { return { messages: [], thinkingLevel: "off", serviceTier: "default", model: null }; }
// push+reverse, not unshift-per-entry: unshift is O(n), making this O(n^2) on long sessions. const path: SessionEntry[] = []; let current: SessionEntry | undefined = leaf; while (current) { path.push(current); current = current.parentId ? byId.get(current.parentId) : undefined; } path.reverse();
let thinkingLevel = "off"; let serviceTier: ServiceTier = "default"; let model: { provider: string; modelId: string } | null = null; let compaction: CompactionEntry | null = null;
for (const entry of path) { if (entry.type === "thinking_level_change") { thinkingLevel = entry.thinkingLevel; } else if (entry.type === "service_tier_change") { serviceTier = entry.serviceTier; } 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 }; } else if (entry.type === "compaction") { compaction = entry; } }
// Build messages and collect corresponding entries // When there's a compaction, model context remains summary-first while the // summary records where clients should present it among retained messages. const messages: AgentMessage[] = [];
const appendMessage = (entry: SessionEntry, target = messages) => { if (entry.type === "message") { target.push(entry.message); } else if (entry.type === "custom_message") { target.push( createCustomMessage(entry.customType, entry.content, entry.display, entry.details, entry.timestamp), ); } else if (entry.type === "branch_summary" && entry.summary) { target.push(createBranchSummaryMessage(entry.summary, entry.fromId, entry.timestamp)); } };
if (compaction) { const compactionIdx = path.findIndex((e) => e.type === "compaction" && e.id === compaction.id);
// Collect kept messages (before compaction, starting from firstKeptEntryId). // The context remains summary-first for the model; retainedMessageCount records // the exact chronological presentation boundary for clients. const retainedMessages: AgentMessage[] = []; let foundFirstKept = false; for (let i = 0; i < compactionIdx; i++) { const entry = path[i]; if (entry.id === compaction.firstKeptEntryId) { foundFirstKept = true; } if (foundFirstKept) { appendMessage(entry, retainedMessages); } }
messages.push( createCompactionSummaryMessage( compaction.summary, compaction.tokensBefore, compaction.timestamp, compaction.customInstructions, retainedMessages.length, compaction.harnessDigest, ), ...retainedMessages, );
for (let i = compactionIdx + 1; i < path.length; i++) { const entry = path[i]; appendMessage(entry); } } else { for (const entry of path) { appendMessage(entry); } }
return { messages, thinkingLevel, serviceTier, model };}
export function getDefaultSessionDir(_cwd: string, agentDir: string = getDefaultAgentDir()): string { const sessionDir = getSessionsDir(agentDir); if (!existsSync(sessionDir)) { mkdirSync(sessionDir, { recursive: true }); } return sessionDir;}
// Decode per line off a Buffer: toString("utf8") on a whole large file is far slower// (one giant UTF-16 string). Splitting on 0x0a is UTF-8-safe.function appendEntryFromBuffer(entries: FileEntry[], buffer: Buffer, start = 0, end = buffer.length): void { if (end <= start) return; try { entries.push(JSON.parse(buffer.toString("utf8", start, end)) as FileEntry); } catch { // Skip malformed or blank lines. }}
function parseEntriesFromBuffer(buffer: Buffer): FileEntry[] { const entries: FileEntry[] = []; let start = 0; while (start < buffer.length) { let end = buffer.indexOf(0x0a, start); if (end === -1) end = buffer.length; appendEntryFromBuffer(entries, buffer, start, end); start = end + 1; } return entries;}
async function parseEntriesFromBufferAsync(buffer: Buffer): Promise<FileEntry[]> { const entries: FileEntry[] = []; let start = 0; let bytesSinceYield = 0; while (start < buffer.length) { let end = buffer.indexOf(0x0a, start); if (end === -1) end = buffer.length; appendEntryFromBuffer(entries, buffer, start, end); bytesSinceYield += end - start + 1; start = end + 1; if (bytesSinceYield >= SESSION_ASYNC_PARSE_YIELD_BYTES) { bytesSinceYield = 0; await new Promise<void>((resolve) => setImmediate(resolve)); } } return entries;}
// Crash damage (torn tail, zero-filled append) poisons the NEXT append into the// same physical line, so the file is repaired once at open, not tolerated in memory.const REPAIR_SUSPICION_WINDOW_BYTES = 1024 * 1024;
// A bounded tail read gates the full repair scan: clean opens stay O(window).function tailLooksDamaged(targetPath: string): boolean { let descriptor: number; try { descriptor = openSync(targetPath, "r"); } catch { return false; } try { const size = fstatSync(descriptor).size; if (size === 0) return false; const windowBytes = Math.min(size, REPAIR_SUSPICION_WINDOW_BYTES); const window = Buffer.allocUnsafe(windowBytes); readSync(descriptor, window, 0, windowBytes, size - windowBytes); if (window.includes(0)) return true; if (window[windowBytes - 1] !== 0x0a) return true; const previousNewline = window.lastIndexOf(0x0a, windowBytes - 2); // No boundary inside the window: the final line exceeds it; scan to be sure. if (previousNewline === -1 && windowBytes < size) return true; const lastLine = window.subarray(previousNewline + 1, windowBytes - 1); // A blank final line is benign (the loader skips it) and appends stay safe. return lastLine.length > 0 && !parsesAsJson(lastLine); } catch { return true; } finally { closeSync(descriptor); }}
function repairJsonlDamage(filePath: string): void { const targetPath = realpathIfPresentSync(filePath); if (!tailLooksDamaged(targetPath)) return; let buffer: Buffer; let snapshot: { size: number; mtimeMs: number }; try { buffer = readFileSync(targetPath); const measured = statSync(targetPath); snapshot = { size: measured.size, mtimeMs: measured.mtimeMs }; } catch { return; } if (buffer.length === 0 || snapshot.size !== buffer.length) return; const keptLines: Buffer[] = []; let recoveredNulLines = 0; let droppedLines = 0; let repairedTail = false; let dirty = false; let start = 0; while (start < buffer.length) { let end = buffer.indexOf(0x0a, start); const terminated = end !== -1; if (!terminated) end = buffer.length; let lineStart = start; while (lineStart < end && buffer[lineStart] === 0) lineStart++; const line = buffer.subarray(lineStart, end); if (lineStart > start) { dirty = true; if (line.length > 0 && parsesAsJson(line)) { keptLines.push(line); recoveredNulLines++; } else { droppedLines++; } } else if (!terminated) { // An unterminated tail merges with the next append: re-terminate or truncate. dirty = true; if (line.length > 0 && parsesAsJson(line)) { keptLines.push(line); repairedTail = true; } else { droppedLines++; } } else if (end + 1 >= buffer.length && line.length > 0 && !parsesAsJson(line)) { dirty = true; droppedLines++; } else { keptLines.push(line); } start = end + 1; } if (!dirty) return; const metadata = statMetadataIfPresent(targetPath); const content = keptLines.length > 0 ? `${keptLines.map((kept) => kept.toString("utf8")).join("\n")}\n` : ""; try { writeFileAtomicSync(targetPath, content, { ...(metadata === undefined ? {} : { mode: metadata.mode }), beforeRename: (tempPath) => { if (metadata !== undefined) chownSync(tempPath, metadata.uid, metadata.gid); // A concurrent appender wins; skipping the repair is safe (next open retries). const current = statSync(targetPath); if (current.size !== snapshot.size || current.mtimeMs !== snapshot.mtimeMs) { throw new RepairSupersededError(); } }, }); } catch (error) { if (error instanceof RepairSupersededError) return; throw error; } console.error( `Repaired crash damage in ${targetPath}: recovered ${recoveredNulLines} zero-filled line(s), dropped ${droppedLines} unrecoverable line(s)${repairedTail ? ", restored the trailing newline" : ""}`, );}
class RepairSupersededError extends Error {}
function parsesAsJson(line: Buffer): boolean { try { JSON.parse(line.toString("utf8")); return true; } catch { return false; }}
function finalizeLoadedEntries(entries: FileEntry[]): FileEntry[] { if (entries.length === 0) return entries; const header = entries[0]; if (header.type !== "session" || typeof (header as any).id !== "string") { return []; } applyChildUsageAttributions(entries); return entries;}
export function loadEntriesFromFile(filePath: string): FileEntry[] { if (!existsSync(filePath)) return []; return finalizeLoadedEntries(parseEntriesFromBuffer(readFileSync(filePath)));}
/** * Read only the session header entry. Callers that already hold the session * summary (daemon attach, selector resolution) use it to avoid parsing a whole * transcript just to learn the session's cwd and id. Returns undefined when the * first physical line is not a session header, so callers fall back to the full * load for legacy files. */export function loadSessionHeaderEntry(filePath: string): FileEntry[] | undefined { if (!existsSync(filePath)) return undefined; const firstLine = readFirstLineSync(filePath); if (!firstLine) return undefined; try { const entry = JSON.parse(firstLine) as FileEntry; return entry && typeof entry === "object" && entry.type === "session" ? [entry] : undefined; } catch { return undefined; }}
// Async loader for the daemon: reads off the event loop and yields while parsing so a// large load doesn't freeze other sessions. Large files stream to avoid retaining both// the full input Buffer and the parsed entry graph at the same time.export async function loadEntriesFromFileAsync( filePath: string, options: { streamThresholdBytes?: number } = {},): Promise<FileEntry[]> { if (!existsSync(filePath)) return []; const streamThresholdBytes = options.streamThresholdBytes ?? SESSION_STREAMING_LOAD_THRESHOLD_BYTES; if ((await stat(filePath)).size < streamThresholdBytes) { return finalizeLoadedEntries(await parseEntriesFromBufferAsync(await readFile(filePath))); }
const entries: FileEntry[] = []; let bytesSinceYield = 0; for await (const line of readLinesAsBuffers(filePath)) { appendEntryFromBuffer(entries, line); bytesSinceYield += line.length + 1; if (bytesSinceYield >= SESSION_ASYNC_PARSE_YIELD_BYTES) { bytesSinceYield = 0; await new Promise<void>((resolve) => setImmediate(resolve)); } } return finalizeLoadedEntries(entries);}
function readSessionHeader(filePath: string): Partial<SessionHeader> | undefined { const firstLine = readFirstLineSync(filePath); if (!firstLine) { return undefined; } return JSON.parse(firstLine) as Partial<SessionHeader>;}
function isValidRlmDepth(value: unknown): value is number { return typeof value === "number" && Number.isSafeInteger(value) && value >= 0;}
export function resolveSessionRlmDepth( header: { rlmDepth?: number; parentSession?: string }, sessionPath: string,): number { return resolveLegacySessionRlmDepth(header, sessionPath, new Set()) ?? legacyChildDepthFromPath(sessionPath);}
function resolveLegacySessionRlmDepth( header: { rlmDepth?: number; parentSession?: string }, sessionPath: string, visitedPaths: Set<string>,): number | undefined { if (isValidRlmDepth(header.rlmDepth)) { return header.rlmDepth; } if (!header.parentSession) { return 0; }
const resolvedSessionPath = resolve(sessionPath); if (visitedPaths.has(resolvedSessionPath)) { return undefined; } visitedPaths.add(resolvedSessionPath);
const pathDepth = legacyChildDepthFromPath(sessionPath); const parentSessionPath = resolve(dirname(sessionPath), header.parentSession); try { const parentHeader = readSessionHeader(parentSessionPath); if (parentHeader) { const parentDepth = resolveLegacySessionRlmDepth(parentHeader, parentSessionPath, visitedPaths); if (parentDepth !== undefined) { return pathDepth > 0 ? parentDepth + 1 : parentDepth; } } } catch { // Fall back to artifact ancestry for unavailable or invalid legacy parents. } finally { visitedPaths.delete(resolvedSessionPath); } return pathDepth;}
function legacyChildDepthFromPath(sessionPath: string): number { let depth = 0; for (const segment of dirname(sessionPath) .split(/[\\/]+/) .reverse()) { if (!/^sub-[0-9a-f]{8}$/.test(segment)) { break; } depth += 1; } return depth;}
function deriveChildRlmDepth(parentHeader: Partial<SessionHeader> | undefined): number | undefined { const depth = parentHeader?.rlmDepth; return isValidRlmDepth(depth) && depth < Number.MAX_SAFE_INTEGER ? depth + 1 : undefined;}
function rootRlmDepthFromEnv(): number { const value = process.env.RLM_DEPTH; if (value === undefined || value === "") { return 0; } const parsed = Number(value); if (!/^\d+$/.test(value) || !isValidRlmDepth(parsed)) { throw new Error("RLM_DEPTH must be a non-negative integer"); } return parsed;}
function isValidSessionFile(filePath: string): boolean { try { const header = readSessionHeader(filePath); return header?.type === "session" && typeof header.id === "string"; } catch { return false; }}
export function findMostRecentSession(sessionDir: string): string | null { try { const files = readdirSync(sessionDir) .filter((f) => f.endsWith(".jsonl")) .map((f) => join(sessionDir, f)) .filter(isValidSessionFile) .map((path) => ({ path, mtime: statSync(path).mtime })) .sort((a, b) => b.mtime.getTime() - a.mtime.getTime());
return files[0]?.path || null; } catch { return null; }}
function normalizeCwd(cwd: string): string { return resolve(cwd);}
export interface DirectSessionMatch { path: string; /** Header cwd of the matched transcript. */ cwd: string; /** True when the transcript belongs to `cwd`, mirroring `SessionManager.list` filtering. */ local: boolean;}
/** * Resolve a saved-session selector against file names instead of the catalog scan * `SessionManager.list`/`listAll` perform. Transcripts are named `<id>.jsonl`, so a * selector that uniquely identifies one file costs a directory listing plus one header * read, while the scanning resolver stays the fallback for unknown and ambiguous * selectors (and for sessions whose header disagrees with the file name). */export function findSessionFileBySelector( selector: string, sessionDir: string, cwd: string,): DirectSessionMatch | undefined { let names: string[]; try { names = readdirSync(sessionDir); } catch { return undefined; }
let exact: string | undefined; const partial: string[] = []; for (const name of names) { if (!name.endsWith(".jsonl")) continue; const id = name.slice(0, -".jsonl".length); if (normalizeSessionId(id) === normalizeSessionId(selector)) { exact = name; break; } if (matchesSavedSessionSelector(id, selector)) partial.push(name); } const candidates = exact === undefined ? partial : [exact]; if (candidates.length !== 1) return undefined;
const path = join(sessionDir, candidates[0]!); const header = readSessionHeader(path); // A transcript whose header is missing, damaged, or keyed to another id is not // addressable by file name; the scanning resolver handles that case. if (header?.type !== "session" || header.id !== candidates[0]!.slice(0, -".jsonl".length)) { return undefined; } if (typeof header.cwd !== "string") return undefined; return { path, cwd: header.cwd, local: normalizeCwd(header.cwd) === normalizeCwd(cwd) };}
function sessionInfoMatchesCwd(session: SessionInfo, cwd: string): boolean { return !!session.cwd && normalizeCwd(session.cwd) === normalizeCwd(cwd);}
function sessionHeaderMatchesCwd(header: Partial<SessionHeader> | undefined, cwd: string): boolean { return ( header?.type === "session" && typeof header.id === "string" && typeof header.cwd === "string" && normalizeCwd(header.cwd) === normalizeCwd(cwd) );}
export function findMostRecentSessionForCwd(sessionDir: string, cwd: string): string | null { try { const files = readdirSync(sessionDir) .filter((f) => f.endsWith(".jsonl")) .map((f) => join(sessionDir, f)) .map((path) => { try { const header = readSessionHeader(path); if (!sessionHeaderMatchesCwd(header, cwd)) { return undefined; } return { path, mtime: statSync(path).mtime }; } catch { return undefined; } }) .filter((entry): entry is { path: string; mtime: Date } => entry !== undefined) .sort((a, b) => b.mtime.getTime() - a.mtime.getTime());
return files[0]?.path || null; } catch { 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 normalizeSessionStateStatus(value: unknown): SessionStateStatus | undefined { if (value === "active" || value === "archived" || value === "crash") { return value; } if (value === "hidden" || value === "sleep") { return "archived"; } return undefined;}
function updateLastActivityTime(lastActivityTime: number | undefined, entry: FileEntry): number | undefined { if (entry.type !== "message") { return lastActivityTime; }
const message = (entry as SessionMessageEntry).message; if (!isMessageWithContent(message)) { return lastActivityTime; } if (message.role !== "user" && message.role !== "assistant") { return lastActivityTime; }
const msgTimestamp = (message as { timestamp?: number }).timestamp; if (typeof msgTimestamp === "number") { return Math.max(lastActivityTime ?? 0, msgTimestamp); }
const entryTimestamp = (entry as SessionEntryBase).timestamp; if (typeof entryTimestamp === "string") { const t = new Date(entryTimestamp).getTime(); if (!Number.isNaN(t)) { return Math.max(lastActivityTime ?? 0, t); } }
return lastActivityTime;}
function getSessionModifiedDateFromLastActivity( lastActivityTime: number | undefined, header: SessionHeader, statsMtime: Date,): Date { if (typeof lastActivityTime === "number" && lastActivityTime > 0) { return new Date(lastActivityTime); }
const headerTime = typeof header.timestamp === "string" ? new Date(header.timestamp).getTime() : NaN; return !Number.isNaN(headerTime) ? new Date(headerTime) : statsMtime;}
function appendCappedSearchText(current: string, text: string): string { if (!text || current.length >= SESSION_LIST_SEARCH_TEXT_MAX_CHARS) { return current; } const next = current ? ` ${text}` : text; return current + next.slice(0, SESSION_LIST_SEARCH_TEXT_MAX_CHARS - current.length);}
function looksLikeMessageEntry(line: string): boolean { return line.includes('"type":"message"') || line.includes('"type": "message"');}
function extractJsonStringPropertyPrefix( text: string, propertyName: string, maxChars: number, startIndex = 0,): string | undefined { const propertyIndex = text.indexOf(`"${propertyName}"`, startIndex); if (propertyIndex < 0) { return undefined; } let index = propertyIndex + propertyName.length + 2; while (index < text.length && /\s/.test(text[index] ?? "")) index++; if (text[index] !== ":") { return undefined; } index++; while (index < text.length && /\s/.test(text[index] ?? "")) index++; if (text[index] !== '"') { return undefined; } index++;
let result = ""; let escaped = false; for (; index < text.length && result.length < maxChars; index++) { const char = text[index]; if (escaped) { result += char; escaped = false; continue; } if (char === "\\") { escaped = true; continue; } if (char === '"') { break; } result += char; } return result;}
function extractOversizedMessageSummary(line: string): { role?: string; timestamp?: number; textPreview?: string;} { const timestampText = extractJsonStringPropertyPrefix(line, "timestamp", 64); const timestamp = timestampText ? new Date(timestampText).getTime() : NaN; const messageIndex = line.indexOf('"message"'); const role = messageIndex >= 0 ? extractJsonStringPropertyPrefix(line, "role", 64, messageIndex) : extractJsonStringPropertyPrefix(line, "role", 64); let textPreview: string | undefined; if (messageIndex >= 0) { textPreview = extractJsonStringPropertyPrefix(line, "content", SESSION_LIST_LARGE_MESSAGE_PREVIEW_MAX_CHARS, messageIndex) ?? extractJsonStringPropertyPrefix(line, "text", SESSION_LIST_LARGE_MESSAGE_PREVIEW_MAX_CHARS, messageIndex); } return { role, ...(Number.isNaN(timestamp) ? {} : { timestamp }), ...(textPreview ? { textPreview } : {}), };}
interface SessionScanAccumulator { header?: SessionHeader; model?: SessionModelRef; /** The first parsed entry was not a session header; appends cannot repair this. */ invalid: boolean; messageCount: number; firstMessage: string; allMessagesText: string; name?: string; state?: SessionState; agentStatus?: AgentStatus; lastActivityTime?: number; // Fold attribution aggregates like the loader: either disk representation cancels to the same own spend. assistantUsageById: Map<string, Usage>; attributedChildUsage: Usage; summarizationUsage: Usage;}
function entryTimestampMs(entry: FileEntry): number | null { const timestamp = (entry as { timestamp?: unknown }).timestamp; if (typeof timestamp !== "string") return null; const parsed = Date.parse(timestamp); return Number.isFinite(parsed) ? parsed : null;}
function entryIndexFields(entry: FileEntry): Omit<SessionEntryStructure, "offset" | "length"> | null | undefined { // The session header is not a tree node; a lazy open reads it from line one. if (entry.type === "session") return undefined; const id = (entry as { id?: unknown }).id; // An entry without an id (a v1 transcript before migration) cannot be indexed, // and a partial index would be worse than none: null makes it unusable. if (typeof id !== "string" || !id) return null; const parentId = (entry as { parentId?: unknown }).parentId; const message = entry.type === "message" ? (entry as SessionMessageEntry).message : undefined; const role = message && typeof message.role === "string" ? message.role : undefined; return { id, parentId: typeof parentId === "string" && parentId ? parentId : null, type: entry.type, timestampMs: entryTimestampMs(entry), role: role ?? null, };}
/** * Structure of an entry too large to parse during a listing refresh: the fields * live in the line prefix, so a bounded scan reads them without the payload. */function oversizedEntryIndexFields(line: string): Omit<SessionEntryStructure, "offset" | "length"> | null { const head = line.slice(0, 4096); const readField = (name: string): string | undefined => { const match = new RegExp(`"${name}"\\s*:\\s*"`).exec(head); if (!match) return undefined; const start = match.index + match[0].length; const end = head.indexOf('"', start); return end === -1 ? undefined : head.slice(start, end); }; const id = readField("id"); const type = readField("type"); if (!id || !type) return null; const timestamp = readField("timestamp"); const parsedTimestamp = timestamp === undefined ? Number.NaN : Date.parse(timestamp); return { id, parentId: readField("parentId") ?? null, type, timestampMs: Number.isFinite(parsedTimestamp) ? parsedTimestamp : null, role: readField("role") ?? null, };}
interface SessionScanState { fileSize: number; mtimeMs: number; /** Identity of the scanned file: a rename rewrite replaces the inode and invalidates the resume state. */ dev: number; ino: number; /** Bytes consumed as complete newline-terminated lines; the resume point for the next scan. */ offset: number; /** Last bytes of the consumed prefix; a resume only proceeds while the file still starts with them. */ tail: Buffer; acc: SessionScanAccumulator; info: SessionInfo | null; /** Usage entries counted against the retained bound at the last store. */ accountedUsageEntries: number;}
const SESSION_SCAN_RESUME_TAIL_BYTES = 16;// The persisted accumulator is a cache of the scan; past this size the catalog// row keeps the summary only and the next append pays a full rescan.const SESSION_CATALOG_MAX_RESUME_BYTES = 256 * 1024;const NEWLINE_BUFFER = Buffer.from("\n");// Memory/performance budget: whole-state LRU eviction can force a full catalog rescan every refresh.// Keep large families (~2k sessions, 150k usage entries) and growth headroom resident.const SESSION_SCAN_MAX_RETAINED_USAGE_ENTRIES = 400_000;
// Session files are append-only between whole-file rewrites, so scans resume// from the last consumed byte offset; rewrites are detected by shrink,// same-size mtime change, replaced inode, or a changed prefix tail. In-place// interior edits that defeat all four are outside the writer model.const sessionScanStates = new Map<string, SessionScanState>();// Scans of a path run one at a time; a later caller chains its own pass (its// stat sees every prior append) instead of joining an earlier scan's result.const sessionScanQueue = new Map<string, Promise<SessionInfo | null>>();
// One catalog database per session directory, opened lazily. A directory whose// store cannot be opened keeps scanning in process for the rest of the run.const sessionCatalogStores = new Map<string, SessionCatalogStore>();
function sessionCatalogStoreForDir(directory: string): SessionCatalogStore | undefined { if (!isSessionCatalogStoreEnabled()) return undefined; const dir = canonicalSessionCatalogDir(directory); const open = sessionCatalogStores.get(dir); if (open) return open; // A failed open is not cached: the directory may appear later in the process. if (!existsSync(dir)) return undefined; const store = openSessionCatalogStore(dir); if (store) sessionCatalogStores.set(dir, store); return store;}
function sessionCatalogStoreFor(filePath: string): SessionCatalogStore | undefined { return sessionCatalogStoreForDir(dirname(filePath));}
/** Close every open catalog store; the next lookup reopens it. */export function closeSessionCatalogStores(): void { for (const store of sessionCatalogStores.values()) { store?.close(); } sessionCatalogStores.clear();}
/** Drops every in-process scan state, so the next scan reads the catalog or the file. */export function forgetSessionScanStates(): void { for (const key of [...sessionScanStates.keys()]) { dropSessionScanState(key); }}
export function readSessionInfo(filePath: string): Promise<SessionInfo | null> { const previous = sessionScanQueue.get(filePath); const scan = previous ? previous.then( () => scanSessionInfo(filePath), () => scanSessionInfo(filePath), ) : scanSessionInfo(filePath); sessionScanQueue.set(filePath, scan); void scan.finally(() => { if (sessionScanQueue.get(filePath) === scan) sessionScanQueue.delete(filePath); }); return scan;}
let retainedUsageEntries = 0;
function dropSessionScanState(filePath: string): void { const state = sessionScanStates.get(filePath); if (!state) return; retainedUsageEntries -= state.accountedUsageEntries; sessionScanStates.delete(filePath);}
/** Refresh LRU recency and enforce the retained-usage bound. */function storeSessionScanState(filePath: string, state: SessionScanState): void { dropSessionScanState(filePath); state.accountedUsageEntries = state.acc.assistantUsageById.size; retainedUsageEntries += state.accountedUsageEntries; sessionScanStates.set(filePath, state); for (const key of sessionScanStates.keys()) { if (retainedUsageEntries <= SESSION_SCAN_MAX_RETAINED_USAGE_ENTRIES) break; dropSessionScanState(key); }}
function createSessionScanAccumulator(): SessionScanAccumulator { return { invalid: false, messageCount: 0, firstMessage: "", allMessagesText: "", assistantUsageById: new Map<string, Usage>(), attributedChildUsage: emptyUsage(), summarizationUsage: emptyUsage(), };}
/** * Persisted catalog summary. `allMessagesText` is deliberately absent: it is a * multi-kilobyte-per-session search blob, and storing it made a listing answered * from the database an order of magnitude heavier than the identity fields the * listing needs. Such a listing searches its name and first message; a live scan * still carries the full text. */interface SerializedCatalogSessionInfo extends Omit<SessionInfo, "path" | "created" | "modified" | "allMessagesText"> { created: number; modified: number;}
function serializeCatalogSessionInfo(info: SessionInfo): string { const payload: SerializedCatalogSessionInfo = { id: info.id, cwd: info.cwd, name: info.name, state: info.state, model: info.model, parentSessionPath: info.parentSessionPath, rlmDepth: info.rlmDepth, created: info.created.getTime(), modified: info.modified.getTime(), messageCount: info.messageCount, firstMessage: info.firstMessage, agentStatus: info.agentStatus, usage: info.usage, }; return JSON.stringify(payload);}
/** Rebuilds a cached summary; anything unexpected reads as a cache miss. */function hydrateCatalogSessionInfo(filePath: string, json: string): SessionInfo | undefined { let parsed: unknown; try { parsed = JSON.parse(json); } catch { return undefined; } if (!parsed || typeof parsed !== "object") return undefined; const candidate = parsed as Partial<SerializedCatalogSessionInfo>; if (typeof candidate.id !== "string" || typeof candidate.cwd !== "string") return undefined; if (typeof candidate.created !== "number" || typeof candidate.modified !== "number") return undefined; if (typeof candidate.messageCount !== "number" || typeof candidate.firstMessage !== "string") return undefined; return { path: filePath, id: candidate.id, cwd: candidate.cwd, name: candidate.name, state: candidate.state, model: candidate.model, parentSessionPath: candidate.parentSessionPath, rlmDepth: typeof candidate.rlmDepth === "number" ? candidate.rlmDepth : 0, created: new Date(candidate.created), modified: new Date(candidate.modified), messageCount: candidate.messageCount, firstMessage: candidate.firstMessage, // Not persisted: a listing answered from the database reports no message // text, so search falls back to the session name and first message. allMessagesText: "", agentStatus: candidate.agentStatus, usage: candidate.usage, };}
interface SerializedSessionScanAccumulator { header?: SessionHeader; model?: SessionModelRef; invalid: boolean; messageCount: number; firstMessage: string; allMessagesText: string; name?: string; state?: SessionState; agentStatus?: AgentStatus; lastActivityTime?: number; assistantUsageById: [string, Usage][]; attributedChildUsage?: Usage; summarizationUsage?: Usage;}
interface SerializedSessionScanState { offset: number; /** Hex of the persisted prefix tail that guards the resume. */ tail: string; acc: SerializedSessionScanAccumulator;}
function serializeSessionScanState(state: SessionScanState): string | undefined { const payload: SerializedSessionScanState = { offset: state.offset, tail: state.tail.toString("hex"), acc: { header: state.acc.header, model: state.acc.model, invalid: state.acc.invalid, messageCount: state.acc.messageCount, firstMessage: state.acc.firstMessage, allMessagesText: state.acc.allMessagesText, name: state.acc.name, state: state.acc.state, agentStatus: state.acc.agentStatus, lastActivityTime: state.acc.lastActivityTime, assistantUsageById: [...state.acc.assistantUsageById.entries()], attributedChildUsage: state.acc.attributedChildUsage, summarizationUsage: state.acc.summarizationUsage, }, }; const json = JSON.stringify(payload); return json.length > SESSION_CATALOG_MAX_RESUME_BYTES ? undefined : json;}
/** * Restores the scan state a previous process persisted for an appended file, so * a cold start reads the appended bytes instead of the whole transcript. The * resume is accepted only when the file still starts with the persisted tail. */
/** Below this size a full parse costs less than the bookkeeping of a lazy open. */const SESSION_LAZY_OPEN_MIN_BYTES = 8 * 1024 * 1024;
/** * Entries that define the current context: everything from the last compaction's * retained boundary to the end of the file, plus the latest entry of every type * whose singleton value the session metadata reads (name, status, model, * thinking level, git state). Everything before it is behind the compaction. */const CONTEXT_WINDOW_SINGLETON_TYPES = new Set([ "model_change", "thinking_level_change", "service_tier_change", "session_info", "session_state", "git_state", "agent_status",]);
interface LazySessionWindow { header: SessionHeader; entries: FileEntry[];}
function lastRecordOfType(records: readonly SessionEntryStructure[], type: string): SessionEntryStructure | undefined { for (let i = records.length - 1; i >= 0; i--) { const record = records[i]!; if (record.type === type) return record; } return undefined;}
function parseEntryLine(line: Buffer): FileEntry | undefined { try { return JSON.parse(line.toString("utf8")) as FileEntry; } catch { return undefined; }}
/** * Reads only the transcript bytes a fresh context needs. The catalog already * knows where every entry is, so this is a handful of reads instead of a full * parse of a session that may be hundreds of megabytes. */function readSessionContextWindow( filePath: string, records: readonly SessionEntryStructure[],): LazySessionWindow | undefined { const headerLine = readFirstLineSync(filePath); const header = headerLine === undefined ? undefined : parseEntryLine(Buffer.from(headerLine, "utf8")); if (!header || header.type !== "session" || (header.version ?? 0) < CURRENT_SESSION_VERSION) return undefined;
let windowStart = records[0]?.offset ?? 0; const lastCompaction = records.filter((record) => record.type === "compaction").at(-1); if (lastCompaction) { const compactionLine = parseEntryLine( readBytesSync(filePath, lastCompaction.offset, lastCompaction.offset + lastCompaction.length), ); const firstKeptId = (compactionLine as CompactionEntry | undefined)?.firstKeptEntryId; const firstKept = firstKeptId ? records.find((record) => record.id === firstKeptId) : undefined; windowStart = firstKept?.offset ?? lastCompaction.offset; }
const lastSingletonByType = new Map<string, SessionEntryStructure>(); for (const record of records) { if (record.offset >= windowStart) continue; if (CONTEXT_WINDOW_SINGLETON_TYPES.has(record.type)) lastSingletonByType.set(record.type, record); } const selected = [...lastSingletonByType.values(), ...records.filter((record) => record.offset >= windowStart)].sort( (left, right) => left.offset - right.offset, ); if (selected.length === 0) return undefined;
const entries: FileEntry[] = []; for (const record of selected) { const entry = parseEntryLine(readBytesSync(filePath, record.offset, record.offset + record.length)); if (!entry || entry.type === "session") return undefined; entries.push(entry); } return { header, entries };}
function restoreSessionScanState( filePath: string, row: SessionCatalogRow, stats: { dev: number; ino: number; size: number; mtimeMs: number },): SessionScanState | undefined { // A rename rewrite replaces the inode: only a same-file append may resume, // or a stale accumulator would fold new bytes onto an old prefix. if (row.resumeJson === null || row.dev !== stats.dev || row.ino !== stats.ino) return undefined; if (stats.size <= row.size) return undefined; let parsed: unknown; try { parsed = JSON.parse(row.resumeJson); } catch { return undefined; } if (!parsed || typeof parsed !== "object") return undefined; const candidate = parsed as Partial<SerializedSessionScanState>; if (typeof candidate.offset !== "number" || typeof candidate.tail !== "string") return undefined; const raw = candidate.acc; if (!raw || typeof raw !== "object") return undefined; const acc = raw as Partial<SerializedSessionScanAccumulator>; if (typeof acc.invalid !== "boolean" || typeof acc.messageCount !== "number") return undefined; if (typeof acc.firstMessage !== "string" || typeof acc.allMessagesText !== "string") return undefined; if (!Array.isArray(acc.assistantUsageById)) return undefined; // A torn final line leaves the consumed prefix short of the recorded size. if (candidate.offset > row.size) return undefined; const tail = Buffer.from(candidate.tail, "hex"); if (tail.length !== Math.min(candidate.offset, SESSION_SCAN_RESUME_TAIL_BYTES)) return undefined; if (!scannedPrefixIntact(filePath, candidate.offset, tail)) return undefined; const usageEntries = acc.assistantUsageById.filter( (entry): entry is [string, Usage] => Array.isArray(entry) && typeof entry[0] === "string" && !!entry[1], ); return { fileSize: row.size, mtimeMs: row.mtimeMs, dev: stats.dev, ino: stats.ino, offset: candidate.offset, tail, acc: { header: acc.header, model: acc.model, invalid: acc.invalid, messageCount: acc.messageCount, firstMessage: acc.firstMessage, allMessagesText: acc.allMessagesText, name: acc.name, state: acc.state, agentStatus: acc.agentStatus, lastActivityTime: acc.lastActivityTime, assistantUsageById: new Map(usageEntries), attributedChildUsage: acc.attributedChildUsage ?? emptyUsage(), summarizationUsage: acc.summarizationUsage ?? emptyUsage(), }, info: null, accountedUsageEntries: 0, };}
function scannedPrefixIntact(filePath: string, offset: number, tail: Buffer): boolean { if (offset === 0) return true; try { const start = Math.max(0, offset - SESSION_SCAN_RESUME_TAIL_BYTES); return readBytesSync(filePath, start, offset).equals(tail); } catch { return false; }}
/** Last bytes of the consumed prefix after appending one line and its newline, copied out of the stream chunk. */function advanceScanTail(tail: Buffer, line: Buffer): Buffer { if (line.length >= SESSION_SCAN_RESUME_TAIL_BYTES - 1) { return Buffer.concat([line.subarray(line.length - (SESSION_SCAN_RESUME_TAIL_BYTES - 1)), NEWLINE_BUFFER]); } const combined = Buffer.concat([tail, line, NEWLINE_BUFFER]); return combined.length <= SESSION_SCAN_RESUME_TAIL_BYTES ? combined : Buffer.from(combined.subarray(combined.length - SESSION_SCAN_RESUME_TAIL_BYTES));}
async function scanSessionInfo(filePath: string, retryOnReplacement = true): Promise<SessionInfo | null> { const store = sessionCatalogStoreFor(filePath); let stats: Awaited<ReturnType<typeof stat>>; try { stats = await stat(filePath); } catch { dropSessionScanState(filePath); store?.remove(filePath); return null; } const previous = sessionScanStates.get(filePath); const sameFile = previous !== undefined && previous.dev === stats.dev && previous.ino === stats.ino; if (previous && sameFile && previous.fileSize === stats.size && previous.mtimeMs === stats.mtimeMs) { storeSessionScanState(filePath, previous); return previous.info; } // A row written by an earlier process answers from the file identity alone. const cached = store?.read(filePath); if ( cached && cached.dev === stats.dev && cached.ino === stats.ino && cached.size === stats.size && cached.mtimeMs === stats.mtimeMs ) { if (cached.summaryJson === null) return null; const cachedInfo = hydrateCatalogSessionInfo(filePath, cached.summaryJson); if (cachedInfo) return cachedInfo; } let state: SessionScanState | undefined; if ( previous && sameFile && stats.size > previous.fileSize && scannedPrefixIntact(filePath, previous.offset, previous.tail) ) { state = previous; } if (!state && cached) { state = restoreSessionScanState(filePath, cached, stats); } state ??= { fileSize: 0, mtimeMs: 0, dev: stats.dev, ino: stats.ino, offset: 0, tail: Buffer.alloc(0), acc: createSessionScanAccumulator(), info: null, accountedUsageEntries: 0, }; // The entry index is collected during the same pass that folds the summary, // so an incremental resume indexes only the bytes it actually consumed. const scanFrom = state.offset; const entryRecords: SessionEntryStructure[] | undefined = store ? [] : undefined; // A line the structure cannot describe makes the whole structure unusable: // a partial index would silently hide entries from a lazy open. let structureUnusable = false; try { const tornTail = await scanSessionLines( filePath, state, stats.size, entryRecords ? (record) => { if (record === null) structureUnusable = true; else entryRecords.push(record); } : undefined, ); state.info = snapshotSessionInfo(state.acc, tornTail, filePath, stats); } catch { dropSessionScanState(filePath); return null; } // A rename rewrite racing the scan can mix two files' bytes into one // accumulator: a changed inode afterwards discards the state and rescans. let after: Awaited<ReturnType<typeof stat>> | undefined; try { after = await stat(filePath); } catch { after = undefined; } if (!after || after.dev !== stats.dev || after.ino !== stats.ino) { dropSessionScanState(filePath); return retryOnReplacement ? scanSessionInfo(filePath, false) : null; } state.fileSize = stats.size; state.mtimeMs = stats.mtimeMs; storeSessionScanState(filePath, state); store?.write({ path: filePath, dir: canonicalSessionCatalogDir(dirname(filePath)), dev: stats.dev, ino: stats.ino, size: stats.size, mtimeMs: stats.mtimeMs, id: state.info?.id ?? null, modifiedMs: state.info ? state.info.modified.getTime() : stats.mtimeMs, summaryJson: state.info ? serializeCatalogSessionInfo(state.info) : null, resumeJson: serializeSessionScanState(state) ?? null, indexedMs: Date.now(), }); // The structure is attached after the summary row: the row carries the file // identity this structure belongs to. if (store && entryRecords) { if (structureUnusable) store.clearEntryStructure(filePath); else store.writeEntryStructure(filePath, scanFrom, state.offset, entryRecords); } return state.info;}
/** * Fold the complete lines in [state.offset, size) into the accumulator. An * unterminated final line may be an in-progress append: it is returned for * snapshot-only folding, never consumed into the resumable accumulator. */async function scanSessionLines( filePath: string, state: SessionScanState, size: number, collectEntry?: (record: SessionEntryStructure | null) => void,): Promise<Buffer | undefined> { if (state.acc.invalid || state.offset >= size) return undefined; for await (const lineBuffer of readLinesAsBuffers(filePath, { start: state.offset, end: size - 1 })) { const lineEnd = state.offset + lineBuffer.length; if (lineEnd >= size) return lineBuffer; const lineOffset = state.offset; foldSessionScanLine( state.acc, lineBuffer, collectEntry ? (fields) => { // `undefined` marks a line that is not an entry (the header); // `null` marks an entry the structure cannot describe. if (fields === undefined) return; collectEntry(fields === null ? null : { ...fields, offset: lineOffset, length: lineBuffer.length }); } : undefined, ); state.tail = advanceScanTail(state.tail, lineBuffer); state.offset = lineEnd + 1; if (state.acc.invalid) break; } return undefined;}
/** * `onEntry` reports the indexable structure of every line: `undefined` for a line * that is not an entry (the header), `null` for an entry that cannot be indexed. */function foldSessionScanLine( acc: SessionScanAccumulator, lineBuffer: Buffer, onEntry?: (fields: Omit<SessionEntryStructure, "offset" | "length"> | null | undefined) => void,): void { const line = lineBuffer.toString("utf8"); if (!line.trim()) return;
// Large tool-result entries can be many MB. They do not carry the // session-list metadata we need, and parsing them during every refresh // can exhaust the daemon heap. if (line.length > SESSION_LIST_PARSE_MAX_LINE_CHARS) { onEntry?.(oversizedEntryIndexFields(line)); if (looksLikeMessageEntry(line)) { acc.messageCount++; const summary = extractOversizedMessageSummary(line); if (typeof summary.timestamp === "number" && (summary.role === "user" || summary.role === "assistant")) { acc.lastActivityTime = Math.max(acc.lastActivityTime ?? 0, summary.timestamp); } if (summary.role === "user" && !acc.firstMessage) { acc.firstMessage = summary.textPreview || "(large message)"; } } return; }
const trimmed = line.trim(); let entry: FileEntry; try { entry = JSON.parse(trimmed) as FileEntry; } catch { // An unparsable line is not an entry for the summary either; the index // cannot claim it, so the whole structure becomes unusable. onEntry?.(null); return; } onEntry?.(entryIndexFields(entry));
if (entry.type === "session_info") { const infoEntry = entry as SessionInfoEntry; acc.name = infoEntry.name?.trim() || undefined; } if (entry.type === "session_state") { const stateEntry = entry as SessionStateEntry; const status = normalizeSessionStateStatus(stateEntry.state?.status); if (status) { acc.state = { status }; } } // Keep the latest recap/verdict so off-daemon sessions don't all show as // unjudged in the agents view. Append-only, so last seen wins. if (entry.type === "agent_status") { acc.agentStatus = (entry as AgentStatusEntry).status; } if (entry.type === "model_change") { const modelEntry = entry as ModelChangeEntry; acc.model = { provider: modelEntry.provider, modelId: modelEntry.modelId }; } if (entry.type === "child_usage_attributed") { const attribution = entry as ChildUsageAttributionEntry; if (acc.assistantUsageById.has(attribution.targetId)) { acc.assistantUsageById.set(attribution.targetId, attribution.aggregateUsage); addAssistantUsage(acc.attributedChildUsage, attribution.childUsage); } } if (entry.type === "compaction" || entry.type === "branch_summary") { const summarizationUsage = (entry as CompactionEntry | BranchSummaryEntry).usage; if (summarizationUsage) addAssistantUsage(acc.summarizationUsage, summarizationUsage); } if (!acc.header) { if (entry.type !== "session") { acc.invalid = true; return; } acc.header = entry as SessionHeader; }
acc.lastActivityTime = updateLastActivityTime(acc.lastActivityTime, entry);
if (entry.type !== "message") return; acc.messageCount++;
const message = (entry as SessionMessageEntry).message; if (message.role === "assistant" && (message as { usage?: Usage }).usage) { acc.assistantUsageById.set(entry.id, (message as { usage: Usage }).usage); } if (message.role === "assistant") { const assistant = message as { provider?: string; model?: string }; if (typeof assistant.provider === "string" && typeof assistant.model === "string") { acc.model = { provider: assistant.provider, modelId: assistant.model }; } } if (!isMessageWithContent(message)) return; if (message.role !== "user" && message.role !== "assistant") return;
const textContent = extractTextContent(message); if (!textContent) return;
acc.allMessagesText = appendCappedSearchText(acc.allMessagesText, textContent); if (!acc.firstMessage && message.role === "user") { acc.firstMessage = textContent; }}
function snapshotSessionInfo( persistent: SessionScanAccumulator, tornTail: Buffer | undefined, filePath: string, stats: Awaited<ReturnType<typeof stat>>,): SessionInfo | null { let acc = persistent; if (tornTail !== undefined && tornTail.length > 0 && !acc.invalid) { acc = { ...persistent, assistantUsageById: new Map(persistent.assistantUsageById), attributedChildUsage: cloneUsage(persistent.attributedChildUsage), summarizationUsage: cloneUsage(persistent.summarizationUsage), }; foldSessionScanLine(acc, tornTail); } if (acc.invalid || !acc.header) return null; const usageTotal = emptyUsage(); for (const usage of acc.assistantUsageById.values()) { addAssistantUsage(usageTotal, usage); } addAssistantUsage(usageTotal, acc.summarizationUsage); subtractAssistantUsage(usageTotal, acc.attributedChildUsage); const header = acc.header; const cwd = typeof header.cwd === "string" ? header.cwd : ""; const parentSessionPath = header.parentSession; const rlmDepth = resolveSessionRlmDepth(header, filePath); const modified = getSessionModifiedDateFromLastActivity(acc.lastActivityTime, header, stats.mtime);
return { path: filePath, id: header.id, cwd, name: acc.name, state: acc.state, model: acc.model, parentSessionPath, rlmDepth, created: new Date(header.timestamp), modified, messageCount: acc.messageCount, firstMessage: acc.firstMessage || "(no messages)", allMessagesText: acc.allMessagesText, agentStatus: acc.agentStatus, usage: sessionUsageSummaryFrom(usageTotal), };}
export type SessionListProgress = (loaded: number, total: number) => void;export type SessionListItem = (session: SessionInfo) => void;
export interface SessionListCallbacks { onProgress?: SessionListProgress; onSession?: SessionListItem;}
async function listSessionsFromDir( dir: string, callbacks?: SessionListCallbacks, progressOffset = 0, progressTotal?: number,): Promise<SessionInfo[]> { const sessions: SessionInfo[] = []; if (!existsSync(dir)) { for (const key of sessionScanStates.keys()) { if (dirname(key) === dir) dropSessionScanState(key); } return sessions; }
try { const dirEntries = await readdir(dir); const files = dirEntries.filter((f) => f.endsWith(".jsonl")).map((f) => join(dir, f)); const total = progressTotal ?? files.length;
const present = new Set(files); for (const key of sessionScanStates.keys()) { if (dirname(key) === dir && !present.has(key)) { dropSessionScanState(key); } } sessionCatalogStoreForDir(dir)?.retain(dir, new Set(files));
let loaded = 0; let sinceYield = 0; const scanSemaphore = new Semaphore(SESSION_SCAN_CONCURRENCY); const infos = await Promise.all( files.map((file) => scanSemaphore.run(async () => { const info = await readSessionInfo(file); loaded++; callbacks?.onProgress?.(progressOffset + loaded, total); if (info) { callbacks?.onSession?.(info); } // Parsing is synchronous; a caller that lists inside its own event // loop (the daemon) still answers other clients during a cold pass. sinceYield++; if (sinceYield >= SESSION_SCAN_CONCURRENCY) { sinceYield = 0; await new Promise<void>((resolveYield) => setImmediate(resolveYield)); } return info; }), ), ); for (const info of infos) { if (info) { sessions.push(info); } } } catch { // Return no sessions when the directory cannot be read. }
return sessions;}
export class SessionManager { private sessionId: string = ""; private sessionFile: string | undefined; private sessionDir: string; private cwd: string; private persist: boolean; private flushed: boolean = false; // Flip-once cache for the no-assistant guard in _persist: true from the first // assistant message on. Refreshed on every fileEntries reassignment; hot // appends only ever set it to true, keeping _persist O(1). private hasAssistantEntry = false; private loadedEntries: FileEntry[] = []; private loadedById: Map<string, SessionEntry> = new Map(); /** Present while a large session is open without its entry payloads materialized. */ private lazyStructure?: { records: SessionEntryStructure[]; through: number; usage?: SessionUsageSummary; byId?: Map<string, SessionEntryStructure>; }; private loadingEntries = false;
/** * Entry payloads are the expensive part of a large transcript, so a lazy open * keeps only the context window and the structure. Anything that needs a * payload materializes the transcript once, then behaves exactly as before. */ private get byId(): Map<string, SessionEntry> { this.ensureEntriesLoaded(); return this.loadedById; }
private set byId(value: Map<string, SessionEntry>) { this.loadedById = value; }
private get fileEntries(): FileEntry[] { this.ensureEntriesLoaded(); return this.loadedEntries; }
private set fileEntries(value: FileEntry[]) { this.loadedEntries = value; }
/** Materializes every entry payload from the transcript file. */ private ensureEntriesLoaded(): void { const lazy = this.lazyStructure; if (!lazy || this.loadingEntries || !this.sessionFile) return; this.loadingEntries = true; try { this.lazyStructure = undefined; this.loadedEntries = loadEntriesFromFile(this.sessionFile); this._buildIndex(); this._refreshHasAssistantEntry(); } finally { this.loadingEntries = false; } }
/** * Whether the caller can answer from the structure alone. Callers that read * payloads must not use this: the accessors materialize on their own. */ hasLoadedEntries(): boolean { return this.lazyStructure === undefined; }
/** Counts entries of a type from the structure, without materializing payloads. */ countEntriesOfType(type: string): number { const lazy = this.lazyStructure; if (lazy) { let count = 0; for (const record of lazy.records) { if (record.type === type) count++; } return count; } return this.loadedEntries.filter((entry) => entry.type === type).length; }
/** Where every entry lives in the transcript, when the catalog could supply it. */ getEntryStructure(): readonly SessionEntryStructure[] | undefined { return this.lazyStructure?.records; }
/** * Whole-file usage the catalog already computed in the same scan that built the * structure, so a session summary does not have to materialize the transcript. * Undefined once the entries are loaded, when the live computation takes over. */ getCachedUsageSummary(): SessionUsageSummary | undefined { return this.lazyStructure?.usage; } private labelsById: Map<string, string> = new Map(); private labelTimestampsById: Map<string, string> = new Map(); private leafId: string | null = null; private persistListeners = new Set<SessionPersistListener>();
private constructor( cwd: string, sessionDir: string, sessionFile: string | undefined, persist: boolean, preloadedEntries?: FileEntry[], partialEntries = false, ) { this.cwd = cwd; this.sessionDir = sessionDir; this.persist = persist; if (persist && sessionDir && !existsSync(sessionDir)) { mkdirSync(sessionDir, { recursive: true }); }
if (sessionFile) { this.setSessionFile(sessionFile, preloadedEntries, partialEntries); } else { this.newSession(); } }
/** * Switch to a different session file (used for resume and branching). * preloadedEntries must be loadEntriesFromFile(sessionFile) for the same path; it * lets the async daemon path skip the synchronous re-read. */ setSessionFile(sessionFile: string, preloadedEntries?: FileEntry[], partial = false): void { this.sessionFile = resolve(sessionFile); if (existsSync(this.sessionFile)) { if (this.persist && preloadedEntries === undefined) repairJsonlDamage(this.sessionFile); this.loadedEntries = preloadedEntries ?? loadEntriesFromFile(this.sessionFile); if (partial) { // A partial window is not the file, so the flip-once cache cannot be read // off it: the transcript is already on disk, which is the only thing the // no-create guard in _persist protects. this.hasAssistantEntry = true; } else { this._refreshHasAssistantEntry(); }
// If file was empty or corrupted (no valid header), truncate and start fresh // to avoid appending messages without a session header (which breaks the session) if (this.loadedEntries.length === 0) { const explicitPath = this.sessionFile; this.newSession(); this.sessionFile = explicitPath; this._rewriteFile(); this.flushed = true; return; }
const header = this.loadedEntries.find((e) => e.type === "session") as SessionHeader | undefined; this.sessionId = header?.id ?? createSessionId();
// A partial window is not the file: migrating it would rewrite the // transcript from the window and drop the entries before it. let shouldRewrite = partial ? false : migrateToCurrentVersion(this.loadedEntries); if (header?.parentSession && !isValidRlmDepth(header.rlmDepth)) { header.rlmDepth = resolveSessionRlmDepth(header, this.sessionFile); shouldRewrite = true; } if (shouldRewrite) { this._rewriteFile(); }
this._buildIndex(); this.flushed = true; } else { const explicitPath = this.sessionFile; this.newSession(); this.sessionFile = explicitPath; // preserve explicit path from --resume selector } }
newSession(options?: NewSessionOptions): string | undefined { let sessionId = options?.id ?? createSessionId(); let sessionFile: string | undefined; const hasExplicitRlmDepth = options !== undefined && Object.hasOwn(options, "rlmDepth"); let parentHeader: Partial<SessionHeader> | undefined; if (options?.parentSession && !hasExplicitRlmDepth) { try { parentHeader = readSessionHeader(options.parentSession); } catch { // Unavailable parent metadata leaves the child depth unknown. } } if (this.persist) { if (options?.id) { sessionFile = getSessionFilePath(this.getSessionDir(), sessionId); if (existsSync(sessionFile)) { throw new Error(`Session file already exists for id "${sessionId}": ${sessionFile}`); } } else { const target = createUniqueSessionFileTarget(this.getSessionDir()); sessionId = target.sessionId; sessionFile = target.sessionFile; } }
this.sessionId = sessionId; const timestamp = new Date().toISOString(); const git = this.persist ? (captureGitContext(this.cwd) ?? undefined) : undefined; const rlmDepth = hasExplicitRlmDepth ? options?.rlmDepth : options?.parentSession ? deriveChildRlmDepth(parentHeader) : rootRlmDepthFromEnv(); const header: SessionHeader = { type: "session", version: CURRENT_SESSION_VERSION, id: this.sessionId, timestamp, cwd: this.cwd, parentSession: options?.parentSession, rlmDepth, git, }; this.lazyStructure = undefined; this.loadedEntries = [header]; this.hasAssistantEntry = false; this.loadedById.clear(); this.labelsById.clear(); this.labelTimestampsById.clear(); this.leafId = null; this.flushed = false;
if (this.persist) { this.sessionFile = sessionFile; } return this.sessionFile; }
// Cold-path recompute of the flip-once has-assistant cache. Hot appends keep // it O(1) in _appendEntry; every fileEntries reassignment (open, branch, // rollback) refreshes it here. private _refreshHasAssistantEntry(): void { if (this.lazyStructure) { // A lazily opened transcript is already on disk, so the no-create guard // cannot apply. Materializing here would defeat the lazy open; reading the // entries later recomputes this exactly. this.hasAssistantEntry = true; return; } this.hasAssistantEntry = this.loadedEntries.some((e) => e.type === "message" && e.message.role === "assistant"); }
private _buildIndex(): void { this.loadedById.clear(); this.labelsById.clear(); this.labelTimestampsById.clear(); this.leafId = null; for (const entry of this.loadedEntries) { if (entry.type === "session") continue; this.loadedById.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 content = `${this.fileEntries.map((e) => JSON.stringify(e)).join("\n")}\n`; const targetPath = realpathIfPresentSync(this.sessionFile); const directory = dirname(targetPath); mkdirSync(directory, { recursive: true }); const metadata = statMetadataIfPresent(targetPath); writeFileAtomicSync(targetPath, content, { ...(metadata === undefined ? {} : { mode: metadata.mode }), beforeRename: (tempPath) => { if (metadata !== undefined) chownSync(tempPath, metadata.uid, metadata.gid); }, }); this._notifyPersistListeners(); }
private _notifyPersistListeners(): void { if (!this.sessionFile) { return; } for (const listener of this.persistListeners) { try { listener(this.sessionFile); } catch { // Persistence observers must not break session writes. } } }
onPersist(listener: SessionPersistListener): () => void { this.persistListeners.add(listener); return () => { this.persistListeners.delete(listener); }; }
isPersisted(): boolean { return this.persist; }
getCwd(): string { return this.cwd; }
getSessionDir(): string { return this.sessionDir; }
getSessionId(): string { return this.sessionId; }
getSessionFile(): string | undefined { return this.sessionFile; }
materializeSessionFile(sessionDir?: string): string { if (this.sessionFile) { return this.sessionFile; } const dir = sessionDir ?? (this.sessionDir || getDefaultSessionDir(this.cwd)); if (!existsSync(dir)) { mkdirSync(dir, { recursive: true }); } const previousHeader = this.getHeader(); const target = createUniqueSessionFileTarget(dir); this.sessionDir = dir; this.sessionId = target.sessionId; this.sessionFile = target.sessionFile; this.persist = true; const timestamp = new Date().toISOString(); const git = captureGitContext(this.cwd) ?? undefined; const header: SessionHeader = { type: "session", version: CURRENT_SESSION_VERSION, id: this.sessionId, timestamp, cwd: this.cwd, parentSession: previousHeader?.parentSession, rlmDepth: resolveSessionRlmDepth(previousHeader ?? {}, target.sessionFile), git, }; this.fileEntries = [header, ...this.getEntries()]; this._refreshHasAssistantEntry(); this._rewriteFile(); this.flushed = true; return this.sessionFile; }
getSessionArtifactDir(): string | undefined { return this.persist ? getSessionArtifactPath(this.sessionDir, this.sessionId) : undefined; }
/** * Force-write all in-memory entries to the session file immediately. * This bypasses the no-assistant guard in {@link _persist} so that * pre-model entries (session header, goal state, settings changes) * are durable on disk before the first assistant response. * No-op for in-memory (non-persisted) sessions. */ flushNow(): void { if (!this.persist || !this.sessionFile) return; if (this.flushed && existsSync(this.sessionFile)) return; this._rewriteFile(); this.flushed = true; }
_persist(entry: SessionEntry): void { if (!this.persist || !this.sessionFile) return;
const shouldPersistWithoutAssistant = entry.type === "session_state" || entry.type === "session_info"; if (!this.hasAssistantEntry && !shouldPersistWithoutAssistant) { this.flushed = false; return; }
// The existsSync check is what lets the next append recover from the session // file being deleted underneath a live session: without it appendFileSync // would recreate a headerless stub (pinned by session-state.test.ts's // "rewrites the full session if the session file disappears after flushing"). if (!this.flushed || !existsSync(this.sessionFile)) { this._rewriteFile(); this.flushed = true; } else { mkdirSync(dirname(this.sessionFile), { recursive: true }); appendFileSync(this.sessionFile, `${JSON.stringify(entry)}\n`); this._notifyPersistListeners(); } }
private _appendEntry(entry: SessionEntry): void { this.fileEntries.push(entry); if (entry.type === "message" && entry.message.role === "assistant") { this.hasAssistantEntry = true; } this.byId.set(entry.id, entry); this.leafId = entry.id; this._persist(entry); }
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; }
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; }
appendServiceTierChange(serviceTier: ServiceTier): string { const entry: ServiceTierChangeEntry = { type: "service_tier_change", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), serviceTier, }; this._appendEntry(entry); return 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; }
appendCompaction<T = unknown>( summary: string, firstKeptEntryId: string, tokensBefore: number, details?: T, fromHook?: boolean, customInstructions?: string, usage?: Usage, harnessDigest?: string, ): string { const entry: CompactionEntry<T> = { type: "compaction", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), summary, firstKeptEntryId, tokensBefore, details, fromHook, customInstructions, usage, harnessDigest, }; this._appendEntry(entry); return 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; }
appendCustomEntryWithRollback(customType: string, data?: unknown): string { return this._appendEntryWithRollback(() => this.appendCustomEntry(customType, data)); }
appendChildUsageAttribution( targetId: string, childUsage: Usage, aggregateUsage: Usage, origin?: ChildUsageAttributionEntry["origin"], ): string { const target = this.byId.get(targetId); if (target?.type !== "message" || target.message.role !== "assistant") { throw new Error(`Assistant message entry ${targetId} not found`); }
target.message.usage = cloneUsage(aggregateUsage); const entry: ChildUsageAttributionEntry = { type: "child_usage_attributed", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), targetId, childUsage: cloneUsage(childUsage), aggregateUsage: cloneUsage(aggregateUsage), ...(origin ? { origin } : {}), }; this._appendEntry(entry); return entry.id; }
appendSessionInfo(name: string): string { const entry: SessionInfoEntry = { type: "session_info", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), name: name.trim(), }; this._appendEntry(entry); return entry.id; }
appendSessionState(state: SessionState): string { const entry: SessionStateEntry = { type: "session_state", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), state: { status: state.status }, }; this._appendEntry(entry); return entry.id; }
getSessionName(): string | undefined { // A named session keeps its name in the last session_info entry, which the // structure already points at; only that one line is read when lazy. With a // complete structure, no record means no name. const lazy = this.lazyStructure; if (lazy) { const record = lastRecordOfType(lazy.records, "session_info"); if (!record) return undefined; const entry = this.readEntryRecord(record); return entry?.type === "session_info" ? entry.name?.trim() || undefined : undefined; } const entries = this.getEntries(); for (let i = entries.length - 1; i >= 0; i--) { const entry = entries[i]; if (entry.type === "session_info") { return entry.name?.trim() || undefined; } } return undefined; }
getSessionState(): SessionState | undefined { const lazy = this.lazyStructure; if (lazy) { const record = lastRecordOfType(lazy.records, "session_state"); if (!record) return undefined; const entry = this.readEntryRecord(record); const status = entry?.type === "session_state" ? normalizeSessionStateStatus(entry.state.status) : undefined; return status ? { status } : undefined; } const entries = this.getEntries(); for (let i = entries.length - 1; i >= 0; i--) { const entry = entries[i]; if (entry.type === "session_state") { const status = normalizeSessionStateStatus(entry.state.status); if (status) { return { status }; } } } return undefined; }
/** * True when the session holds user-meaningful persisted content, as opposed to * only daemon-written bookkeeping (session_state, agent_status, git_state) or * the default model/thinking entries every new session is created with. Used by * the daemon discard guard to decide whether a message-less draft is safe to * delete (that guard always also requires zero messages). * * createAgentSession opens a new session with an optional leading `model_change` * followed by `thinking_level_change` and `service_tier_change`. That creation * prefix is skipped; anything beyond it is user content. */ hasUserContent(): boolean { const contentEntries = this.getEntries().filter((entry) => CONTENT_ENTRY_TYPES.has(entry.type)); let start = 0; if (contentEntries[start]?.type === "model_change") { start++; } if (contentEntries[start]?.type === "thinking_level_change") { start++; } if (contentEntries[start]?.type === "service_tier_change") { start++; } return contentEntries.length > start; }
appendAgentStatus(status: AgentStatus): string { const entry: AgentStatusEntry = { type: "agent_status", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), status: { summary: status.summary, taskState: status.taskState, basedOnMessageCount: status.basedOnMessageCount, }, }; this._appendEntry(entry); return entry.id; }
appendGitState(git: GitContext): string { const entry: GitStateEntry = { type: "git_state", id: generateId(this.byId), parentId: this.leafId, timestamp: new Date().toISOString(), git, }; this._appendEntry(entry); return entry.id; }
recordGitStateIfChanged(): string | undefined { if (!this.persist) return undefined; const git = captureGitContext(this.cwd); if (!git) return undefined; const last = this.getActiveGitContext(); if (last && gitContextsEqual(last, git)) return undefined; return this.appendGitState(git); }
private getActiveGitContext(): GitContext | undefined { let current = this.leafId ? this.byId.get(this.leafId) : undefined; while (current) { if (current.type === "git_state") return current.git; current = current.parentId ? this.byId.get(current.parentId) : undefined; } const header = this.fileEntries[0]; return header?.type === "session" ? header.git : undefined; }
getLatestAgentStatus(): AgentStatus | undefined { // Walk the current leaf to root so we only read status on the active branch, // not a sibling branch's status that happens to sit later in the file. A // lazily opened transcript walks the structure and reads only the entry it // stops at, instead of materializing the session for one summary line. const lazy = this.lazyStructure; if (lazy) { const record = this.branchRecordWhere((candidate) => candidate.type === "agent_status"); const entry = record ? this.readEntryRecord(record) : undefined; return entry?.type === "agent_status" ? { ...entry.status } : undefined; } let current = this.leafId ? this.byId.get(this.leafId) : undefined; while (current) { if (current.type === "agent_status") { return { ...current.status }; } current = current.parentId ? this.byId.get(current.parentId) : undefined; } return undefined; }
/** Structure records on the active branch, root to leaf. */ private branchRecords(type?: string): SessionEntryStructure[] { const lazy = this.lazyStructure; if (!lazy) return []; lazy.byId ??= new Map(lazy.records.map((record) => [record.id, record])); const chain: SessionEntryStructure[] = []; let current = this.leafId ? lazy.byId.get(this.leafId) : undefined; while (current) { if (!type || current.type === type) chain.push(current); current = current.parentId ? lazy.byId.get(current.parentId) : undefined; } return chain.reverse(); }
/** * Active branch entries, root to leaf, the same order `getBranch()` returns. A * lazily opened transcript reads only the entries a caller's type filter * admits, instead of materializing the whole session to look at a few. */ *branchEntries(type?: string): Generator<SessionEntry> { if (!this.lazyStructure) { for (const entry of this.getBranch()) { if (!type || entry.type === type) yield entry; } return; } for (const record of this.branchRecords(type)) { const entry = this.readEntryRecord(record); if (entry) yield entry; } }
/** Entry types on the active branch, root to leaf, answered from the structure. */ branchEntryTypes(): string[] { if (!this.lazyStructure) return this.getBranch().map((entry) => entry.type); return this.branchRecords().map((record) => record.type); }
/** * Nearest entry on the active branch matching a predicate, reading only entries * of `type`. Answered from the structure when the transcript is not * materialized, so session creation reads one line instead of the whole file. */ findLastBranchEntry(type: string, predicate: (entry: SessionEntry) => boolean): SessionEntry | undefined { if (!this.lazyStructure) { const branch = this.getBranch(); for (let i = branch.length - 1; i >= 0; i--) { const entry = branch[i]!; if (entry.type === type && predicate(entry)) return entry; } return undefined; } const chain = this.branchRecords(type); for (let i = chain.length - 1; i >= 0; i--) { const entry = this.readEntryRecord(chain[i]!); if (entry && predicate(entry)) return entry; } return undefined; }
/** * Whether the active branch already records an entry of this type. Answered * from the structure when the transcript is not materialized, so session * creation does not have to read a large session's payloads. */ branchHasEntryType(type: string): boolean { if (this.lazyStructure) return this.branchRecordWhere((record) => record.type === type) !== undefined; return this.getBranch().some((entry) => entry.type === type); }
/** First structure record on the active branch, walking leaf to root. */ private branchRecordWhere(matches: (record: SessionEntryStructure) => boolean): SessionEntryStructure | undefined { const lazy = this.lazyStructure; if (!lazy) return undefined; lazy.byId ??= new Map(lazy.records.map((record) => [record.id, record])); let current = this.leafId ? lazy.byId.get(this.leafId) : undefined; while (current) { if (matches(current)) return current; current = current.parentId ? lazy.byId.get(current.parentId) : undefined; } return undefined; }
/** Reads and parses one entry by its structure record, without materializing the rest. */ private readEntryRecord(record: SessionEntryStructure): SessionEntry | undefined { const cached = this.loadedById.get(record.id); if (cached) return cached; if (!this.sessionFile) return undefined; const entry = parseEntryLine(readBytesSync(this.sessionFile, record.offset, record.offset + record.length)); if (!entry || entry.type === "session") return undefined; this.loadedById.set(entry.id, entry); return entry; }
appendCustomMessageEntry<T = unknown>( customType: string, content: string | (TextContent | ImageContent)[], display: boolean, details?: T, ): string { const entry: CustomMessageEntry<T> = { 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 custom message, undoing the append if persistence fails so a * best-effort record never leaves an unsaved leaf for later entries. */ appendCustomMessageEntryWithRollback<T = unknown>( customType: string, content: string | (TextContent | ImageContent)[], display: boolean, details?: T, ): string { return this._appendEntryWithRollback(() => this.appendCustomMessageEntry(customType, content, display, details)); }
private _appendEntryWithRollback(append: () => string): string { const previousLeafId = this.leafId; try { const entryId = append(); this.flushNow(); return entryId; } catch (error) { // The append indexes the entry before persisting it; undo exactly that. if (this.leafId !== null && this.leafId !== previousLeafId) { this.byId.delete(this.leafId); this.fileEntries.pop(); this._refreshHasAssistantEntry(); this.leafId = previousLeafId; // The failed append may have left a torn line on disk. Restore the file // from the rolled-back entries now; if that also fails (e.g. the disk is // still full), fall back to forcing the next persist to rewrite. this.flushed = false; try { this.flushNow(); } catch { this.flushed = false; } } throw error; } }
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); }
getChildren(parentId: string): SessionEntry[] { const children: SessionEntry[] = []; for (const entry of this.byId.values()) { if (entry.parentId === parentId) { children.push(entry); } } return children; }
getLabel(id: string): string | undefined { return this.labelsById.get(id); }
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; }
getBranch(fromId?: string): SessionEntry[] { // push+reverse, not unshift-per-entry: unshift is O(n), which makes this O(n^2) on long sessions. 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; }
/** * The branch as far as the transcript has been materialized. With a lazy open * that is the context window, which is all the context-usage computation reads * (the latest compaction and any assistant usage after it); every other caller * wants {@link getBranch}, which materializes first. */ getContextBranch(): SessionEntry[] { // push+reverse, not unshift-per-entry: unshift is O(n), which makes this O(n^2) on long sessions. const path: SessionEntry[] = []; let current = this.leafId ? this.loadedById.get(this.leafId) : undefined; while (current) { path.push(current); current = current.parentId ? this.loadedById.get(current.parentId) : undefined; } path.reverse(); return path; }
buildSessionContext(): SessionContext { // Pass loadedEntries directly rather than getEntries(): the resolved context // is computed from the leaf-to-root walk over loadedById (which already // excludes the header), so the entries argument is only a fallback for an // undefined leaf — never hit here since leafId is always set or null. Avoids // an O(n) array copy on every call (attach, get_session_context, agent init, // ...) and keeps a lazily opened transcript from materializing. return buildSessionContext(this.loadedEntries as SessionEntry[], this.leafId, this.loadedById); }
getHeader(): SessionHeader | null { // The window always carries the header, so this needs no materialization. const source = this.lazyStructure ? this.loadedEntries : this.fileEntries; const h = source.find((e) => e.type === "session"); return h ? (h as SessionHeader) : null; }
getEntries(): SessionEntry[] { return this.fileEntries.filter((e): e is SessionEntry => e.type !== "session"); }
getFlatTree(): SessionTreeFlatNode[] { return this.getEntries().map((entry) => ({ entry, label: this.labelsById.get(entry.id), labelTimestamp: this.labelTimestampsById.get(entry.id), })); }
getTree(): SessionTreeNode[] { const entries = this.getFlatTree(); const nodeMap = new Map<string, SessionTreeNode>(); const roots: SessionTreeNode[] = [];
for (const flatNode of entries) { nodeMap.set(flatNode.entry.id, { ...flatNode, children: [] }); }
for (const flatNode of entries) { const entry = flatNode.entry; 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 { 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; }
branch(branchFromId: string): void { if (!this.byId.has(branchFromId)) { throw new Error(`Entry ${branchFromId} not found`); } this.leafId = branchFromId; }
resetLeaf(): void { this.leafId = null; }
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`); } this.leafId = branchFromId; const entry: BranchSummaryEntry = { type: "branch_summary", id: generateId(this.byId), parentId: branchFromId, timestamp: new Date().toISOString(), fromId: branchFromId ?? "root", summary, details, fromHook, usage, }; this._appendEntry(entry); return entry.id; }
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`); }
const pathWithoutLabels = path.filter((e) => e.type !== "label");
const target = this.persist ? createUniqueSessionFileTarget(this.getSessionDir()) : { sessionId: createSessionId(), sessionFile: undefined }; const newSessionId = target.sessionId; const timestamp = new Date().toISOString(); const newSessionFile = target.sessionFile;
const header: SessionHeader = { type: "session", version: CURRENT_SESSION_VERSION, id: newSessionId, timestamp, cwd: this.cwd, parentSession: this.persist ? previousSessionFile : undefined, rlmDepth: resolveSessionRlmDepth(this.getHeader() ?? {}, previousSessionFile ?? newSessionFile ?? ""), git: this.persist ? (captureGitContext(this.cwd) ?? undefined) : undefined, };
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) { 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._refreshHasAssistantEntry(); this.sessionId = newSessionId; this.sessionFile = newSessionFile; this._buildIndex();
// Only write the file now if it contains an assistant message. // Otherwise defer to _persist(), which creates the file on the // first assistant response, matching the newSession() contract // and avoiding the duplicate-header bug when _persist()'s // no-assistant guard later resets flushed to false. if (this.hasAssistantEntry) { this._rewriteFile(); this.flushed = true; } else { this.flushed = false; }
return newSessionFile; }
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._refreshHasAssistantEntry(); this.sessionId = newSessionId; this._buildIndex(); return undefined; }
static create(cwd: string, sessionDir?: string): SessionManager { const dir = sessionDir ?? getDefaultSessionDir(cwd); return new SessionManager(cwd, dir, undefined, true); }
static open(path: string, sessionDir?: string, cwdOverride?: string): SessionManager { // Only the header's cwd is needed to construct the manager; the constructor // (setSessionFile) performs the full parse. Read just the first line here // instead of parsing the entire file a second time — that double parse is a // needless O(n) cost on open and is noticeable for long sessions. let cwd = cwdOverride; if (cwd === undefined) { let header: Partial<SessionHeader> | undefined; try { header = readSessionHeader(path); } catch { header = undefined; } // readSessionHeader only inspects the first physical line. If that isn't a // valid session header (e.g. a leading blank/whitespace or malformed line), // fall back to the full loader, which trims and skips such lines exactly // like setSessionFile does — so this.cwd stays consistent with the header // the session is actually loaded with. This slow path is rare. if (header?.type !== "session" || typeof header.id !== "string") { header = loadEntriesFromFile(path).find((e) => e.type === "session") as SessionHeader | undefined; } cwd = header?.cwd; } const dir = sessionDir ?? resolve(path, ".."); return new SessionManager(cwd ?? process.cwd(), dir, path, true); }
/** * Opens a large transcript from the catalog's entry structure plus the bytes the * context actually needs. Returns undefined whenever that is not provably safe: * no catalog, no complete structure, a file below the lazy threshold, or a * structure that does not cover the whole file. */ private static async openLazy(path: string, dir: string, cwdOverride?: string): Promise<SessionManager | undefined> { if (!isSessionCatalogStoreEnabled()) return undefined; let stats: ReturnType<typeof statSync>; try { stats = statSync(path); } catch { return undefined; } if (stats.size < SESSION_LAZY_OPEN_MIN_BYTES) return undefined; const store = sessionCatalogStoreFor(path); let cached = store?.read(path); if (!cached || cached.dev !== stats.dev || cached.ino !== stats.ino) return undefined; if (cached.size !== stats.size) { // The transcript grew or was rewritten since the catalog saw it: fold the // new bytes in first, so the structure still describes the whole file. if (!(await readSessionInfo(path))) return undefined; stats = statSync(path); cached = store?.read(path); if (!cached || cached.size !== stats.size) return undefined; } const structure = store?.readEntryStructure(path); if (!structure?.blob || structure.throughOffset < stats.size || structure.count === 0) return undefined; const records = readSessionEntryStructure(structure.blob); const last = records[records.length - 1]; if (!last || last.offset + last.length > stats.size) return undefined; const window = readSessionContextWindow(path, records); if (!window) return undefined; const cwd = cwdOverride ?? window.header.cwd; const manager = new SessionManager( cwd ?? process.cwd(), dir, path, true, [window.header, ...window.entries], true, ); manager.lazyStructure = { records, through: structure.throughOffset, usage: cached.summaryJson ? hydrateCatalogSessionInfo(path, cached.summaryJson)?.usage : undefined, }; return manager; }
static async openAsync(path: string, sessionDir?: string, cwdOverride?: string): Promise<SessionManager> { if (!existsSync(path)) { return SessionManager.open(path, sessionDir, cwdOverride); } repairJsonlDamage(path); const dir = sessionDir ?? resolve(path, ".."); const opened = await SessionManager.openLazy(path, dir, cwdOverride); if (opened) return opened;
const entries = await loadEntriesFromFileAsync(path); if (entries.length === 0) { return SessionManager.open(path, sessionDir, cwdOverride); } const cwd = cwdOverride ?? (entries[0] as SessionHeader).cwd; return new SessionManager(cwd ?? process.cwd(), dir, path, true, entries); }
static continueRecent(cwd: string, sessionDir?: string): SessionManager { const dir = sessionDir ?? getDefaultSessionDir(cwd); const mostRecent = findMostRecentSessionForCwd(dir, cwd); if (mostRecent) { return new SessionManager(cwd, dir, mostRecent, true); } return new SessionManager(cwd, dir, undefined, true); }
static inMemory(cwd: string = process.cwd(), sessionDir = ""): SessionManager { return new SessionManager(cwd, sessionDir, undefined, false); }
static forkFrom(sourcePath: string, targetCwd: string, sessionDir?: string): SessionManager { const sourceEntries = loadEntriesFromFile(sourcePath); if (sourceEntries.length === 0) { throw new Error(`Cannot fork: source session file is empty or invalid: ${sourcePath}`); }
const sourceHeader = sourceEntries.find((e) => e.type === "session") as SessionHeader | undefined; if (!sourceHeader) { throw new Error(`Cannot fork: source session has no header: ${sourcePath}`); } migrateToCurrentVersion(sourceEntries);
const dir = sessionDir ?? getDefaultSessionDir(targetCwd); if (!existsSync(dir)) { mkdirSync(dir, { recursive: true }); }
const target = createUniqueSessionFileTarget(dir); const newSessionId = target.sessionId; const timestamp = new Date().toISOString(); const newSessionFile = target.sessionFile;
const newHeader: SessionHeader = { type: "session", version: CURRENT_SESSION_VERSION, id: newSessionId, timestamp, cwd: targetCwd, parentSession: sourcePath, rlmDepth: resolveSessionRlmDepth(sourceHeader, sourcePath), git: captureGitContext(targetCwd) ?? undefined, }; // The whole fork flows through one descriptor: one openSync, one // writeAllSync per line, one closeSync. writeAllSync loops past short // counts so every JSONL line lands whole, and "w" creates the target // fresh because createUniqueSessionFileTarget guarantees it does not // exist yet. const descriptor = openSync(newSessionFile, "w"); try { writeAllSync(descriptor, `${JSON.stringify(newHeader)}\n`, newSessionFile);
// Drop the source's git_state entries (re-linking children): they describe the source repo, // so the fork would otherwise report the source's git instead of its own target context. const droppedParent = new Map<string, string | null>(); for (const entry of sourceEntries) { if (entry.type === "git_state") droppedParent.set(entry.id, entry.parentId); } const liveParent = (parentId: string | null): string | null => { let pid = parentId; while (pid !== null && droppedParent.has(pid)) pid = droppedParent.get(pid) ?? null; return pid; }; for (const entry of sourceEntries) { if (entry.type === "session" || entry.type === "git_state") continue; const parentId = liveParent(entry.parentId); const out = parentId === entry.parentId ? entry : { ...entry, parentId }; writeAllSync(descriptor, `${JSON.stringify(out)}\n`, newSessionFile); } } finally { closeSync(descriptor); }
return new SessionManager(targetCwd, dir, newSessionFile, true); }
static async list(cwd: string, sessionDir?: string, callbacks?: SessionListCallbacks): Promise<SessionInfo[]> { const dir = sessionDir ?? getDefaultSessionDir(cwd); const matchesCwd = (session: SessionInfo) => sessionInfoMatchesCwd(session, cwd); const sessions = ( await listSessionsFromDir(dir, { onProgress: callbacks?.onProgress, onSession: callbacks?.onSession ? (session) => { if (matchesCwd(session)) { callbacks.onSession?.(session); } } : undefined, }) ).filter(matchesCwd); sessions.sort((a, b) => b.modified.getTime() - a.modified.getTime()); return sessions; }
static async listAll(callbacks?: SessionListCallbacks, sessionDir?: string): Promise<SessionInfo[]> { const sessionsDir = sessionDir ?? getSessionsDir(); const sessions = await listSessionsFromDir(sessionsDir, callbacks); sessions.sort((a, b) => b.modified.getTime() - a.modified.getTime()); return sessions; }}