Skip to content

Commit a918b42

Browse files
amknightamknight
authored
Merge 0f1f377 into 66b91d7 
2 parents 66b91d7 + 0f1f377 commit a918b42

2 files changed

Lines changed: 262 additions & 5 deletions

File tree

‎packages/memory-host-sdk/src/host/session-files.test.ts‎

Lines changed: 134 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,14 @@
11
// Memory Host SDK tests cover session files behavior.
22
import fsSync from "node:fs";
3+
import fsPromises from "node:fs/promises";
34
import os from "node:os";
45
import path from "node:path";
5-
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
6+
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
7+
import { emitSessionTranscriptUpdate } from "../../../../src/sessions/transcript-events.js";
68
import {
79
buildSessionEntry,
810
listSessionFilesForAgent,
11+
resetSessionFilesListingCache,
912
sessionPathForFile,
1013
type SessionFileEntry,
1114
} from "./session-files.js";
@@ -74,6 +77,136 @@ describe("listSessionFilesForAgent", () => {
7477
});
7578
});
7679

80+
describe("listSessionFilesForAgent listing cache", () => {
81+
let sessionsDir: string;
82+
83+
beforeEach(() => {
84+
resetSessionFilesListingCache();
85+
sessionsDir = path.join(tmpDir, "agents", "main", "sessions");
86+
fsSync.mkdirSync(sessionsDir, { recursive: true });
87+
});
88+
89+
afterEach(() => {
90+
vi.restoreAllMocks();
91+
vi.useRealTimers();
92+
resetSessionFilesListingCache();
93+
});
94+
95+
function writeSession(name: string): void {
96+
fsSync.writeFileSync(path.join(sessionsDir, name), "");
97+
}
98+
99+
function baseNames(files: string[]): string[] {
100+
return files.map((filePath) => path.basename(filePath)).toSorted();
101+
}
102+
103+
it("coalesces concurrent reads into a single READDIR", async () => {
104+
writeSession("a.jsonl");
105+
const readdirSpy = vi.spyOn(fsPromises, "readdir");
106+
107+
const [first, second] = await Promise.all([
108+
listSessionFilesForAgent("main"),
109+
listSessionFilesForAgent("main"),
110+
]);
111+
112+
expect(readdirSpy).toHaveBeenCalledTimes(1);
113+
expect(first).toEqual(second);
114+
expect(baseNames(first)).toEqual(["a.jsonl"]);
115+
});
116+
117+
it("serves the cached snapshot within the TTL and skips repeat scans", async () => {
118+
writeSession("a.jsonl");
119+
const readdirSpy = vi.spyOn(fsPromises, "readdir");
120+
121+
const firstCall = await listSessionFilesForAgent("main");
122+
// Mutate the dir without emitting a transcript event; the cache must hold.
123+
writeSession("b.jsonl");
124+
const secondCall = await listSessionFilesForAgent("main");
125+
126+
expect(readdirSpy).toHaveBeenCalledTimes(1);
127+
expect(baseNames(firstCall)).toEqual(["a.jsonl"]);
128+
expect(baseNames(secondCall)).toEqual(["a.jsonl"]);
129+
});
130+
131+
it("returns an isolated copy so caller mutation cannot corrupt the cache", async () => {
132+
writeSession("a.jsonl");
133+
const readdirSpy = vi.spyOn(fsPromises, "readdir");
134+
135+
// Concurrent burst exercises the in-flight originator + joiner return paths;
136+
// a shared reference here would poison the snapshot for everyone.
137+
const [a, b] = await Promise.all([
138+
listSessionFilesForAgent("main"),
139+
listSessionFilesForAgent("main"),
140+
]);
141+
a.length = 0;
142+
b.push("/phantom.jsonl");
143+
144+
// Sequential cache hit exercises the snapshot return path.
145+
const cached = await listSessionFilesForAgent("main");
146+
cached.length = 0;
147+
148+
const afterMutation = await listSessionFilesForAgent("main");
149+
150+
// One scan total; no caller's mutation leaked into the cached snapshot.
151+
expect(readdirSpy).toHaveBeenCalledTimes(1);
152+
expect(baseNames(afterMutation)).toEqual(["a.jsonl"]);
153+
});
154+
155+
it("refreshes the snapshot after the TTL elapses", async () => {
156+
vi.useFakeTimers({ toFake: ["Date"] });
157+
writeSession("a.jsonl");
158+
const readdirSpy = vi.spyOn(fsPromises, "readdir");
159+
160+
await listSessionFilesForAgent("main");
161+
writeSession("b.jsonl");
162+
vi.setSystemTime(Date.now() + 1_001);
163+
const refreshed = await listSessionFilesForAgent("main");
164+
165+
expect(readdirSpy).toHaveBeenCalledTimes(2);
166+
expect(baseNames(refreshed)).toEqual(["a.jsonl", "b.jsonl"]);
167+
});
168+
169+
it("invalidates the snapshot when a transcript update lands for the agent", async () => {
170+
writeSession("a.jsonl");
171+
const readdirSpy = vi.spyOn(fsPromises, "readdir");
172+
173+
await listSessionFilesForAgent("main");
174+
writeSession("b.jsonl");
175+
emitSessionTranscriptUpdate({ sessionFile: path.join(sessionsDir, "b.jsonl") });
176+
const afterInvalidate = await listSessionFilesForAgent("main");
177+
178+
expect(readdirSpy).toHaveBeenCalledTimes(2);
179+
expect(baseNames(afterInvalidate)).toEqual(["a.jsonl", "b.jsonl"]);
180+
});
181+
182+
it("does not cache a transient scan failure and retries on the next call", async () => {
183+
writeSession("a.jsonl");
184+
const transient = Object.assign(new Error("nfs blip"), { code: "EIO" });
185+
const readdirSpy = vi.spyOn(fsPromises, "readdir").mockRejectedValueOnce(transient);
186+
187+
// A transient READDIR failure is surfaced to this caller as empty...
188+
const failed = await listSessionFilesForAgent("main");
189+
expect(failed).toEqual([]);
190+
// ...but must not be cached: the next call within the TTL re-scans and
191+
// recovers the real listing instead of serving the false-empty snapshot.
192+
const recovered = await listSessionFilesForAgent("main");
193+
expect(baseNames(recovered)).toEqual(["a.jsonl"]);
194+
expect(readdirSpy).toHaveBeenCalledTimes(2);
195+
});
196+
197+
it("caches an absent sessions dir as empty without re-scanning", async () => {
198+
fsSync.rmSync(sessionsDir, { recursive: true, force: true });
199+
const readdirSpy = vi.spyOn(fsPromises, "readdir");
200+
201+
const first = await listSessionFilesForAgent("main");
202+
const second = await listSessionFilesForAgent("main");
203+
204+
expect(first).toEqual([]);
205+
expect(second).toEqual([]);
206+
expect(readdirSpy).toHaveBeenCalledTimes(1);
207+
});
208+
});
209+
77210
describe("sessionPathForFile", () => {
78211
it("includes the owning agent id when the transcript lives under an agent sessions dir", () => {
79212
const absPath = path.join(

‎packages/memory-host-sdk/src/host/session-files.ts‎

Lines changed: 128 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import {
1616
isSessionArchiveArtifactName,
1717
isSilentReplyPayloadText,
1818
isUsageCountedSessionTranscriptFileName,
19+
onSessionTranscriptUpdate,
1920
parseUsageCountedSessionIdFromFileName,
2021
resolveSessionTranscriptsDirForAgent,
2122
stripInboundMetadata,
@@ -300,18 +301,141 @@ function classifySessionTranscriptFromSessionStore(absPath: string): {
300301
};
301302
}
302303

303-
export async function listSessionFilesForAgent(agentId: string): Promise<string[]> {
304-
const dir = resolveSessionTranscriptsDirForAgent(agentId);
304+
// `listSessionFilesForAgent` performs a directory READDIR that is expensive on
305+
// networked filesystems (NFS): resolving each entry's type turns one logical
306+
// scan into roughly one attribute fetch per session file. Several subsystems
307+
// (startup catch-up, incremental sync, dreaming, qmd export) scan the same
308+
// agent sessions dir in close succession, so we coalesce concurrent reads and
309+
// serve a short-lived snapshot between scans.
310+
//
311+
// Correctness comes from event-driven invalidation, not from the TTL: every
312+
// in-process mutation to a session-owned path (append, compaction, rewrite,
313+
// chat inject, command exec, and `.reset.`/`.deleted.` archive rotation) emits
314+
// `onSessionTranscriptUpdate`, which drops the affected directory's snapshot.
315+
// The TTL is only a backstop for writers that never reach our event bus, such
316+
// as another node sharing the NFS sessions dir or low-frequency archive prunes;
317+
// 1s keeps that eventual-consistency window tight.
318+
const SESSION_FILES_LISTING_TTL_MS = 1_000;
319+
// Bound the snapshot map so a process that touches many agents cannot grow it
320+
// without limit; agents are few in practice, so eviction is rare.
321+
const SESSION_FILES_LISTING_MAX_DIRS = 64;
322+
323+
type SessionFilesListingEntry = {
324+
// Shared in-flight read so concurrent callers issue a single READDIR.
325+
inFlight?: Promise<string[]>;
326+
// Last resolved listing, served until `expiresAt` or invalidation.
327+
value?: string[];
328+
expiresAt: number;
329+
};
330+
331+
// Keyed by the resolved sessions directory rather than the agent id: the dir
332+
// folds in OPENCLAW_STATE_DIR so it stays correct across env changes, and it
333+
// matches `path.dirname(update.sessionFile)` for precise invalidation.
334+
const sessionFilesListingByDir = new Map<string, SessionFilesListingEntry>();
335+
let sessionFilesInvalidationSubscribed = false;
336+
337+
function ensureSessionFilesInvalidationSubscription(): void {
338+
if (sessionFilesInvalidationSubscribed) {
339+
return;
340+
}
341+
sessionFilesInvalidationSubscribed = true;
342+
// Single canonical invalidation channel. Archive rotation also emits here, so
343+
// each rotation/mutation write path is covered without a separate hook.
344+
onSessionTranscriptUpdate((update) => {
345+
if (update.agentId) {
346+
sessionFilesListingByDir.delete(resolveSessionTranscriptsDirForAgent(update.agentId));
347+
}
348+
if (update.sessionFile) {
349+
sessionFilesListingByDir.delete(path.dirname(update.sessionFile));
350+
}
351+
});
352+
}
353+
354+
function pruneSessionFilesListing(): void {
355+
while (sessionFilesListingByDir.size > SESSION_FILES_LISTING_MAX_DIRS) {
356+
const oldest = sessionFilesListingByDir.keys().next().value;
357+
if (oldest === undefined) {
358+
return;
359+
}
360+
sessionFilesListingByDir.delete(oldest);
361+
}
362+
}
363+
364+
// A successful scan (including a legitimately absent dir) yields a listing we
365+
// can cache; a transient failure must not be cached as an empty set, or one
366+
// NFS blip would hide real session files from sync/export for the whole TTL.
367+
type SessionFilesReadResult = { ok: true; files: string[] } | { ok: false };
368+
369+
async function readSessionFilesForDir(dir: string): Promise<SessionFilesReadResult> {
305370
try {
306371
const entries = await fs.readdir(dir, { withFileTypes: true });
307-
return entries
372+
const files = entries
308373
.filter((entry) => entry.isFile())
309374
.map((entry) => entry.name)
310375
.filter((name) => isUsageCountedSessionTranscriptFileName(name))
311376
.map((name) => path.join(dir, name));
312-
} catch {
377+
return { ok: true, files };
378+
} catch (err) {
379+
// A missing sessions dir is the normal "no sessions yet" state, safe to
380+
// cache as empty (the first session write emits an invalidation event).
381+
// Any other failure is transient (EIO/ESTALE/timeout) and stays uncached.
382+
if ((err as NodeJS.ErrnoException)?.code === "ENOENT") {
383+
return { ok: true, files: [] };
384+
}
385+
return { ok: false };
386+
}
387+
}
388+
389+
async function loadSessionFilesSnapshot(
390+
dir: string,
391+
entry: SessionFilesListingEntry,
392+
): Promise<string[]> {
393+
const result = await readSessionFilesForDir(dir);
394+
// Only act if this read is still the active entry; an invalidation or a newer
395+
// read during the await must win.
396+
const active = sessionFilesListingByDir.get(dir) === entry;
397+
if (!result.ok) {
398+
// Drop the entry so the next caller retries; never publish a false-empty
399+
// snapshot from a transient failure.
400+
if (active) {
401+
sessionFilesListingByDir.delete(dir);
402+
}
313403
return [];
314404
}
405+
if (active) {
406+
entry.inFlight = undefined;
407+
entry.value = result.files;
408+
entry.expiresAt = Date.now() + SESSION_FILES_LISTING_TTL_MS;
409+
}
410+
return result.files;
411+
}
412+
413+
export async function listSessionFilesForAgent(agentId: string): Promise<string[]> {
414+
ensureSessionFilesInvalidationSubscription();
415+
const dir = resolveSessionTranscriptsDirForAgent(agentId);
416+
const existing = sessionFilesListingByDir.get(dir);
417+
// Return a copy: coalesced and cached callers share the stored snapshot, so
418+
// handing out the array directly would let one caller's in-place mutation
419+
// corrupt the cache and every other caller's view.
420+
if (existing?.inFlight) {
421+
return (await existing.inFlight).slice();
422+
}
423+
if (existing?.value && Date.now() < existing.expiresAt) {
424+
return existing.value.slice();
425+
}
426+
const entry: SessionFilesListingEntry = { expiresAt: 0 };
427+
sessionFilesListingByDir.set(dir, entry);
428+
pruneSessionFilesListing();
429+
// Publish inFlight before this function awaits so concurrent callers join a
430+
// single read; the snapshot is only stored on success.
431+
const inFlight = loadSessionFilesSnapshot(dir, entry);
432+
entry.inFlight = inFlight;
433+
return (await inFlight).slice();
434+
}
435+
436+
/** Clears cached session-file listing snapshots. Used by tests to isolate runs. */
437+
export function resetSessionFilesListingCache(): void {
438+
sessionFilesListingByDir.clear();
315439
}
316440

317441
function extractAgentIdFromSessionPath(absPath: string): string | null {

0 commit comments

Comments
 (0)