vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-04 11:51:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(agents): remove unused JSONL session paths (#112775)

Browse Source 
* refactor(agents): remove legacy session file discovery

* refactor(agents): remove dead sessions directory helper
This commit is contained in:
Peter Steinberger
2026-07-22 19:22:26 -04:00
committed by GitHub
parent cdb8d32bcc
commit e13bc7c63e
7 changed files with 4 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/agents/config.ts
View File

@@ -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");
}

297
src/agents/sessions/session-manager-list.ts
View File

@@ -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<SessionInfo | null> {
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<SessionInfo[]> {
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<SessionInfo[]> {
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<SessionInfo[]> {
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 [];
}
}

16
src/agents/sessions/session-manager-types.ts
View File

@@ -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;

138
src/agents/sessions/session-manager.test.ts
View File

@@ -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 {

72
src/agents/sessions/session-manager.ts
View File

@@ -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<SessionInfo[]> {
return await listSessions(cwd, sessionDir ?? getDefaultSessionDir(cwd), onProgress);
}
static async listAll(onProgress?: SessionListProgress): Promise<SessionInfo[]> {
return await listAllSessions(onProgress);
}
}
export type ReadonlySessionManager = Pick<

8
src/config/sessions/transcript-jsonl.ts
View File

@@ -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,

6
src/gateway/server.sessions.compaction.test.ts
View File

@@ -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");