diff --git a/src/agents/config.ts b/src/agents/config.ts index d50283cead85..344e5dedb5f6 100644 --- a/src/agents/config.ts +++ b/src/agents/config.ts @@ -132,8 +132,3 @@ export function getAgentDir(): string { export function getBinDir(): string { return join(getAgentDir(), "bin"); } - -/** Get path to sessions directory */ -export function getSessionsDir(): string { - return join(getAgentDir(), "sessions"); -} diff --git a/src/agents/sessions/session-manager-list.ts b/src/agents/sessions/session-manager-list.ts deleted file mode 100644 index 413b53230980..000000000000 --- a/src/agents/sessions/session-manager-list.ts +++ /dev/null @@ -1,297 +0,0 @@ -import { closeSync, existsSync, openSync, readdirSync, readSync, statSync } from "node:fs"; -import { readdir, readFile, stat } from "node:fs/promises"; -import { join } from "node:path"; -import pMap, { pMapSkip } from "p-map"; -import type { Message, TextContent } from "../../llm/types.js"; -import { logWarn } from "../../logger.js"; -import { getSessionsDir } from "../config.js"; -import type { AgentMessage } from "../runtime/index.js"; -import type { - FileEntry, - SessionEntryBase, - SessionHeader, - SessionInfo, - SessionListProgress, -} from "./session-manager-types.js"; - -const SESSION_HEADER_READ_CHUNK_BYTES = 4096; -const MAX_SESSION_HEADER_BYTES = 64 * 1024; -const MAX_CONCURRENT_SESSION_INFO_LOADS = 10; - -function readFirstSessionFileLine(filePath: string): string | undefined { - const fd = openSync(filePath, "r"); - try { - const chunks: Buffer[] = []; - let totalBytes = 0; - while (totalBytes < MAX_SESSION_HEADER_BYTES) { - const buffer = Buffer.alloc( - Math.min(SESSION_HEADER_READ_CHUNK_BYTES, MAX_SESSION_HEADER_BYTES - totalBytes), - ); - const bytesRead = readSync(fd, buffer, 0, buffer.length, totalBytes); - if (bytesRead === 0) { - break; - } - const newlineIndex = buffer.indexOf(0x0a); - if (newlineIndex >= 0 && newlineIndex < bytesRead) { - chunks.push(buffer.subarray(0, newlineIndex)); - return Buffer.concat(chunks).toString("utf8"); - } - chunks.push(buffer.subarray(0, bytesRead)); - totalBytes += bytesRead; - } - return chunks.length > 0 ? Buffer.concat(chunks).toString("utf8") : undefined; - } finally { - closeSync(fd); - } -} - -function readSessionHeaderFromFile(filePath: string): SessionHeader | undefined { - try { - const firstLine = readFirstSessionFileLine(filePath); - if (!firstLine) { - return undefined; - } - const header = JSON.parse(firstLine); - return header.type === "session" && typeof header.id === "string" ? header : undefined; - } catch { - return undefined; - } -} - -export function findMostRecentSession(sessionDir: string, cwd?: string): string | null { - try { - const files = readdirSync(sessionDir) - .filter((file) => file.endsWith(".jsonl")) - .map((file) => join(sessionDir, file)) - .map((path) => ({ path, header: readSessionHeaderFromFile(path) })) - .filter( - (candidate): candidate is { path: string; header: SessionHeader } => - candidate.header !== undefined && (cwd === undefined || candidate.header.cwd === cwd), - ) - .map((candidate) => ({ path: candidate.path, mtime: statSync(candidate.path).mtime })) - .toSorted((left, right) => right.mtime.getTime() - left.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 { - if (typeof message.content === "string") { - return message.content; - } - return message.content - .filter((block): block is TextContent => block.type === "text") - .map((block) => block.text) - .join(" "); -} - -function getLastActivityTime(entries: FileEntry[]): number | undefined { - let lastActivityTime: number | undefined; - for (const entry of entries) { - if (entry.type !== "message") { - continue; - } - const message = entry.message; - if ( - !isMessageWithContent(message) || - (message.role !== "user" && message.role !== "assistant") - ) { - continue; - } - const messageTimestamp = (message as { timestamp?: number }).timestamp; - if (typeof messageTimestamp === "number") { - lastActivityTime = Math.max(lastActivityTime ?? 0, messageTimestamp); - continue; - } - const entryTimestamp = (entry as SessionEntryBase).timestamp; - if (typeof entryTimestamp === "string") { - const timestamp = new Date(entryTimestamp).getTime(); - if (!Number.isNaN(timestamp)) { - lastActivityTime = Math.max(lastActivityTime ?? 0, timestamp); - } - } - } - return lastActivityTime; -} - -function getSessionModifiedDate( - entries: FileEntry[], - header: SessionHeader, - statsMtime: Date, -): Date { - const lastActivityTime = getLastActivityTime(entries); - if (typeof lastActivityTime === "number" && lastActivityTime > 0) { - return new Date(lastActivityTime); - } - const headerTime = - typeof header.timestamp === "string" ? new Date(header.timestamp).getTime() : Number.NaN; - return !Number.isNaN(headerTime) ? new Date(headerTime) : statsMtime; -} - -async function buildSessionInfo(filePath: string): Promise { - try { - const content = await readFile(filePath, "utf8"); - const entries: FileEntry[] = []; - let skipped = 0; - for (const line of content.trim().split("\n")) { - if (!line.trim()) { - continue; - } - try { - entries.push(JSON.parse(line) as FileEntry); - } catch { - skipped += 1; - } - } - if (skipped > 0) { - logWarn( - `buildSessionInfo: skipped ${skipped} malformed JSONL line(s) in ${filePath} — ` + - `${entries.length} valid entries were loaded`, - ); - } - const header = entries[0]; - if (!header || header.type !== "session") { - return null; - } - - const stats = await stat(filePath); - let messageCount = 0; - let firstMessage = ""; - const allMessages: string[] = []; - let name: string | undefined; - for (const entry of entries) { - if (entry.type === "session_info") { - name = entry.name?.trim() || undefined; - } - if (entry.type !== "message") { - continue; - } - messageCount += 1; - const message = entry.message; - if ( - !isMessageWithContent(message) || - (message.role !== "user" && message.role !== "assistant") - ) { - continue; - } - const textContent = extractTextContent(message); - if (!textContent) { - continue; - } - allMessages.push(textContent); - if (!firstMessage && message.role === "user") { - firstMessage = textContent; - } - } - return { - path: filePath, - id: header.id, - cwd: typeof header.cwd === "string" ? header.cwd : "", - name, - parentSessionPath: header.parentSession, - created: new Date(header.timestamp), - modified: getSessionModifiedDate(entries, header, stats.mtime), - messageCount, - firstMessage: firstMessage || "(no messages)", - allMessagesText: allMessages.join(" "), - }; - } catch { - return null; - } -} - -async function listSessionsFromDir( - dir: string, - onProgress?: SessionListProgress, - progressOffset = 0, - progressTotal?: number, - cwd?: string, -): Promise { - if (!existsSync(dir)) { - return []; - } - try { - const files = (await readdir(dir)) - .filter((file) => file.endsWith(".jsonl")) - .map((file) => join(dir, file)); - const total = progressTotal ?? files.length; - let loaded = 0; - const sessions = await pMap( - files, - async (file) => { - try { - return (await buildSessionInfo(file)) ?? pMapSkip; - } catch { - return pMapSkip; - } finally { - loaded += 1; - onProgress?.(progressOffset + loaded, total); - } - }, - { concurrency: MAX_CONCURRENT_SESSION_INFO_LOADS, stopOnError: false }, - ); - return sessions.filter((info) => cwd === undefined || info.cwd === cwd); - } catch { - return []; - } -} - -export async function listSessions( - cwd: string, - sessionDir: string, - onProgress?: SessionListProgress, -): Promise { - const sessions = await listSessionsFromDir(sessionDir, onProgress, 0, undefined, cwd); - sessions.sort((left, right) => right.modified.getTime() - left.modified.getTime()); - return sessions; -} - -export async function listAllSessions(onProgress?: SessionListProgress): Promise { - try { - const sessionsDir = getSessionsDir(); - if (!existsSync(sessionsDir)) { - return []; - } - const directories = (await readdir(sessionsDir, { withFileTypes: true })) - .filter((entry) => entry.isDirectory()) - .map((entry) => join(sessionsDir, entry.name)); - const directoryFiles: string[][] = []; - let totalFiles = 0; - for (const directory of directories) { - try { - const files = (await readdir(directory)) - .filter((file) => file.endsWith(".jsonl")) - .map((file) => join(directory, file)); - directoryFiles.push(files); - totalFiles += files.length; - } catch { - directoryFiles.push([]); - } - } - - let loaded = 0; - const sessions = await pMap( - directoryFiles.flat(), - async (file) => { - try { - return (await buildSessionInfo(file)) ?? pMapSkip; - } catch { - return pMapSkip; - } finally { - loaded += 1; - onProgress?.(loaded, totalFiles); - } - }, - { concurrency: MAX_CONCURRENT_SESSION_INFO_LOADS, stopOnError: false }, - ); - sessions.sort((left, right) => right.modified.getTime() - left.modified.getTime()); - return sessions; - } catch { - return []; - } -} diff --git a/src/agents/sessions/session-manager-types.ts b/src/agents/sessions/session-manager-types.ts index 37434494fe25..bcff052f8fbe 100644 --- a/src/agents/sessions/session-manager-types.ts +++ b/src/agents/sessions/session-manager-types.ts @@ -125,22 +125,6 @@ export interface SessionContext { model: { provider: string; modelId: string } | null; } -export interface SessionInfo { - path: string; - id: string; - /** Working directory where the session started. Empty for old sessions. */ - cwd: string; - name?: string; - parentSessionPath?: string; - created: Date; - modified: Date; - messageCount: number; - firstMessage: string; - allMessagesText: string; -} - -export type SessionListProgress = (loaded: number, total: number) => void; - interface PromptReleasedOpaqueEntry { type: "prompt_released_opaque"; record: unknown; diff --git a/src/agents/sessions/session-manager.test.ts b/src/agents/sessions/session-manager.test.ts index fa72b007ce0a..d666bfd73b50 100644 --- a/src/agents/sessions/session-manager.test.ts +++ b/src/agents/sessions/session-manager.test.ts @@ -22,7 +22,6 @@ import { loadSqliteMarkedSessionFile } from "./session-manager-file.js"; import { buildSessionContext, CURRENT_SESSION_VERSION, - findMostRecentSession, loadEntriesFromFile, parseSessionEntries, SessionManager, @@ -686,102 +685,6 @@ describe("SessionManager.open", () => { expect(entries.filter((entry) => entry.type === "session")).toHaveLength(1); }); - it("continues a valid recent session when the header exceeds the first read chunk", async () => { - const dir = await makeTempDir(); - const sessionFile = path.join(dir, "long-header-session.jsonl"); - const longCwd = `/tmp/${"deep/".repeat(120)}`; - const header = { - type: "session", - version: CURRENT_SESSION_VERSION, - id: "long-header-session", - timestamp: "2026-06-18T00:00:00.000Z", - cwd: longCwd, - }; - const userEntry = { - type: "message", - id: "user-1", - parentId: null, - timestamp: "2026-06-18T00:00:01.000Z", - message: { role: "user", content: "resume me" }, - }; - await fs.writeFile( - sessionFile, - `${JSON.stringify(header)}\n${JSON.stringify(userEntry)}\n`, - "utf8", - ); - - expect(Buffer.byteLength(JSON.stringify(header), "utf8")).toBeGreaterThan(512); - expect(loadEntriesFromFile(sessionFile)).toHaveLength(2); - expect(findMostRecentSession(dir)).toBe(sessionFile); - expect(SessionManager.continueRecent(longCwd, dir).getSessionFile()).toBe(sessionFile); - }); - - it("does not continue a different cwd from a colliding session directory", async () => { - const dir = await makeTempDir(); - const cwdA = "/home/alice/dev/client/app"; - const cwdB = "/home/alice/dev/client-app"; - const sessionA = path.join(dir, "session-a.jsonl"); - const sessionB = path.join(dir, "session-b.jsonl"); - const headerA = buildSessionHeader(cwdA, "session-a"); - const headerB = buildSessionHeader(cwdB, "session-b"); - - await fs.writeFile(sessionA, `${JSON.stringify(headerA)}\n`, "utf8"); - await fs.writeFile(sessionB, `${JSON.stringify(headerB)}\n`, "utf8"); - await fs.utimes( - sessionA, - new Date("2026-06-18T00:00:00.000Z"), - new Date("2026-06-18T00:00:00.000Z"), - ); - await fs.utimes( - sessionB, - new Date("2026-06-18T00:00:01.000Z"), - new Date("2026-06-18T00:00:01.000Z"), - ); - - expect(findMostRecentSession(dir)).toBe(sessionB); - expect(findMostRecentSession(dir, cwdA)).toBe(sessionA); - expect(SessionManager.continueRecent(cwdA, dir).getSessionFile()).toBe(sessionA); - await expect(SessionManager.list(cwdA, dir)).resolves.toEqual([ - expect.objectContaining({ path: sessionA, cwd: cwdA }), - ]); - }); - - it("skips oversized recent session headers instead of hiding valid sessions", async () => { - const dir = await makeTempDir(); - const validSessionFile = path.join(dir, "valid-session.jsonl"); - const oversizedSessionFile = path.join(dir, "oversized-header-session.jsonl"); - const validHeader = { - type: "session", - version: CURRENT_SESSION_VERSION, - id: "valid-session", - timestamp: "2026-06-18T00:00:00.000Z", - cwd: "/tmp/task-repo", - }; - const oversizedHeader = { - type: "session", - version: CURRENT_SESSION_VERSION, - id: "oversized-header-session", - timestamp: "2026-06-18T00:00:01.000Z", - cwd: `/tmp/${"deep/".repeat(14_000)}`, - }; - - await fs.writeFile(validSessionFile, `${JSON.stringify(validHeader)}\n`, "utf8"); - await fs.writeFile(oversizedSessionFile, `${JSON.stringify(oversizedHeader)}\n`, "utf8"); - await fs.utimes( - validSessionFile, - new Date("2026-06-18T00:00:00.000Z"), - new Date("2026-06-18T00:00:00.000Z"), - ); - await fs.utimes( - oversizedSessionFile, - new Date("2026-06-18T00:00:01.000Z"), - new Date("2026-06-18T00:00:01.000Z"), - ); - - expect(Buffer.byteLength(JSON.stringify(oversizedHeader), "utf8")).toBeGreaterThan(64 * 1024); - expect(findMostRecentSession(dir)).toBe(validSessionFile); - }); - it("still migrates old transcript versions while bypassing the warm cache", async () => { const dir = await makeTempDir(); const sessionFile = path.join(dir, "session.jsonl"); @@ -3313,47 +3216,6 @@ describe("parseSessionEntries", () => { ), ).toBe(true); }); - - it("buildSessionInfo logs warning for malformed lines via SessionManager.list", async () => { - const warnSpy = vi.spyOn(Logger, "logWarn").mockImplementation(() => {}); - const dir = await makeTempDir(); - const sessionFile = path.join(dir, "session.jsonl"); - const header = buildSessionHeader(dir); - const content = [ - JSON.stringify(header), - "not valid json {{{", - JSON.stringify(buildMessageEntry(1, null)), - ].join("\n"); - await fs.writeFile(sessionFile, content, "utf8"); - - const sessions = await SessionManager.list(dir, dir); - - expect(sessions).toHaveLength(1); - expect(warnSpy).toHaveBeenCalled(); - expect( - warnSpy.mock.calls.some((call) => - call[0].includes("buildSessionInfo: skipped 1 malformed JSONL line"), - ), - ).toBe(true); - }); - - it("buildSessionInfo does not log warning for clean session listing", async () => { - const warnSpy = vi.spyOn(Logger, "logWarn").mockImplementation(() => {}); - const dir = await makeTempDir(); - const sessionFile = path.join(dir, "session.jsonl"); - const header = buildSessionHeader(dir); - const content = [JSON.stringify(header), JSON.stringify(buildMessageEntry(1, null))].join("\n"); - await fs.writeFile(sessionFile, content, "utf8"); - - const sessions = await SessionManager.list(dir, dir); - - expect(sessions).toHaveLength(1); - // buildSessionInfo must not log any warning for a clean listing. - const buildSessionInfoCalls = warnSpy.mock.calls.filter((call) => - call[0].includes("buildSessionInfo"), - ); - expect(buildSessionInfoCalls).toHaveLength(0); - }); }); function readMessageContent(entry: SessionEntry): unknown { diff --git a/src/agents/sessions/session-manager.ts b/src/agents/sessions/session-manager.ts index e23e1f008153..5a0fd4deaa7b 100644 --- a/src/agents/sessions/session-manager.ts +++ b/src/agents/sessions/session-manager.ts @@ -1,13 +1,11 @@ /** - * JSONL-backed session tree manager. + * Session tree manager backed by SQLite markers or explicit standalone files. * - * The public facade lives here; codec, storage, discovery, persistence, and - * branching behavior are split into focused internal modules. + * The public facade lives here; codec, storage, persistence, and branching + * behavior are split into focused internal modules. */ -import { existsSync, mkdirSync } from "node:fs"; -import { join, resolve } from "node:path"; +import { resolve } from "node:path"; import { loadTranscriptEventsSync } from "../../config/sessions/session-accessor.js"; -import { appendJsonlEntrySync } from "../../config/sessions/transcript-jsonl.js"; import { CURRENT_SESSION_VERSION } from "../../config/sessions/version.js"; import type { ImageContent, Message, TextContent } from "../../llm/types.js"; import type { BashExecutionMessage, CustomMessage } from "./messages.js"; @@ -15,14 +13,11 @@ import { SessionManagerBranching } from "./session-manager-branching.js"; import type { SqliteSessionManagerPersistence } from "./session-manager-core.js"; import { getDefaultSessionDir, - loadEntriesFromFile, loadEntriesFromFileWithSnapshot, loadSqliteMarkedSessionFile, revalidateLoadedSessionFile, type LoadedSessionFile, } from "./session-manager-file.js"; -import { createSessionId } from "./session-manager-id.js"; -import { findMostRecentSession, listAllSessions, listSessions } from "./session-manager-list.js"; import type { AppendPersistenceOptions, FileEntry, @@ -32,8 +27,6 @@ import type { SessionContext, SessionEntry, SessionHeader, - SessionInfo, - SessionListProgress, SessionTreeNode, } from "./session-manager-types.js"; @@ -46,7 +39,6 @@ export { parseSessionEntries, } from "./session-manager-codec.js"; export { getDefaultSessionDir, loadEntriesFromFile } from "./session-manager-file.js"; -export { findMostRecentSession } from "./session-manager-list.js"; export type { BranchSummaryEntry, CompactionEntry, @@ -60,9 +52,7 @@ export type { SessionEntry, SessionEntryBase, SessionHeader, - SessionInfo, SessionInfoEntry, - SessionListProgress, SessionMessageEntry, SessionTreeNode, ThinkingLevelChangeEntry, @@ -287,63 +277,9 @@ export class SessionManager extends SessionManagerBranching { return new SessionManager(cwd, directory, path, true, loaded); } - static continueRecent(cwd: string, sessionDir?: string): SessionManager { - const directory = sessionDir ?? getDefaultSessionDir(cwd); - const mostRecent = findMostRecentSession(directory, cwd); - return mostRecent - ? new SessionManager(cwd, directory, mostRecent, true) - : new SessionManager(cwd, directory, undefined, true); - } - static inMemory(cwd: string = process.cwd()): SessionManager { return new SessionManager(cwd, "", 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}`); - } - if (!sourceEntries.some((entry) => entry.type === "session")) { - throw new Error(`Cannot fork: source session has no header: ${sourcePath}`); - } - - const directory = sessionDir ?? getDefaultSessionDir(targetCwd); - if (!existsSync(directory)) { - mkdirSync(directory, { recursive: true }); - } - const newSessionId = createSessionId(); - const timestamp = new Date().toISOString(); - const fileTimestamp = timestamp.replace(/[:.]/g, "-"); - const newSessionFile = join(directory, `${fileTimestamp}_${newSessionId}.jsonl`); - const newHeader: SessionHeader = { - type: "session", - version: CURRENT_SESSION_VERSION, - id: newSessionId, - timestamp, - cwd: targetCwd, - parentSession: sourcePath, - }; - appendJsonlEntrySync(newSessionFile, newHeader); - for (const entry of sourceEntries) { - if (entry.type !== "session") { - appendJsonlEntrySync(newSessionFile, entry); - } - } - return new SessionManager(targetCwd, directory, newSessionFile, true); - } - - static async list( - cwd: string, - sessionDir?: string, - onProgress?: SessionListProgress, - ): Promise { - return await listSessions(cwd, sessionDir ?? getDefaultSessionDir(cwd), onProgress); - } - - static async listAll(onProgress?: SessionListProgress): Promise { - return await listAllSessions(onProgress); - } } export type ReadonlySessionManager = Pick< diff --git a/src/config/sessions/transcript-jsonl.ts b/src/config/sessions/transcript-jsonl.ts index cb5b90aeb8be..9942b36578ba 100644 --- a/src/config/sessions/transcript-jsonl.ts +++ b/src/config/sessions/transcript-jsonl.ts @@ -43,14 +43,6 @@ export function writeJsonlEntriesSync(filePath: string, entries: readonly unknow return content; } -export function appendJsonlEntrySync( - filePath: string, - entry: unknown, - options?: { prefixNewline?: boolean }, -): string { - return appendSerializedJsonlEntrySync(filePath, serializeJsonlEntry(entry), options); -} - export function appendSerializedJsonlEntrySync( filePath: string, serializedEntry: string, diff --git a/src/gateway/server.sessions.compaction.test.ts b/src/gateway/server.sessions.compaction.test.ts index 1464fb6687c5..2c86ff491f58 100644 --- a/src/gateway/server.sessions.compaction.test.ts +++ b/src/gateway/server.sessions.compaction.test.ts @@ -285,7 +285,6 @@ test("sessions.compaction.* lists checkpoints and branches or restores from comp expect(checkpoint.payload?.checkpoint.preCompaction.sessionFile).toBeUndefined(); const sessionManagerOpenSpy = vi.spyOn(SessionManager, "open"); - const sessionManagerForkFromSpy = vi.spyOn(SessionManager, "forkFrom"); let branched: Awaited< ReturnType< typeof rpcReq<{ @@ -319,10 +318,8 @@ test("sessions.compaction.* lists checkpoints and branches or restores from comp checkpointId: "checkpoint-1", }); expect(sessionManagerOpenSpy).not.toHaveBeenCalled(); - expect(sessionManagerForkFromSpy).not.toHaveBeenCalled(); } finally { sessionManagerOpenSpy.mockRestore(); - sessionManagerForkFromSpy.mockRestore(); } expect(branched.ok).toBe(true); expect(branched.payload?.sourceKey).toBe("agent:main:main"); @@ -351,7 +348,6 @@ test("sessions.compaction.* lists checkpoints and branches or restores from comp expect(branchedEntry?.compactionCheckpoints).toBeUndefined(); const restoreSessionManagerOpenSpy = vi.spyOn(SessionManager, "open"); - const restoreSessionManagerForkFromSpy = vi.spyOn(SessionManager, "forkFrom"); let restored: Awaited< ReturnType< typeof rpcReq<{ @@ -385,10 +381,8 @@ test("sessions.compaction.* lists checkpoints and branches or restores from comp checkpointId: "checkpoint-1", }); expect(restoreSessionManagerOpenSpy).not.toHaveBeenCalled(); - expect(restoreSessionManagerForkFromSpy).not.toHaveBeenCalled(); } finally { restoreSessionManagerOpenSpy.mockRestore(); - restoreSessionManagerForkFromSpy.mockRestore(); } expect(restored.ok).toBe(true); expect(restored.payload?.key).toBe("agent:main:main");