From 82d3ae3e20fdc818dd8cd06c1f3d116df98d21c9 Mon Sep 17 00:00:00 2001 From: khustup2 Date: Mon, 13 Jul 2026 22:13:20 +0000 Subject: [PATCH 1/2] fix(wiki-worker): read current session from a local cache, not a full DB re-scan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wiki-worker rebuilt each session's JSONL by re-`SELECT`ing the entire `message` column for the current session on every periodic (~every 50 events) and session-end trigger: SELECT message, creation_date FROM sessions WHERE path LIKE '%%' ORDER BY creation_date That `path LIKE '%...%'` is an unindexed full scan of the fat `message` column for the one session the client is *already* appending to. On a long mega-session it re-materializes the whole history on a cold backend — measured in prod at 15,234 rows / 72 MB, ~1.6-2.9s total_us with only ~180ms exec_us, rss ~1GB (org 5a000e2c); and 4,711 rows / 28 MB, 13-16s cold (org 1ad8b15e). The pooler's DISCARD ALL evicts the warm handle between statements, so every trigger re-pays the cold load. A secondary `SELECT DISTINCT path ... LIKE` scan of the same self-session added ~1.1s. The client already owns this data: capture writes each normalized event to the sessions table. Now it also mirrors that exact line into a local append-only cache (~/.claude/hooks/session-cache/.jsonl). The wiki-worker reads the cache (a few-ms local file read, zero backend load) instead of the fat scan, and derives the server path locally via buildSessionPath instead of the DISTINCT-path scan. The cache is a strict optimization, never a source of truth: the worker falls back to the DB SELECT whenever the cache is absent (session resumed on another machine), empty, or shorter than the offset already summarized (an incomplete local copy). Master opt-out: HIVEMIND_SESSION_EVENT_CACHE=0. Stale caches are pruned (14-day TTL) at session-end. Scope: claude-code path only. cursor/ and hermes/ wiki-workers carry the same pattern and can adopt this module as a fast follow. Tests: session-event-cache unit tests (roundtrip, opt-out, prune); wiki-worker local-cache path (no self-session SELECTs, DB fallback, short-cache refetch); capture wiring (append after successful INSERT only). --- src/hooks/capture.ts | 8 + src/hooks/session-end.ts | 5 + src/hooks/session-event-cache.ts | 106 ++++++++ src/hooks/spawn-wiki-worker.ts | 1 + src/hooks/wiki-worker.ts | 90 +++++-- tests/claude-code/capture-hook.test.ts | 20 ++ tests/claude-code/session-event-cache.test.ts | 149 ++++++++++++ .../wiki-worker-local-cache.test.ts | 229 ++++++++++++++++++ 8 files changed, 588 insertions(+), 20 deletions(-) create mode 100644 src/hooks/session-event-cache.ts create mode 100644 tests/claude-code/session-event-cache.test.ts create mode 100644 tests/claude-code/wiki-worker-local-cache.test.ts diff --git a/src/hooks/capture.ts b/src/hooks/capture.ts index 9f8768fe..9928e145 100644 --- a/src/hooks/capture.ts +++ b/src/hooks/capture.ts @@ -24,6 +24,7 @@ import { ensureSessionOwner, } from "./summary-state.js"; import { bundleDirFromImportMeta, spawnWikiWorker, wikiLog } from "./spawn-wiki-worker.js"; +import { appendSessionEvent } from "./session-event-cache.js"; import { tryStopCounterTrigger } from "../skillify/triggers.js"; import { reactSkillOpt } from "./shared/skillopt-hook.js"; import { EmbedClient } from "../embeddings/client.js"; @@ -184,6 +185,13 @@ async function main(): Promise { log("capture ok → cloud"); + // Mirror the event into the local per-session cache (row-for-row identical + // to the `message` column just INSERTed). The wiki-worker reads this instead + // of re-scanning the entire fat `message` column for the current session on + // every periodic / session-end summary trigger. Best-effort; DB stays the + // source of truth. Only reached after a successful INSERT above. + appendSessionEvent(input.session_id, line); + // Commit-driven KPI auto-extract is disabled for now — the // fire-and-forget sub-agent spawned per `git commit` (see // src/hooks/commit-kpi-extract.ts) consumed a high amount of tokens diff --git a/src/hooks/session-end.ts b/src/hooks/session-end.ts index 98d01d74..e95d3578 100644 --- a/src/hooks/session-end.ts +++ b/src/hooks/session-end.ts @@ -14,6 +14,7 @@ import { resolveDirConfig } from "../dir-config.js"; import { log as _log } from "../utils/debug.js"; import { bundleDirFromImportMeta, spawnWikiWorker, wikiLog } from "./spawn-wiki-worker.js"; import { tryAcquireLock, releaseLock, markSessionEnded } from "./summary-state.js"; +import { pruneStaleSessionEventCaches } from "./session-event-cache.js"; import { forceSessionEndTrigger } from "../skillify/triggers.js"; import { parseTranscript } from "../notifications/transcript-parser.js"; import { appendUsageRecord } from "../notifications/usage-tracker.js"; @@ -67,6 +68,10 @@ async function main(): Promise { // the activity window to lapse). Independent of the wiki-worker lock below. markSessionEnded(sessionId); + // Housekeeping: drop long-dead per-session event caches (14-day TTL). This + // session's own cache has a fresh mtime and is left intact for the worker. + try { pruneStaleSessionEventCaches(); } catch { /* best-effort */ } + const base = loadConfig(); if (!base) { log("no config"); return; } diff --git a/src/hooks/session-event-cache.ts b/src/hooks/session-event-cache.ts new file mode 100644 index 00000000..f3a4eb23 --- /dev/null +++ b/src/hooks/session-event-cache.ts @@ -0,0 +1,106 @@ +/** + * Local per-session event cache. + * + * The capture hook appends every normalized session event (the exact JSON + * line it also INSERTs into the sessions-table `message` column) to a local + * append-only file. The wiki-worker then reads this file instead of + * re-`SELECT`ing the entire fat `message` column for the *current* session on + * every periodic / session-end summary trigger. + * + * Why this exists: on a long "mega-session" (thousands of rows, tens of MB of + * message payload) the worker's + * SELECT message, creation_date FROM sessions + * WHERE path LIKE '%%' ORDER BY creation_date + * is an unindexed full scan of the fat `message` column. On a cold backend it + * materializes the whole session (hundreds of MB, 10-16 s) — and the periodic + * trigger re-pays that cost every ~50 events, re-reading the whole history the + * client is *already* appending to. The local cache is row-for-row identical + * to those DB rows, so the worker gets the same content from a local file read + * (a few ms, zero backend load). + * + * The cache is a strict optimization, never a source of truth: whenever it is + * absent (session resumed on another machine), empty, or shorter than the + * offset already summarized (an incomplete local copy), the worker falls back + * to the DB `SELECT`. + * + * File: ~/.claude/hooks/session-cache/.jsonl + */ + +import { + appendFileSync, mkdirSync, readFileSync, readdirSync, statSync, unlinkSync, +} from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { log as _log } from "../utils/debug.js"; + +const dlog = (msg: string) => _log("session-event-cache", msg); + +const CACHE_DIR = join(homedir(), ".claude", "hooks", "session-cache"); + +/** + * Master opt-out. Set HIVEMIND_SESSION_EVENT_CACHE=0 (or "false") to disable + * both the append (capture side) and the read (worker side), forcing the + * wiki-worker back onto the DB `SELECT` for every trigger. + */ +export function sessionEventCacheDisabled(): boolean { + const v = process.env.HIVEMIND_SESSION_EVENT_CACHE; + return v === "0" || v === "false"; +} + +export function sessionEventCachePath(sessionId: string): string { + return join(CACHE_DIR, `${sessionId}.jsonl`); +} + +/** + * Append one already-serialized event line to the session's cache. `line` + * must not contain a newline — capture builds it via `JSON.stringify`, which + * escapes embedded newlines, so one file line maps to exactly one event/row. + * Best-effort: never throws into the capture hot path. + */ +export function appendSessionEvent(sessionId: string, line: string): void { + if (!sessionId || sessionEventCacheDisabled()) return; + try { + mkdirSync(CACHE_DIR, { recursive: true }); + appendFileSync(sessionEventCachePath(sessionId), line + "\n"); + } catch (e: any) { + dlog(`append failed for ${sessionId}: ${e?.message ?? e}`); + } +} + +/** + * Read the session's cached event lines in append (chronological) order. + * Returns null when the cache is disabled, missing, or unreadable — the caller + * then falls back to the DB. Blank lines (including the trailing newline) are + * dropped so the returned length equals the event/row count. + */ +export function readSessionEventCache(sessionId: string): string[] | null { + if (!sessionId || sessionEventCacheDisabled()) return null; + try { + const raw = readFileSync(sessionEventCachePath(sessionId), "utf-8"); + return raw.split("\n").filter(l => l.length > 0); + } catch { + // ENOENT (never captured on this machine) or any read error → DB fallback. + return null; + } +} + +/** + * Best-effort removal of caches whose last write is older than `ttlMs`. Called + * opportunistically (session-end) so per-session files don't accumulate + * forever. A freshly-written (current-session) cache has a recent mtime and is + * never pruned. + */ +export function pruneStaleSessionEventCaches( + ttlMs: number = 14 * 24 * 3600 * 1000, + now: number = Date.now(), +): void { + try { + for (const name of readdirSync(CACHE_DIR)) { + if (!name.endsWith(".jsonl")) continue; + const p = join(CACHE_DIR, name); + try { + if (now - statSync(p).mtimeMs > ttlMs) unlinkSync(p); + } catch { /* ignore individual file errors */ } + } + } catch { /* dir missing → nothing to prune */ } +} diff --git a/src/hooks/spawn-wiki-worker.ts b/src/hooks/spawn-wiki-worker.ts index a983ae48..07482913 100644 --- a/src/hooks/spawn-wiki-worker.ts +++ b/src/hooks/spawn-wiki-worker.ts @@ -112,6 +112,7 @@ export function spawnWikiWorker(opts: SpawnOptions): void { sessionsTable: config.sessionsTableName, sessionId, userName: config.userName, + orgName: config.orgName, project: projectName, agent: opts.agent ?? "claude_code", pluginVersion, diff --git a/src/hooks/wiki-worker.ts b/src/hooks/wiki-worker.ts index b2174675..6c6d32d0 100644 --- a/src/hooks/wiki-worker.ts +++ b/src/hooks/wiki-worker.ts @@ -17,6 +17,8 @@ import { deeplakeClientHeader } from "../utils/client-header.js"; const dlog = (msg: string) => _log("wiki-worker", msg); import { finalizeSummary, releaseLock, readState } from "./summary-state.js"; +import { readSessionEventCache } from "./session-event-cache.js"; +import { buildSessionPath } from "../utils/session-path.js"; import { capLinesByBytes, stampOffset, WIKI_JSONL_MAX_BYTES } from "./wiki-offset.js"; import { uploadSummary } from "./upload-summary.js"; import { embedSummaryWithWarmup } from "../embeddings/embed-summary.js"; @@ -31,6 +33,7 @@ interface WorkerConfig { sessionsTable: string; sessionId: string; userName: string; + orgName: string; project: string; agent?: string; pluginVersion?: string; @@ -131,20 +134,44 @@ function cleanup(): void { async function main(): Promise { try { - // 1. Fetch session events from sessions table, reconstruct JSONL. - // Retry on an empty result: the async capture writes (or Deeplake read - // consistency) may simply be lagging behind SessionEnd under load. - wlog("fetching session events"); - const fetchEvents = () => query( + // 1. Load session events and reconstruct JSONL. + // + // Prefer the local per-session event cache the capture hook appends to as + // the session runs: it is row-for-row identical to the sessions-table + // `message` column, but reading it avoids re-scanning the entire fat + // `message` column on the backend for THIS session on every periodic / + // session-end trigger — the dominant cold-start cost on long + // "mega-sessions" (e.g. 15k rows / 72 MB re-materialized at ~2s each). The + // DB is still the source of truth: fall back to it whenever the cache is + // absent (session resumed on another machine), empty, or — checked once + // the offset is known — shorter than the offset already summarized. + const dbFetch = () => query( `SELECT message, creation_date FROM "${cfg.sessionsTable}" ` + `WHERE path LIKE '${esc(`/sessions/%${cfg.sessionId}%`)}' ORDER BY creation_date ASC` ); - let rows = await fetchEvents(); - for (let attempt = 1; rows.length === 0 && attempt <= EVENT_FETCH_RETRIES; attempt++) { - const delay = EVENT_FETCH_BACKOFF_MS * attempt; - wlog(`no events yet — retry ${attempt}/${EVENT_FETCH_RETRIES} in ${delay}ms`); - await sleep(delay); - rows = await fetchEvents(); + // Retry on an empty result: the async capture writes (or Deeplake read + // consistency) may simply be lagging behind SessionEnd under load. + const dbFetchWithRetry = async (): Promise[]> => { + let r = await dbFetch(); + for (let attempt = 1; r.length === 0 && attempt <= EVENT_FETCH_RETRIES; attempt++) { + const delay = EVENT_FETCH_BACKOFF_MS * attempt; + wlog(`no events yet — retry ${attempt}/${EVENT_FETCH_RETRIES} in ${delay}ms`); + await sleep(delay); + r = await dbFetch(); + } + return r; + }; + + let usedLocalCache = false; + let rows: Record[]; + const cachedLines = readSessionEventCache(cfg.sessionId); + if (cachedLines && cachedLines.length > 0) { + rows = cachedLines.map(message => ({ message })); + usedLocalCache = true; + wlog(`loaded ${rows.length} events from local cache`); + } else { + wlog("fetching session events"); + rows = await dbFetchWithRetry(); } if (rows.length === 0) { @@ -165,16 +192,28 @@ async function main(): Promise { return; } - const jsonlLines = rows.length; + let jsonlLines = rows.length; - // Derive the server path - const pathRows = await query( - `SELECT DISTINCT path FROM "${cfg.sessionsTable}" ` + - `WHERE path LIKE '${esc(`/sessions/%${cfg.sessionId}%`)}' LIMIT 1` - ); - const jsonlServerPath = pathRows.length > 0 - ? pathRows[0].path as string - : `/sessions/unknown/${cfg.sessionId}.jsonl`; + // Derive the server path. When the events came from the local cache we've + // already avoided the backend round-trip, so reproduce the canonical path + // locally rather than paying a second `SELECT DISTINCT path` scan of the + // same self-session (observed at ~1.1s on the 72 MB mega-session). The DB + // branch keeps its lookup for sessions whose path predates this code. + let jsonlServerPath: string; + if (usedLocalCache) { + jsonlServerPath = buildSessionPath( + { userName: cfg.userName, orgName: cfg.orgName, workspaceId: cfg.workspaceId }, + cfg.sessionId, + ); + } else { + const pathRows = await query( + `SELECT DISTINCT path FROM "${cfg.sessionsTable}" ` + + `WHERE path LIKE '${esc(`/sessions/%${cfg.sessionId}%`)}' LIMIT 1` + ); + jsonlServerPath = pathRows.length > 0 + ? pathRows[0].path as string + : `/sessions/unknown/${cfg.sessionId}.jsonl`; + } // 2. Determine how many rows were already summarized (resumed session). // The sidecar count is authoritative: finalizeSummary writes it after every @@ -216,6 +255,17 @@ async function main(): Promise { if (sidecarCount > prevOffset) prevOffset = sidecarCount; } + // Safety net for the local-cache path: if the cache holds fewer rows than + // the offset already summarized, it is an incomplete copy (e.g. the + // session was resumed on a different machine that captured the earlier + // rows). Slicing by `prevOffset` would then drop every genuinely-new row + // to nothing, so re-load the full session from the DB instead. + if (usedLocalCache && rows.length < prevOffset) { + wlog(`local cache (${rows.length}) < summarized offset (${prevOffset}) — refetching from DB`); + rows = await dbFetchWithRetry(); + jsonlLines = rows.length; + } + // Feed claude only the rows added since the last summary. Reprocessing the // full session on every run is what drives ENOBUFS / 120s-timeout failures // on long (4000+ event) sessions — a stuck offset re-summarizes everything. diff --git a/tests/claude-code/capture-hook.test.ts b/tests/claude-code/capture-hook.test.ts index 822591cc..8ecb865d 100644 --- a/tests/claude-code/capture-hook.test.ts +++ b/tests/claude-code/capture-hook.test.ts @@ -31,6 +31,7 @@ const debugLogMock = vi.fn(); const queryMock = vi.fn(); const ensureSessionsTableMock = vi.fn(); const apiCtorMock = vi.fn(); +const appendSessionEventMock = vi.fn(); vi.mock("../../src/utils/stdin.js", () => ({ readStdin: (...a: any[]) => stdinMock(...a) })); vi.mock("../../src/config.js", () => ({ loadConfig: (...a: any[]) => loadConfigMock(...a) })); @@ -63,6 +64,9 @@ vi.mock("../../src/embeddings/client.js", () => ({ warmup() { return Promise.resolve(false); } }, })); +vi.mock("../../src/hooks/session-event-cache.js", () => ({ + appendSessionEvent: (...a: any[]) => appendSessionEventMock(...a), +})); async function runHook(env: Record = {}): Promise { delete process.env.HIVEMIND_WIKI_WORKER; @@ -104,6 +108,7 @@ beforeEach(() => { queryMock.mockReset().mockResolvedValue([]); ensureSessionsTableMock.mockReset().mockResolvedValue(undefined); apiCtorMock.mockReset(); + appendSessionEventMock.mockReset(); }); afterEach(() => { vi.restoreAllMocks(); }); @@ -189,6 +194,21 @@ describe("capture hook — event-type branches", () => { expect(sql).toContain('"type":"user_message"'); expect(sql).toContain('"content":"hello"'); expect(debugLogMock).toHaveBeenCalledWith(expect.stringMatching(/^user session=sid-1$/)); + // The same normalized line is mirrored into the local per-session cache so + // the wiki-worker never has to re-`SELECT` the fat message column. + expect(appendSessionEventMock).toHaveBeenCalledTimes(1); + const [cachedSid, cachedLine] = appendSessionEventMock.mock.calls[0]; + expect(cachedSid).toBe("sid-1"); + expect(JSON.parse(cachedLine)).toMatchObject({ type: "user_message", content: "hello" }); + }); + + it("does NOT append to the local cache when the INSERT fails", async () => { + // A failed INSERT re-throws before the append — the cache must not diverge + // from the DB by recording an event that was never persisted. + vi.spyOn(process, "exit").mockImplementation(() => undefined as never); + queryMock.mockReset().mockRejectedValue(new Error("random SQL boom")); + await runHook(); + expect(appendSessionEventMock).not.toHaveBeenCalled(); }); it("tool_call: INSERT contains tool_name + serialized input/response", async () => { diff --git a/tests/claude-code/session-event-cache.test.ts b/tests/claude-code/session-event-cache.test.ts new file mode 100644 index 00000000..1be628f7 --- /dev/null +++ b/tests/claude-code/session-event-cache.test.ts @@ -0,0 +1,149 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { mkdtempSync, rmSync, existsSync, readFileSync, writeFileSync, utimesSync, mkdirSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +/** + * Unit tests for src/hooks/session-event-cache.ts. + * + * The module computes CACHE_DIR from os.homedir() at import time, so we mock + * node:os.homedir to a throwaway tmp dir and import the module fresh in each + * test via vi.resetModules(). + */ + +let home: string; + +vi.mock("node:os", async () => { + const actual = await vi.importActual("node:os"); + return { ...actual, homedir: () => home }; +}); + +type Mod = typeof import("../../src/hooks/session-event-cache.js"); + +async function load(): Promise { + vi.resetModules(); + return import("../../src/hooks/session-event-cache.js"); +} + +beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "sec-test-")); + delete process.env.HIVEMIND_SESSION_EVENT_CACHE; +}); + +afterEach(() => { + try { rmSync(home, { recursive: true, force: true }); } catch { /* ignore */ } + vi.restoreAllMocks(); +}); + +describe("session-event-cache — append + read roundtrip", () => { + it("appends lines and reads them back in order", async () => { + const m = await load(); + m.appendSessionEvent("sid", JSON.stringify({ type: "user_message", content: "a" })); + m.appendSessionEvent("sid", JSON.stringify({ type: "assistant_message", content: "b" })); + m.appendSessionEvent("sid", JSON.stringify({ type: "tool_call", tool_name: "Bash" })); + + const lines = m.readSessionEventCache("sid"); + expect(lines).not.toBeNull(); + expect(lines!).toHaveLength(3); + expect(JSON.parse(lines![0]).content).toBe("a"); + expect(JSON.parse(lines![1]).content).toBe("b"); + expect(JSON.parse(lines![2]).tool_name).toBe("Bash"); + }); + + it("each event maps to exactly one line even with embedded newlines", async () => { + const m = await load(); + // JSON.stringify escapes newlines, so a multi-line message is one file line. + m.appendSessionEvent("sid", JSON.stringify({ content: "line1\nline2\nline3" })); + const lines = m.readSessionEventCache("sid"); + expect(lines!).toHaveLength(1); + expect(JSON.parse(lines![0]).content).toBe("line1\nline2\nline3"); + }); + + it("read returns null when the cache file does not exist", async () => { + const m = await load(); + expect(m.readSessionEventCache("never-written")).toBeNull(); + }); + + it("read drops the trailing blank line (length equals event count)", async () => { + const m = await load(); + m.appendSessionEvent("sid", "{}"); + m.appendSessionEvent("sid", "{}"); + const raw = readFileSync(m.sessionEventCachePath("sid"), "utf-8"); + expect(raw.endsWith("\n")).toBe(true); // trailing newline present on disk + expect(m.readSessionEventCache("sid")!).toHaveLength(2); // but not counted + }); +}); + +describe("session-event-cache — opt-out flag", () => { + it("append is a no-op and read returns null when disabled", async () => { + process.env.HIVEMIND_SESSION_EVENT_CACHE = "0"; + const m = await load(); + m.appendSessionEvent("sid", "{}"); + expect(existsSync(m.sessionEventCachePath("sid"))).toBe(false); + expect(m.readSessionEventCache("sid")).toBeNull(); + }); + + it('"false" also disables', async () => { + process.env.HIVEMIND_SESSION_EVENT_CACHE = "false"; + const m = await load(); + m.appendSessionEvent("sid", "{}"); + expect(existsSync(m.sessionEventCachePath("sid"))).toBe(false); + }); + + it("read of an already-written cache returns null once disabled (forces DB fallback)", async () => { + const m1 = await load(); + m1.appendSessionEvent("sid", "{}"); + expect(m1.readSessionEventCache("sid")!).toHaveLength(1); + process.env.HIVEMIND_SESSION_EVENT_CACHE = "1"; // any non-off value keeps it on + const m2 = await load(); + expect(m2.readSessionEventCache("sid")!).toHaveLength(1); + process.env.HIVEMIND_SESSION_EVENT_CACHE = "0"; + const m3 = await load(); + expect(m3.readSessionEventCache("sid")).toBeNull(); + }); +}); + +describe("session-event-cache — empty / missing session id", () => { + it("ignores a blank session id on both append and read", async () => { + const m = await load(); + m.appendSessionEvent("", "{}"); + expect(m.readSessionEventCache("")).toBeNull(); + }); +}); + +describe("session-event-cache — prune", () => { + it("removes caches older than the TTL and keeps fresh ones", async () => { + const m = await load(); + m.appendSessionEvent("old", "{}"); + m.appendSessionEvent("fresh", "{}"); + const oldPath = m.sessionEventCachePath("old"); + const freshPath = m.sessionEventCachePath("fresh"); + // Age "old" 30 days back. + const now = Date.now(); + const thirtyDays = 30 * 24 * 3600 * 1000; + utimesSync(oldPath, new Date(now - thirtyDays), new Date(now - thirtyDays)); + + m.pruneStaleSessionEventCaches(14 * 24 * 3600 * 1000, now); + + expect(existsSync(oldPath)).toBe(false); + expect(existsSync(freshPath)).toBe(true); + }); + + it("is a safe no-op when the cache dir does not exist", async () => { + const m = await load(); + expect(() => m.pruneStaleSessionEventCaches()).not.toThrow(); + }); + + it("ignores non-.jsonl files in the cache dir", async () => { + const m = await load(); + m.appendSessionEvent("keep", "{}"); + const dir = join(home, ".claude", "hooks", "session-cache"); + mkdirSync(dir, { recursive: true }); + const strayPath = join(dir, "notes.txt"); + writeFileSync(strayPath, "hi"); + const old = Date.now() - 100 * 24 * 3600 * 1000; + utimesSync(strayPath, new Date(old), new Date(old)); + m.pruneStaleSessionEventCaches(); + expect(existsSync(strayPath)).toBe(true); // untouched: not a .jsonl + }); +}); diff --git a/tests/claude-code/wiki-worker-local-cache.test.ts b/tests/claude-code/wiki-worker-local-cache.test.ts new file mode 100644 index 00000000..fb0d102b --- /dev/null +++ b/tests/claude-code/wiki-worker-local-cache.test.ts @@ -0,0 +1,229 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { mkdtempSync, rmSync, writeFileSync, readFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +/** + * Integration tests for the wiki-worker's local-cache fast path. + * + * The worker prefers the local per-session event cache over re-`SELECT`ing the + * fat `message` column for the current session. Here we mock the cache module + * (so nothing touches the real ~/.claude) and assert: + * - when the cache has rows, NO `SELECT message ...` and NO + * `SELECT DISTINCT path ...` self-session scan is issued; + * - the server path is derived locally via buildSessionPath; + * - an absent cache falls back to the DB SELECT (existing behavior); + * - a cache shorter than the summarized offset refetches from the DB. + */ + +const finalizeSummaryMock = vi.fn(); +const releaseLockMock = vi.fn(); +const readStateMock = vi.fn(); +const uploadSummaryMock = vi.fn(); +const execFileSyncMock = vi.fn(); +const embedSummaryMock = vi.fn(); +const readCacheMock = vi.fn(); + +vi.mock("../../src/hooks/summary-state.js", () => ({ + finalizeSummary: (...a: any[]) => finalizeSummaryMock(...a), + releaseLock: (...a: any[]) => releaseLockMock(...a), + readState: (...a: any[]) => readStateMock(...a), +})); +vi.mock("../../src/hooks/upload-summary.js", () => ({ + uploadSummary: (...a: any[]) => uploadSummaryMock(...a), +})); +vi.mock("../../src/hooks/session-event-cache.js", () => ({ + readSessionEventCache: (...a: any[]) => readCacheMock(...a), +})); +vi.mock("../../src/embeddings/client.js", () => ({ + EmbedClient: class { + async embed(text: string, kind: string) { return embedSummaryMock(text, kind); } + }, +})); +vi.mock("node:child_process", async () => { + const actual = await vi.importActual("node:child_process"); + return { ...actual, execFileSync: (...a: any[]) => execFileSyncMock(...a) }; +}); + +const originalFetch = global.fetch; +const fetchMock = vi.fn(); +const originalArgv2 = process.argv[2]; + +let rootDir: string; +let tmpDir: string; +let hooksDir: string; +let configPath: string; + +const defaultConfig = () => ({ + apiUrl: "http://fake.local", + token: "tok", + orgId: "org", + workspaceId: "default", + memoryTable: "memory", + sessionsTable: "sessions", + sessionId: "sid-cache", + userName: "alice", + orgName: "org", + project: "proj", + tmpDir, + claudeBin: "/fake/claude", + wikiLog: join(hooksDir, "wiki.log"), + hooksDir, + promptTemplate: "JSONL=__JSONL__ SUMMARY=__SUMMARY__ SID=__SESSION_ID__ PROJ=__PROJECT__ OFFSET=__PREV_OFFSET__ LINES=__JSONL_LINES__ SRC=__JSONL_SERVER_PATH__", +}); + +function writeConfig(overrides: Record = {}): void { + writeFileSync(configPath, JSON.stringify({ ...defaultConfig(), ...overrides })); +} + +function jsonResp(body: unknown, ok = true, status = 200): Response { + return { + ok, status, + json: async () => body, + text: async () => typeof body === "string" ? body : JSON.stringify(body), + } as Response; +} + +async function runWorker(): Promise { + vi.resetModules(); + global.fetch = fetchMock; + await import("../../src/hooks/wiki-worker.js"); + await new Promise(r => setImmediate(r)); + await new Promise(r => setImmediate(r)); + await new Promise(r => setImmediate(r)); +} + +/** Fetch mock that records SQL and answers the summary lookup as "no summary". */ +function trackingFetch(sqls: string[], summaryRows: unknown[][] = []): void { + fetchMock.mockImplementation(async (_url: string, init: any) => { + const sql = JSON.parse(init.body).query as string; + sqls.push(sql); + if (sql.startsWith("SELECT summary FROM")) return jsonResp({ columns: ["summary"], rows: summaryRows }); + if (sql.startsWith("SELECT message, creation_date")) { + // DB fallback path — a handful of generic rows. + return jsonResp({ + columns: ["message", "creation_date"], + rows: Array.from({ length: 6 }, (_, i) => [JSON.stringify({ type: "user_message", content: `db ${i}` }), "2026-01-01T00:00:00Z"]), + }); + } + if (sql.startsWith("SELECT DISTINCT path")) { + return jsonResp({ columns: ["path"], rows: [["/sessions/alice/db_path.jsonl"]] }); + } + return jsonResp({ columns: [], rows: [] }); + }); +} + +function writesSummary(): void { + execFileSyncMock.mockImplementation((_bin: string, args: string[]) => { + const prompt = args[args.indexOf("-p") + 1]; + const summaryPath = prompt.match(/SUMMARY=(\S+)/)![1]; + writeFileSync(summaryPath, "# Session sid-cache\n\n## What Happened\ndone.\n"); + return Buffer.from(""); + }); +} + +beforeEach(() => { + rootDir = mkdtempSync(join(tmpdir(), "wiki-local-cache-test-")); + tmpDir = join(rootDir, "tmp"); + hooksDir = join(rootDir, "hooks"); + require("node:fs").mkdirSync(tmpDir, { recursive: true }); + require("node:fs").mkdirSync(hooksDir, { recursive: true }); + configPath = join(rootDir, "config.json"); + writeConfig(); + process.argv[2] = configPath; + fetchMock.mockReset(); + finalizeSummaryMock.mockReset(); + releaseLockMock.mockReset(); + readStateMock.mockReset().mockReturnValue(null); + uploadSummaryMock.mockReset().mockResolvedValue({ path: "insert", summaryLength: 100, descLength: 20, sql: "..." }); + embedSummaryMock.mockReset().mockResolvedValue([0.1, 0.2, 0.3]); + execFileSyncMock.mockReset(); + readCacheMock.mockReset().mockReturnValue(null); +}); + +afterEach(() => { + global.fetch = originalFetch; + process.argv[2] = originalArgv2; + try { rmSync(rootDir, { recursive: true, force: true }); } catch { /* ignore */ } + vi.restoreAllMocks(); +}); + +describe("wiki-worker — local cache fast path", () => { + it("reads events from the local cache and issues NO self-session SELECTs", async () => { + const cached = Array.from({ length: 5 }, (_, i) => JSON.stringify({ type: "user_message", content: `cache ${i}` })); + readCacheMock.mockReturnValue(cached); + const sqls: string[] = []; + trackingFetch(sqls); + let capturedJsonl: string | null = null; + execFileSyncMock.mockImplementation((_bin: string, args: string[]) => { + const prompt = args[args.indexOf("-p") + 1]; + capturedJsonl = readFileSync(prompt.match(/JSONL=(\S+)/)![1], "utf-8"); + writeFileSync(prompt.match(/SUMMARY=(\S+)/)![1], "# s\n\n## What Happened\nok\n"); + return Buffer.from(""); + }); + + await runWorker(); + + // The dominant fat-column scan must NOT run, nor the secondary path scan. + expect(sqls.some(s => s.startsWith("SELECT message, creation_date"))).toBe(false); + expect(sqls.some(s => s.startsWith("SELECT DISTINCT path"))).toBe(false); + // Only the (cheap, exact-path) summary lookup + upload remain. + expect(sqls.some(s => s.startsWith("SELECT summary FROM"))).toBe(true); + + // Events came from the cache. + expect(capturedJsonl!.trim().split("\n")).toHaveLength(5); + expect(capturedJsonl).toContain("cache 0"); + + const log = readFileSync(join(hooksDir, "wiki.log"), "utf-8"); + expect(log).toContain("loaded 5 events from local cache"); + + // Server path derived locally via buildSessionPath (canonical 4-tuple). + const prompt = execFileSyncMock.mock.calls[0][1][execFileSyncMock.mock.calls[0][1].indexOf("-p") + 1] as string; + expect(prompt).toContain("SRC=/sessions/alice/alice_org_default_sid-cache.jsonl"); + expect(prompt).toContain("LINES=5"); + expect(finalizeSummaryMock).toHaveBeenCalledWith("sid-cache", 5); + }); + + it("falls back to the DB SELECT when the cache is absent", async () => { + readCacheMock.mockReturnValue(null); + const sqls: string[] = []; + trackingFetch(sqls); + writesSummary(); + + await runWorker(); + + expect(sqls.some(s => s.startsWith("SELECT message, creation_date"))).toBe(true); + expect(sqls.some(s => s.startsWith("SELECT DISTINCT path"))).toBe(true); + const log = readFileSync(join(hooksDir, "wiki.log"), "utf-8"); + expect(log).toContain("fetching session events"); + expect(finalizeSummaryMock).toHaveBeenCalledWith("sid-cache", 6); + }); + + it("falls back to the DB SELECT when the cache is empty", async () => { + readCacheMock.mockReturnValue([]); + const sqls: string[] = []; + trackingFetch(sqls); + writesSummary(); + + await runWorker(); + + expect(sqls.some(s => s.startsWith("SELECT message, creation_date"))).toBe(true); + }); + + it("refetches from the DB when the cache is shorter than the summarized offset", async () => { + // Cache has only 2 rows, but a prior summary recorded offset 12 — the + // cache is an incomplete local copy (session resumed elsewhere), so the + // worker must re-load the full session from the DB rather than slice to 0. + readCacheMock.mockReturnValue([JSON.stringify({ content: "a" }), JSON.stringify({ content: "b" })]); + const sqls: string[] = []; + trackingFetch(sqls, [["# Session X\n- **JSONL offset**: 12\n\n## What Happened\nprior"]]); + writesSummary(); + + await runWorker(); + + const log = readFileSync(join(hooksDir, "wiki.log"), "utf-8"); + expect(log).toContain("local cache (2) < summarized offset (12) — refetching from DB"); + // The DB SELECT was issued as the refetch. + expect(sqls.some(s => s.startsWith("SELECT message, creation_date"))).toBe(true); + }); +}); From 0254a7206ebdcb787f16ab839bd647e0b0df44bc Mon Sep 17 00:00:00 2001 From: khustup2 Date: Mon, 13 Jul 2026 22:27:53 +0000 Subject: [PATCH 2/2] fix(wiki-worker): extend local session-event cache to cursor + hermes Apply the same self-session-fetch elimination to the cursor/ and hermes/ wiki-worker forks: their capture hooks now mirror each event into the shared local cache (session-event-cache.ts), and their workers read it instead of re-`SELECT`ing the fat `message` column (`WHERE path LIKE '%%'`) and the secondary `SELECT DISTINCT path` on every periodic / session-end trigger. Server path is derived locally via buildSessionPath; DB fallback and the short-cache safety net match the claude-code path. orgName is threaded through both spawn configs. Tests: cursor/hermes wiki-worker suites gain a cache-path case (no self-session SELECTs, path derived locally) and a DB-fallback case; the cursor/hermes capture-hook tests now mock session-event-cache (hermetic, no real ~/.claude writes). --- src/hooks/cursor/capture.ts | 7 +++ src/hooks/cursor/spawn-wiki-worker.ts | 1 + src/hooks/cursor/wiki-worker.ts | 62 +++++++++++++++++++----- src/hooks/hermes/capture.ts | 7 +++ src/hooks/hermes/spawn-wiki-worker.ts | 1 + src/hooks/hermes/wiki-worker.ts | 62 +++++++++++++++++++----- tests/cursor/cursor-capture-hook.test.ts | 5 ++ tests/cursor/cursor-wiki-worker.test.ts | 59 ++++++++++++++++++++++ tests/hermes/hermes-capture-hook.test.ts | 5 ++ tests/hermes/hermes-wiki-worker.test.ts | 60 ++++++++++++++++++++++- 10 files changed, 246 insertions(+), 23 deletions(-) diff --git a/src/hooks/cursor/capture.ts b/src/hooks/cursor/capture.ts index cd7c5836..58680ea6 100644 --- a/src/hooks/cursor/capture.ts +++ b/src/hooks/cursor/capture.ts @@ -34,6 +34,7 @@ import { releaseLock, } from "../summary-state.js"; import { bundleDirFromImportMeta, spawnCursorWikiWorker, wikiLog } from "./spawn-wiki-worker.js"; +import { appendSessionEvent } from "../session-event-cache.js"; import { tryStopCounterTrigger } from "../../skillify/triggers.js"; import type { Config } from "../../config.js"; import { getInstalledVersion } from "../../utils/version-check.js"; @@ -180,6 +181,12 @@ async function main(): Promise { log("capture ok → cloud"); + // Mirror the event into the local per-session cache (row-for-row identical + // to the `message` column just INSERTed) so the wiki-worker reads it instead + // of re-scanning the fat `message` column. Best-effort; only after a + // successful INSERT above. + appendSessionEvent(sessionId, line); + maybeTriggerPeriodicSummary(sessionId, cwd, config); // Skillify Stop counter — afterAgentResponse is the assistant-complete event. diff --git a/src/hooks/cursor/spawn-wiki-worker.ts b/src/hooks/cursor/spawn-wiki-worker.ts index 4bfc2c5e..12b113e0 100644 --- a/src/hooks/cursor/spawn-wiki-worker.ts +++ b/src/hooks/cursor/spawn-wiki-worker.ts @@ -104,6 +104,7 @@ export function spawnCursorWikiWorker(opts: SpawnOptions): void { sessionsTable: config.sessionsTableName, sessionId, userName: config.userName, + orgName: config.orgName, project: projectName, pluginVersion, tmpDir, diff --git a/src/hooks/cursor/wiki-worker.ts b/src/hooks/cursor/wiki-worker.ts index a80c516b..f69c4ba5 100644 --- a/src/hooks/cursor/wiki-worker.ts +++ b/src/hooks/cursor/wiki-worker.ts @@ -18,6 +18,8 @@ import { buildTrailingPromptInvocation } from "../wiki-worker-spawn.js"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; import { finalizeSummary, releaseLock, readState } from "../summary-state.js"; +import { readSessionEventCache } from "../session-event-cache.js"; +import { buildSessionPath } from "../../utils/session-path.js"; import { capLinesByBytes, stampOffset, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js"; import { uploadSummary } from "../upload-summary.js"; import { log as _log } from "../../utils/debug.js"; @@ -36,6 +38,7 @@ interface WorkerConfig { sessionsTable: string; sessionId: string; userName: string; + orgName: string; project: string; pluginVersion?: string; tmpDir: string; @@ -114,27 +117,55 @@ function cleanup(): void { async function main(): Promise { try { - // 1. Fetch session events from sessions table - wlog("fetching session events"); - const rows = await query( + // 1. Load session events. Prefer the local per-session event cache the + // capture hook appends to as the session runs — it is row-for-row + // identical to the sessions-table `message` column but avoids re-scanning + // the entire fat `message` column on the backend for THIS session on every + // periodic / session-end trigger (the dominant cold-start cost on long + // mega-sessions). Falls back to the DB whenever the cache is absent + // (session resumed on another machine), empty, or — once the offset is + // known — shorter than the offset already summarized. + const dbFetch = () => query( `SELECT message, creation_date FROM "${cfg.sessionsTable}" ` + `WHERE path LIKE E'${esc(`/sessions/%${cfg.sessionId}%`)}' ORDER BY creation_date ASC` ); + let usedLocalCache = false; + let rows: Record[]; + const cachedLines = readSessionEventCache(cfg.sessionId); + if (cachedLines && cachedLines.length > 0) { + rows = cachedLines.map(message => ({ message })); + usedLocalCache = true; + wlog(`loaded ${rows.length} events from local cache`); + } else { + wlog("fetching session events"); + rows = await dbFetch(); + } + if (rows.length === 0) { wlog("no session events found — exiting"); return; } - const jsonlLines = rows.length; + let jsonlLines = rows.length; - const pathRows = await query( - `SELECT DISTINCT path FROM "${cfg.sessionsTable}" ` + - `WHERE path LIKE '${esc(`/sessions/%${cfg.sessionId}%`)}' LIMIT 1` - ); - const jsonlServerPath = pathRows.length > 0 - ? pathRows[0].path as string - : `/sessions/unknown/${cfg.sessionId}.jsonl`; + // Derive the server path locally when using the cache (avoids a second + // self-session `SELECT DISTINCT path` scan); the DB branch keeps its lookup. + let jsonlServerPath: string; + if (usedLocalCache) { + jsonlServerPath = buildSessionPath( + { userName: cfg.userName, orgName: cfg.orgName, workspaceId: cfg.workspaceId }, + cfg.sessionId, + ); + } else { + const pathRows = await query( + `SELECT DISTINCT path FROM "${cfg.sessionsTable}" ` + + `WHERE path LIKE '${esc(`/sessions/%${cfg.sessionId}%`)}' LIMIT 1` + ); + jsonlServerPath = pathRows.length > 0 + ? pathRows[0].path as string + : `/sessions/unknown/${cfg.sessionId}.jsonl`; + } // 2. Determine how many rows were already summarized (resumed session). // The sidecar count is authoritative: finalizeSummary writes it after every @@ -176,6 +207,15 @@ async function main(): Promise { if (sidecarCount > prevOffset) prevOffset = sidecarCount; } + // Safety net: a local cache shorter than the summarized offset is an + // incomplete copy (session resumed on another machine) — re-load the full + // session from the DB so no genuinely-new rows get sliced to nothing. + if (usedLocalCache && rows.length < prevOffset) { + wlog(`local cache (${rows.length}) < summarized offset (${prevOffset}) — refetching from DB`); + rows = await dbFetch(); + jsonlLines = rows.length; + } + // Feed the agent only the rows added since the last summary. Reprocessing // the full session on every run is what drives ENOBUFS / 120s-timeout // failures on long (4000+ event) sessions — a stuck offset re-summarizes diff --git a/src/hooks/hermes/capture.ts b/src/hooks/hermes/capture.ts index 836b22f4..68956257 100644 --- a/src/hooks/hermes/capture.ts +++ b/src/hooks/hermes/capture.ts @@ -35,6 +35,7 @@ import { releaseLock, } from "../summary-state.js"; import { bundleDirFromImportMeta, spawnHermesWikiWorker, wikiLog } from "./spawn-wiki-worker.js"; +import { appendSessionEvent } from "../session-event-cache.js"; import { tryStopCounterTrigger } from "../../skillify/triggers.js"; import type { Config } from "../../config.js"; import { getInstalledVersion } from "../../utils/version-check.js"; @@ -163,6 +164,12 @@ async function main(): Promise { log("capture ok → cloud"); + // Mirror the event into the local per-session cache (row-for-row identical + // to the `message` column just INSERTed) so the wiki-worker reads it instead + // of re-scanning the fat `message` column. Best-effort; only after a + // successful INSERT above. + appendSessionEvent(sessionId, line); + // SkillOpt: a pre_llm_call prompt is the user's reaction to a recently-used org skill. // Swallowed; no-op unless a judgment window is open for this session. reactSkillOpt(sessionId, reactPrompt, "hermes"); diff --git a/src/hooks/hermes/spawn-wiki-worker.ts b/src/hooks/hermes/spawn-wiki-worker.ts index 3456cfa3..de0dec8a 100644 --- a/src/hooks/hermes/spawn-wiki-worker.ts +++ b/src/hooks/hermes/spawn-wiki-worker.ts @@ -105,6 +105,7 @@ export function spawnHermesWikiWorker(opts: SpawnOptions): void { sessionsTable: config.sessionsTableName, sessionId, userName: config.userName, + orgName: config.orgName, project: projectName, pluginVersion, tmpDir, diff --git a/src/hooks/hermes/wiki-worker.ts b/src/hooks/hermes/wiki-worker.ts index 23512afb..c84b88c9 100644 --- a/src/hooks/hermes/wiki-worker.ts +++ b/src/hooks/hermes/wiki-worker.ts @@ -17,6 +17,8 @@ import { execFileSync } from "node:child_process"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; import { finalizeSummary, releaseLock, readState } from "../summary-state.js"; +import { readSessionEventCache } from "../session-event-cache.js"; +import { buildSessionPath } from "../../utils/session-path.js"; import { capLinesByBytes, stampOffset, WIKI_JSONL_MAX_BYTES } from "../wiki-offset.js"; import { uploadSummary } from "../upload-summary.js"; import { log as _log } from "../../utils/debug.js"; @@ -35,6 +37,7 @@ interface WorkerConfig { sessionsTable: string; sessionId: string; userName: string; + orgName: string; project: string; pluginVersion?: string; tmpDir: string; @@ -114,27 +117,55 @@ function cleanup(): void { async function main(): Promise { try { - // 1. Fetch session events from sessions table - wlog("fetching session events"); - const rows = await query( + // 1. Load session events. Prefer the local per-session event cache the + // capture hook appends to as the session runs — it is row-for-row + // identical to the sessions-table `message` column but avoids re-scanning + // the entire fat `message` column on the backend for THIS session on every + // periodic / session-end trigger (the dominant cold-start cost on long + // mega-sessions). Falls back to the DB whenever the cache is absent + // (session resumed on another machine), empty, or — once the offset is + // known — shorter than the offset already summarized. + const dbFetch = () => query( `SELECT message, creation_date FROM "${cfg.sessionsTable}" ` + `WHERE path LIKE E'${esc(`/sessions/%${cfg.sessionId}%`)}' ORDER BY creation_date ASC` ); + let usedLocalCache = false; + let rows: Record[]; + const cachedLines = readSessionEventCache(cfg.sessionId); + if (cachedLines && cachedLines.length > 0) { + rows = cachedLines.map(message => ({ message })); + usedLocalCache = true; + wlog(`loaded ${rows.length} events from local cache`); + } else { + wlog("fetching session events"); + rows = await dbFetch(); + } + if (rows.length === 0) { wlog("no session events found — exiting"); return; } - const jsonlLines = rows.length; + let jsonlLines = rows.length; - const pathRows = await query( - `SELECT DISTINCT path FROM "${cfg.sessionsTable}" ` + - `WHERE path LIKE '${esc(`/sessions/%${cfg.sessionId}%`)}' LIMIT 1` - ); - const jsonlServerPath = pathRows.length > 0 - ? pathRows[0].path as string - : `/sessions/unknown/${cfg.sessionId}.jsonl`; + // Derive the server path locally when using the cache (avoids a second + // self-session `SELECT DISTINCT path` scan); the DB branch keeps its lookup. + let jsonlServerPath: string; + if (usedLocalCache) { + jsonlServerPath = buildSessionPath( + { userName: cfg.userName, orgName: cfg.orgName, workspaceId: cfg.workspaceId }, + cfg.sessionId, + ); + } else { + const pathRows = await query( + `SELECT DISTINCT path FROM "${cfg.sessionsTable}" ` + + `WHERE path LIKE '${esc(`/sessions/%${cfg.sessionId}%`)}' LIMIT 1` + ); + jsonlServerPath = pathRows.length > 0 + ? pathRows[0].path as string + : `/sessions/unknown/${cfg.sessionId}.jsonl`; + } // 2. Determine how many rows were already summarized (resumed session). // The sidecar count is authoritative: finalizeSummary writes it after every @@ -176,6 +207,15 @@ async function main(): Promise { if (sidecarCount > prevOffset) prevOffset = sidecarCount; } + // Safety net: a local cache shorter than the summarized offset is an + // incomplete copy (session resumed on another machine) — re-load the full + // session from the DB so no genuinely-new rows get sliced to nothing. + if (usedLocalCache && rows.length < prevOffset) { + wlog(`local cache (${rows.length}) < summarized offset (${prevOffset}) — refetching from DB`); + rows = await dbFetch(); + jsonlLines = rows.length; + } + // Feed the agent only the rows added since the last summary. Reprocessing // the full session on every run is what drives ENOBUFS / 120s-timeout // failures on long (4000+ event) sessions — a stuck offset re-summarizes diff --git a/tests/cursor/cursor-capture-hook.test.ts b/tests/cursor/cursor-capture-hook.test.ts index 1d291e89..4e8110ea 100644 --- a/tests/cursor/cursor-capture-hook.test.ts +++ b/tests/cursor/cursor-capture-hook.test.ts @@ -16,6 +16,7 @@ const debugLogMock = vi.fn(); const queryMock = vi.fn(); const ensureSessionsTableMock = vi.fn(); const buildSessionPathMock = vi.fn(); +const appendSessionEventMock = vi.fn(); vi.mock("../../src/utils/stdin.js", () => ({ readStdin: (...a: unknown[]) => stdinMock(...a) })); vi.mock("../../src/config.js", () => ({ loadConfig: (...a: unknown[]) => loadConfigMock(...a) })); @@ -35,6 +36,9 @@ vi.mock("../../src/embeddings/client.js", () => ({ vi.mock("../../src/utils/session-path.js", () => ({ buildSessionPath: (...a: unknown[]) => buildSessionPathMock(...a), })); +vi.mock("../../src/hooks/session-event-cache.js", () => ({ + appendSessionEvent: (...a: unknown[]) => appendSessionEventMock(...a), +})); const validConfig = { token: "t", orgId: "o", orgName: "acme", workspaceId: "default", @@ -62,6 +66,7 @@ beforeEach(() => { queryMock.mockReset().mockResolvedValue([]); ensureSessionsTableMock.mockReset().mockResolvedValue(undefined); buildSessionPathMock.mockReset().mockReturnValue("/sessions/alice/foo.jsonl"); + appendSessionEventMock.mockReset(); }); afterEach(() => { vi.restoreAllMocks(); }); diff --git a/tests/cursor/cursor-wiki-worker.test.ts b/tests/cursor/cursor-wiki-worker.test.ts index 411d197f..7d79ca09 100644 --- a/tests/cursor/cursor-wiki-worker.test.ts +++ b/tests/cursor/cursor-wiki-worker.test.ts @@ -16,12 +16,16 @@ const readStateMock = vi.fn(); const uploadSummaryMock = vi.fn(); const execFileSyncMock = vi.fn(); const embedSummaryMock = vi.fn(); +const readCacheMock = vi.fn(); vi.mock("../../src/hooks/summary-state.js", () => ({ finalizeSummary: (...a: any[]) => finalizeSummaryMock(...a), releaseLock: (...a: any[]) => releaseLockMock(...a), readState: (...a: any[]) => readStateMock(...a), })); +vi.mock("../../src/hooks/session-event-cache.js", () => ({ + readSessionEventCache: (...a: any[]) => readCacheMock(...a), +})); vi.mock("../../src/hooks/upload-summary.js", () => ({ uploadSummary: (...a: any[]) => uploadSummaryMock(...a), })); @@ -51,6 +55,7 @@ const defaultConfig = () => ({ sessionsTable: "sessions", sessionId: "sid-cursor", userName: "alice", + orgName: "org", project: "proj", tmpDir, cursorBin: "/fake/cursor-agent", @@ -97,6 +102,7 @@ beforeEach(() => { uploadSummaryMock.mockReset().mockResolvedValue({ path: "insert", summaryLength: 80, descLength: 15, sql: "..." }); embedSummaryMock.mockReset().mockResolvedValue([0.1, 0.2, 0.3]); execFileSyncMock.mockReset(); + readCacheMock.mockReset().mockReturnValue(null); }); afterEach(() => { @@ -165,4 +171,57 @@ describe("cursor wiki-worker — behavior", () => { expect(uploadSummaryMock).not.toHaveBeenCalled(); expect(releaseLockMock).toHaveBeenCalledWith("sid-cursor"); }); + + it("reads events from the local cache and issues NO self-session SELECTs", async () => { + readCacheMock.mockReturnValue( + Array.from({ length: 4 }, (_, i) => JSON.stringify({ type: "user_message", content: `cache ${i}` })), + ); + const sqls: string[] = []; + fetchMock.mockImplementation(async (_u: string, init: any) => { + const sql = JSON.parse(init.body).query as string; + sqls.push(sql); + if (sql.startsWith("SELECT summary FROM")) return jsonResp({ columns: ["summary"], rows: [] }); + return jsonResp({ columns: [], rows: [] }); + }); + execFileSyncMock.mockImplementation((_bin: string, args: string[]) => { + const prompt = args[args.length - 1]; + writeFileSync(prompt.match(/SUMMARY=(\S+)/)![1], "# s\n\n## What Happened\nok\n"); + return Buffer.from(""); + }); + + await runWorker(); + + expect(sqls.some(s => s.startsWith("SELECT message, creation_date"))).toBe(false); + expect(sqls.some(s => s.startsWith("SELECT DISTINCT path"))).toBe(false); + const log = readFileSync(join(hooksDir, "wiki.log"), "utf-8"); + expect(log).toContain("loaded 4 events from local cache"); + const prompt = execFileSyncMock.mock.calls[0][1].at(-1) as string; + expect(prompt).toContain("SRC=/sessions/alice/alice_org_default_sid-cursor.jsonl"); + expect(prompt).toContain("LINES=4"); + expect(finalizeSummaryMock).toHaveBeenCalledWith("sid-cursor", 4); + }); + + it("falls back to the DB SELECT when the local cache is absent", async () => { + readCacheMock.mockReturnValue(null); + const sqls: string[] = []; + fetchMock.mockImplementation(async (_u: string, init: any) => { + const sql = JSON.parse(init.body).query as string; + sqls.push(sql); + if (sql.startsWith("SELECT message, creation_date")) { + return jsonResp({ columns: ["message", "creation_date"], rows: [[JSON.stringify({ type: "user_message", content: "db" }), "2026-04-20T00:00:00Z"]] }); + } + if (sql.startsWith("SELECT DISTINCT path")) return jsonResp({ columns: ["path"], rows: [["/sessions/alice/db.jsonl"]] }); + if (sql.startsWith("SELECT summary FROM")) return jsonResp({ columns: ["summary"], rows: [] }); + throw new Error(`unexpected query: ${sql}`); + }); + execFileSyncMock.mockImplementation((_bin: string, args: string[]) => { + writeFileSync(args.at(-1)!.match(/SUMMARY=(\S+)/)![1], "# s\n\n## What Happened\nok\n"); + return Buffer.from(""); + }); + + await runWorker(); + + expect(sqls.some(s => s.startsWith("SELECT message, creation_date"))).toBe(true); + expect(sqls.some(s => s.startsWith("SELECT DISTINCT path"))).toBe(true); + }); }); diff --git a/tests/hermes/hermes-capture-hook.test.ts b/tests/hermes/hermes-capture-hook.test.ts index 864df5e5..1611a4a9 100644 --- a/tests/hermes/hermes-capture-hook.test.ts +++ b/tests/hermes/hermes-capture-hook.test.ts @@ -15,6 +15,7 @@ const debugLogMock = vi.fn(); const queryMock = vi.fn(); const ensureSessionsTableMock = vi.fn(); const buildSessionPathMock = vi.fn(); +const appendSessionEventMock = vi.fn(); vi.mock("../../src/utils/stdin.js", () => ({ readStdin: (...a: unknown[]) => stdinMock(...a) })); vi.mock("../../src/config.js", () => ({ loadConfig: (...a: unknown[]) => loadConfigMock(...a) })); @@ -34,6 +35,9 @@ vi.mock("../../src/embeddings/client.js", () => ({ vi.mock("../../src/utils/session-path.js", () => ({ buildSessionPath: (...a: unknown[]) => buildSessionPathMock(...a), })); +vi.mock("../../src/hooks/session-event-cache.js", () => ({ + appendSessionEvent: (...a: unknown[]) => appendSessionEventMock(...a), +})); const validConfig = { token: "t", orgId: "o", orgName: "acme", workspaceId: "default", @@ -77,6 +81,7 @@ beforeEach(() => { queryMock.mockReset().mockResolvedValue([]); ensureSessionsTableMock.mockReset().mockResolvedValue(undefined); buildSessionPathMock.mockReset().mockReturnValue("/sessions/alice/foo.jsonl"); + appendSessionEventMock.mockReset(); }); afterEach(async () => { diff --git a/tests/hermes/hermes-wiki-worker.test.ts b/tests/hermes/hermes-wiki-worker.test.ts index 4e5fb283..5dd5ffc5 100644 --- a/tests/hermes/hermes-wiki-worker.test.ts +++ b/tests/hermes/hermes-wiki-worker.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; -import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from "node:fs"; +import { mkdtempSync, rmSync, writeFileSync, readFileSync, mkdirSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -17,7 +17,11 @@ const readStateMock = vi.fn(); const uploadSummaryMock = vi.fn(); const execFileSyncMock = vi.fn(); const embedSummaryMock = vi.fn(); +const readCacheMock = vi.fn(); +vi.mock("../../src/hooks/session-event-cache.js", () => ({ + readSessionEventCache: (...a: any[]) => readCacheMock(...a), +})); vi.mock("../../src/hooks/summary-state.js", () => ({ finalizeSummary: (...a: any[]) => finalizeSummaryMock(...a), releaseLock: (...a: any[]) => releaseLockMock(...a), @@ -52,6 +56,7 @@ const defaultConfig = () => ({ sessionsTable: "sessions", sessionId: "sid-hermes", userName: "alice", + orgName: "org", project: "proj", tmpDir, hermesBin: "/fake/hermes", @@ -99,6 +104,7 @@ beforeEach(() => { uploadSummaryMock.mockReset().mockResolvedValue({ path: "insert", summaryLength: 80, descLength: 15, sql: "..." }); embedSummaryMock.mockReset().mockResolvedValue([0.1, 0.2, 0.3]); execFileSyncMock.mockReset(); + readCacheMock.mockReset().mockReturnValue(null); }); afterEach(() => { @@ -167,4 +173,56 @@ describe("hermes wiki-worker — behavior", () => { expect(uploadSummaryMock).not.toHaveBeenCalled(); expect(releaseLockMock).toHaveBeenCalledWith("sid-hermes"); }); + + it("reads events from the local cache and issues NO self-session SELECTs", async () => { + readCacheMock.mockReturnValue( + Array.from({ length: 4 }, (_, i) => JSON.stringify({ type: "user_message", content: `cache ${i}` })), + ); + const sqls: string[] = []; + fetchMock.mockImplementation(async (_u: string, init: any) => { + const sql = JSON.parse(init.body).query as string; + sqls.push(sql); + if (sql.startsWith("SELECT summary FROM")) return jsonResp({ columns: ["summary"], rows: [] }); + return jsonResp({ columns: [], rows: [] }); + }); + execFileSyncMock.mockImplementation((_bin: string, args: string[]) => { + writeFileSync(args[1].match(/SUMMARY=(\S+)/)![1], "# s\n\n## What Happened\nok\n"); + return Buffer.from(""); + }); + + await runWorker(); + + expect(sqls.some(s => s.startsWith("SELECT message, creation_date"))).toBe(false); + expect(sqls.some(s => s.startsWith("SELECT DISTINCT path"))).toBe(false); + const log = readFileSync(join(hooksDir, "wiki.log"), "utf-8"); + expect(log).toContain("loaded 4 events from local cache"); + const prompt = execFileSyncMock.mock.calls[0][1][1] as string; + expect(prompt).toContain("SRC=/sessions/alice/alice_org_default_sid-hermes.jsonl"); + expect(prompt).toContain("LINES=4"); + expect(finalizeSummaryMock).toHaveBeenCalledWith("sid-hermes", 4); + }); + + it("falls back to the DB SELECT when the local cache is absent", async () => { + readCacheMock.mockReturnValue(null); + const sqls: string[] = []; + fetchMock.mockImplementation(async (_u: string, init: any) => { + const sql = JSON.parse(init.body).query as string; + sqls.push(sql); + if (sql.startsWith("SELECT message, creation_date")) { + return jsonResp({ columns: ["message", "creation_date"], rows: [[JSON.stringify({ type: "user_message", content: "db" }), "2026-04-20T00:00:00Z"]] }); + } + if (sql.startsWith("SELECT DISTINCT path")) return jsonResp({ columns: ["path"], rows: [["/sessions/alice/db.jsonl"]] }); + if (sql.startsWith("SELECT summary FROM")) return jsonResp({ columns: ["summary"], rows: [] }); + throw new Error(`unexpected query: ${sql}`); + }); + execFileSyncMock.mockImplementation((_bin: string, args: string[]) => { + writeFileSync(args[1].match(/SUMMARY=(\S+)/)![1], "# s\n\n## What Happened\nok\n"); + return Buffer.from(""); + }); + + await runWorker(); + + expect(sqls.some(s => s.startsWith("SELECT message, creation_date"))).toBe(true); + expect(sqls.some(s => s.startsWith("SELECT DISTINCT path"))).toBe(true); + }); });