From 693c7196f0e3731d064147c717258a1d913e3aba Mon Sep 17 00:00:00 2001 From: khustup2 Date: Fri, 24 Jul 2026 19:29:14 +0000 Subject: [PATCH] perf(deeplake-fs): serve whole-session reads from the local event cache The VFS whole-session read (readFile/readFileBuffer) issued an unbounded SELECT message FROM sessions WHERE path = ... ORDER BY creation_date ASC, re-materializing the entire fat message column on every fresh process. On mega-sessions (thousands of rows, tens of MB) that is a multi-second cold read, and it is the dominant driver of fat-payload p95 for orgs whose sessions are re-read repeatedly (observed 6+ s per read). The capture hook already appends every event to a local per-session cache (session-event-cache.ts, row-for-row identical to the message column); the wiki-workers already prefer it. Wire the same cache-first strategy into the VFS read: recover the sessionId from the path and read the local cache first, falling back to the full ordered DB read on any miss (session captured elsewhere, cache disabled/pruned, unreadable). Both readFile and readFileBuffer now share one loadSessionMessages helper. Content is provably identical: both cache lines and DB rows flow through joinSessionMessages -> normalizeContent, which JSON.parses each row and rebuilds output from parsed fields, so jsonb re-serialization vs the cached string cannot diverge. The DB fallback stays unbounded because the VFS needs the whole session (cat/grep), unlike the wiki-worker's bounded tail read. --- src/shell/deeplake-fs.ts | 64 ++++++++++--- tests/claude-code/deeplake-fs.test.ts | 125 ++++++++++++++++++++++++++ 2 files changed, 176 insertions(+), 13 deletions(-) diff --git a/src/shell/deeplake-fs.ts b/src/shell/deeplake-fs.ts index 95770db5..257091aa 100644 --- a/src/shell/deeplake-fs.ts +++ b/src/shell/deeplake-fs.ts @@ -8,6 +8,7 @@ import type { FileContent, BufferEncoding, } from "just-bash"; import { normalizeContent, emptySessionBodyNotice } from "./grep-core.js"; +import { readSessionEventCache } from "../hooks/session-event-cache.js"; import { EmbedClient } from "../embeddings/client.js"; import { embeddingSqlLiteral } from "../embeddings/sql.js"; import { embeddingsDisabled } from "../embeddings/disable.js"; @@ -92,6 +93,20 @@ function fsErr(code: string, msg: string, path: string): Error { return Object.assign(new Error(`${code}: ${msg}, '${path}'`), { code }); } +// Recover the bare capture sessionId from a VFS session path so a whole-session +// read can hit the local event cache. Path shape (see buildSessionPath): +// /sessions//___.jsonl +// The sessionId is a UUID (no underscores), so it is the last `_`-delimited +// segment of the filename stem — recoverable even when user/org/ws contain +// underscores. A non-matching path yields a value that simply misses the cache +// (→ DB fallback), so this is best-effort and never load-bearing. +function sessionIdFromVfsPath(p: string): string | null { + const file = p.split("/").pop() ?? ""; + if (!file.endsWith(".jsonl")) return null; + const sid = file.slice(0, -".jsonl".length).split("_").pop() ?? ""; + return sid.length > 0 ? sid : null; +} + // ── types ───────────────────────────────────────────────────────────────────── interface FileMeta { size: number; mime: string; mtime: Date; } @@ -708,6 +723,33 @@ export class DeeplakeFs implements IFileSystem { // ── IFileSystem: reads ──────────────────────────────────────────────────── + /** + * Load a session's message rows for a whole-file read. + * + * Prefers the local per-session event cache the capture hook appends to + * (row-for-row identical to the sessions-table `message` column) so a re-read + * of a fat "mega-session" costs a few ms + zero backend load instead of + * re-materializing the entire `message` column (tens of MB, multi-second cold + * on the backend). The cache is a strict optimization: on any miss — session + * captured on another machine, cache disabled/pruned, or unreadable — we fall + * back to the full ordered DB read, which stays the source of truth. This + * mirrors the wiki-worker's cache-first strategy (see session-event-cache.ts); + * the difference is the VFS needs the WHOLE session (cat/grep), so the DB + * fallback stays the unbounded `ORDER BY creation_date ASC` read. + */ + private async loadSessionMessages(p: string): Promise { + const sid = sessionIdFromVfsPath(p); + if (sid) { + const cached = readSessionEventCache(sid); + if (cached && cached.length > 0) return cached; + } + if (!this.sessionsTable) return []; + const rows = await this.client.query( + `SELECT message FROM "${this.sessionsTable}" WHERE path = '${esc(p)}' ORDER BY creation_date ASC` + ); + return rows.map((row) => row["message"]); + } + async readFileBuffer(path: string): Promise { const p = normPath(path); if (this.dirs.has(p) && !this.files.has(p)) throw fsErr("EISDIR", "illegal operation on a directory", p); @@ -721,14 +763,12 @@ export class DeeplakeFs implements IFileSystem { const pend = this.pending.get(p); if (pend) { const buf = Buffer.from(pend.contentText, "utf-8"); this.files.set(p, buf); return buf; } - // 3. Session files: concatenate rows from sessions table + // 3. Session files: local event cache first, then full DB read. if (this.sessionPaths.has(p) && this.sessionsTable) { - const rows = await this.client.query( - `SELECT message FROM "${this.sessionsTable}" WHERE path = '${esc(p)}' ORDER BY creation_date ASC` - ); - if (rows.length === 0) throw fsErr("ENOENT", "no such file or directory", p); - const text = joinSessionMessages(p, rows.map((row) => row["message"])); - const buf = Buffer.from(text || emptySessionBodyNotice(rows.length), "utf-8"); + const messages = await this.loadSessionMessages(p); + if (messages.length === 0) throw fsErr("ENOENT", "no such file or directory", p); + const text = joinSessionMessages(p, messages) || emptySessionBodyNotice(messages.length); + const buf = Buffer.from(text, "utf-8"); this.files.set(p, buf); return buf; } @@ -789,13 +829,11 @@ export class DeeplakeFs implements IFileSystem { const pend = this.pending.get(p); if (pend) return pend.contentText; - // Session files: concatenate rows from sessions table, ordered by creation_date + // Session files: local event cache first, then full DB read. if (this.sessionPaths.has(p) && this.sessionsTable) { - const rows = await this.client.query( - `SELECT message FROM "${this.sessionsTable}" WHERE path = '${esc(p)}' ORDER BY creation_date ASC` - ); - if (rows.length === 0) throw fsErr("ENOENT", "no such file or directory", p); - const text = joinSessionMessages(p, rows.map((row) => row["message"])) || emptySessionBodyNotice(rows.length); + const messages = await this.loadSessionMessages(p); + if (messages.length === 0) throw fsErr("ENOENT", "no such file or directory", p); + const text = joinSessionMessages(p, messages) || emptySessionBodyNotice(messages.length); const buf = Buffer.from(text, "utf-8"); this.files.set(p, buf); return text; diff --git a/tests/claude-code/deeplake-fs.test.ts b/tests/claude-code/deeplake-fs.test.ts index fc80107d..1213c42b 100644 --- a/tests/claude-code/deeplake-fs.test.ts +++ b/tests/claude-code/deeplake-fs.test.ts @@ -4,6 +4,13 @@ vi.mock("../../src/docs/embed.js", () => ({ makeDocEmbedder: () => async () => null, makeQueryEmbedder: () => async () => null, })); +// Local session event cache: default to a miss (null) so every existing test +// exercises the DB read path unchanged. Individual tests override the return +// value to exercise the cache-first path. +const readSessionEventCacheMock = vi.fn<(sessionId: string) => string[] | null>(() => null); +vi.mock("../../src/hooks/session-event-cache.js", () => ({ + readSessionEventCache: (sessionId: string) => readSessionEventCacheMock(sessionId), +})); import { DeeplakeFs, guessMime } from "../../src/shell/deeplake-fs.js"; // ── Mock ManagedClient ──────────────────────────────────────────────────────── @@ -857,6 +864,124 @@ describe("readFile: session bodies", () => { }); }); +// ── Session reads: local event cache is preferred over the fat DB read ─────── +describe("readFile: session local-cache-first", () => { + const SID = "019e2d5c-fec2-72d1-81f8-4edf525479b8"; + // buildSessionPath shape: /sessions//___.jsonl + const PATH = `/sessions/ivo/ivo_Alka_default_${SID}.jsonl`; + + // A client that knows PATH as a session and, on the fat `SELECT message` + // read, returns `dbMessages`. Tracks whether that fat read was issued. + function makeFs(dbMessages: unknown[]) { + let fatRead = 0; + const client = { + ensureTable: vi.fn().mockResolvedValue(undefined), + query: vi.fn(async (sql: string) => { + if (sql.includes("SELECT path, MAX(size_bytes) as total_size")) { + return [{ path: PATH, total_size: 4096 }]; + } + if (sql.includes("SELECT message FROM")) { + fatRead++; + return dbMessages.map((message) => ({ message })); + } + return []; + }), + }; + return { client, fatRead: () => fatRead }; + } + + beforeEach(() => { + readSessionEventCacheMock.mockReset(); + readSessionEventCacheMock.mockReturnValue(null); + }); + + it("serves a session from the local cache without hitting the fat DB read", async () => { + readSessionEventCacheMock.mockReturnValue([ + "{\"type\":\"user_message\",\"content\":\"hello\"}", + "{\"type\":\"assistant_message\",\"content\":\"hi\"}", + ]); + const { client, fatRead } = makeFs([/* DB should never be consulted */]); + const fs = await DeeplakeFs.create(client as never, "memory", "/", "sessions"); + + const text = await fs.readFile(PATH); + + expect(readSessionEventCacheMock).toHaveBeenCalledWith(SID); + expect(text).toBe("[user] hello\n[assistant] hi"); + expect(fatRead()).toBe(0); // no `SELECT message` fat read + }); + + it("produces identical content from cache lines and from DB rows", async () => { + const events = [ + "{\"type\":\"user_message\",\"content\":\"hello\"}", + "{\"type\":\"assistant_message\",\"content\":\"hi\"}", + ]; + // DB path (cache miss) + const a = makeFs(events); + const fsDb = await DeeplakeFs.create(a.client as never, "memory", "/", "sessions"); + const fromDb = await fsDb.readFile(PATH); + // Cache path (same rows served locally) + readSessionEventCacheMock.mockReturnValue(events); + const b = makeFs([]); + const fsCache = await DeeplakeFs.create(b.client as never, "memory", "/", "sessions"); + const fromCache = await fsCache.readFile(PATH); + + expect(fromCache).toBe(fromDb); + expect(a.fatRead()).toBe(1); + expect(b.fatRead()).toBe(0); + }); + + it("falls back to the DB read when the cache misses (null)", async () => { + readSessionEventCacheMock.mockReturnValue(null); + const { client, fatRead } = makeFs(["{\"type\":\"user_message\",\"content\":\"bye\"}"]); + const fs = await DeeplakeFs.create(client as never, "memory", "/", "sessions"); + + const text = await fs.readFile(PATH); + + expect(text).toBe("[user] bye"); + expect(fatRead()).toBe(1); + }); + + it("falls back to the DB read when the cache is present but empty", async () => { + readSessionEventCacheMock.mockReturnValue([]); + const { client, fatRead } = makeFs(["{\"type\":\"user_message\",\"content\":\"bye\"}"]); + const fs = await DeeplakeFs.create(client as never, "memory", "/", "sessions"); + + await fs.readFile(PATH); + + expect(fatRead()).toBe(1); + }); + + it("readFileBuffer also prefers the local cache", async () => { + readSessionEventCacheMock.mockReturnValue(["{\"type\":\"user_message\",\"content\":\"hello\"}"]); + const { client, fatRead } = makeFs([]); + const fs = await DeeplakeFs.create(client as never, "memory", "/", "sessions"); + + const buf = await fs.readFileBuffer(PATH); + + expect(Buffer.from(buf).toString("utf-8")).toBe("[user] hello"); + expect(fatRead()).toBe(0); + }); + + it("recovers the sessionId even when user/org/workspace contain underscores", async () => { + const path = `/sessions/my_user/my_user_my_org_my_ws_${SID}.jsonl`; + const client = { + ensureTable: vi.fn().mockResolvedValue(undefined), + query: vi.fn(async (sql: string) => { + if (sql.includes("SELECT path, MAX(size_bytes) as total_size")) { + return [{ path, total_size: 10 }]; + } + return []; + }), + }; + readSessionEventCacheMock.mockReturnValue(["{\"type\":\"user_message\",\"content\":\"hi\"}"]); + const fs = await DeeplakeFs.create(client as never, "memory", "/", "sessions"); + + await fs.readFile(path); + + expect(readSessionEventCacheMock).toHaveBeenCalledWith(SID); + }); +}); + // ── Upsert: id stability & dates ───────────────────────────────────────────── describe("flush upsert", () => { it("INSERT for new file sets id, creation_date and last_update_date", async () => {