perf(deeplake-fs): serve whole-session reads from the local event cache

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/<user>/<user>_<org>_<ws>_<sessionId>.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<unknown[]> {
+    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<Uint8Array> {
     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;

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/<user>/<user>_<org>_<ws>_<sessionId>.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 () => {