diff --git a/src/session-index/first-prompt.ts b/src/session-index/first-prompt.ts index 65f5e39c..fcf2bdfa 100644 --- a/src/session-index/first-prompt.ts +++ b/src/session-index/first-prompt.ts @@ -1,7 +1,8 @@ /** * The first prompt of a transcript, extracted the way the SDK does for a * session title (its `rf`, `ti` and `jP` in `@anthropic-ai/claude-agent-sdk`). - * Used when the SDK `getSessionInfo` read another copy of the session. + * Part of the SDK's title rule that the session index applies itself (see + * `sdkTitles`). * * - Only main-chain user records count: no tool result, meta or compact * summary record. diff --git a/src/session-index/list-subscriptions.ts b/src/session-index/list-subscriptions.ts index 34dfcbf9..7240daa1 100644 --- a/src/session-index/list-subscriptions.ts +++ b/src/session-index/list-subscriptions.ts @@ -20,6 +20,9 @@ * - A registry event names the record: only that record is read, and only * the sessions whose record changed get their `state` recomputed. * - A session this connection runs reports its SDK state at once. + * - A session whose transcripts this connection wrote (a rename, archive, + * unarchive or delete) is stat'ed and read again at once, without waiting + * for its transcript event. * * Events are coalesced: 150 ms after the last one, at most 1 s after the * first. Each subscription compares a recomputed row with the last row it @@ -345,6 +348,27 @@ export class ListSubscriptions { } } + /** + * This connection wrote the transcripts of a session: a rename, archive, + * unarchive or delete. Its transcripts are stat'ed and its row read again + * at once, as a transcript event would after its quiet period; the event + * that follows finds the row unchanged and sends nothing more. A copy the + * watch does not know yet is left to its event. + */ + sessionWritten(id: string): void { + for (const watch of this.watches.values()) { + if (!watch.ready) { + // Read when the watch is ready, as the events of its start. + watch.dirtyRows.add(id); + continue; + } + const files = watch.filesById.get(id); + if (!files) continue; + for (const filePath of files) watch.dirtyFiles.add(filePath); + this.schedule(watch, "now"); + } + } + dispose(): void { this.disposed = true; for (const id of [...this.subscriptions.keys()]) this.unsubscribe(id); diff --git a/src/session-index/service.ts b/src/session-index/service.ts index 5fff02c7..5f893942 100644 --- a/src/session-index/service.ts +++ b/src/session-index/service.ts @@ -16,10 +16,7 @@ import { type ListSessionsRequest, type ListSessionsResponse, } from "@agentclientprotocol/sdk"; -import { - deleteSession as sdkDeleteSession, - getSessionInfo as sdkGetSessionInfo, -} from "@anthropic-ai/claude-agent-sdk"; +import { deleteSession as sdkDeleteSession } from "@anthropic-ai/claude-agent-sdk"; import { airExtensionMeta } from "../air-extension.js"; import { sanitizeTitle } from "../session-titles.js"; import type { OwnSessionState } from "./activity.js"; @@ -34,6 +31,7 @@ import { ListSubscriptions, type ListChanges, type ListSubscribeResponse, + type ListSubscriptionDeps, } from "./list-subscriptions.js"; import { LiveSessionRegistry } from "./live-registry.js"; import { @@ -59,7 +57,6 @@ import { readSidecarTitle, SessionIndex, type ArchivedFilter, - type GetSessionInfo, type ListCursor, } from "./session-index.js"; import { sessionInfoOf } from "./session-info.js"; @@ -294,7 +291,7 @@ async function copyTitle(filePath: string, sessionId: string): Promise Promise; registry?: LiveSessionRegistry; now?: () => number; @@ -434,6 +430,11 @@ export type SessionIndexDeps = { ownSessionState?: (sessionId: string) => OwnSessionState | undefined; /** Sends `_session/list/changes`. */ notifyListChanges?: (changes: ListChanges) => Promise; + /** The timing of the list subscriptions, for tests. */ + listSubscriptionTiming?: Pick< + ListSubscriptionDeps, + "debounceMs" | "maxWaitMs" | "rescanMs" | "minSessionIntervalMs" + >; logError: (message: string, error: unknown) => void; }; @@ -449,7 +450,7 @@ export class SessionIndexService { private readonly deleteSession: (sessionId: string) => Promise; constructor(private readonly deps: SessionIndexDeps) { - this.index = new SessionIndex(deps.getSessionInfo ?? sdkGetSessionInfo); + this.index = new SessionIndex(); this.registry = deps.registry ?? new LiveSessionRegistry(); this.now = deps.now ?? Date.now; this.deleteSession = deps.deleteSession ?? ((id) => sdkDeleteSession(id)); @@ -588,6 +589,9 @@ export class SessionIndexService { stored = await this.titleCopies(sessionId, transcripts, change, options.sidecar); } return stored; + }).finally(() => { + // Whatever was written, also by a change that failed part way. + this.subscriptions?.sessionWritten(sessionId); }); } @@ -666,6 +670,7 @@ export class SessionIndexService { } } finally { this.index.invalidate(all); + this.subscriptions?.sessionWritten(sessionId); } }); } @@ -682,6 +687,7 @@ export class SessionIndexService { own: (sessionId) => this.deps.ownSessionState?.(sessionId), notify: (changes) => this.deps.notifyListChanges?.(changes) ?? Promise.resolve(), logError: this.deps.logError, + ...this.deps.listSubscriptionTiming, }); return this.subscriptions.subscribe(cwd); } diff --git a/src/session-index/session-index.ts b/src/session-index/session-index.ts index 5f892c38..83b55762 100644 --- a/src/session-index/session-index.ts +++ b/src/session-index/session-index.ts @@ -1,13 +1,14 @@ /** * The session list of a `sessionIndex` client. * - * Each list enumerates the transcripts of the requested cwd and of every + * A list enumerates the transcripts of the requested cwd and of every * worktree of its repository: one `readdir` per project directory and one - * `stat` per transcript, no long-lived cache. The metadata of a transcript is - * cached by `(path, mtime, size)` in an LRU and read on a miss with the SDK - * `getSessionInfo` (titles, sidecar) plus one head and tail read for - * what the SDK does not report (see {@link scanTranscript}). Misses are read in - * parallel batches. + * `stat` per transcript. The pages after a cursor reuse that enumeration for + * a short while (see {@link SessionIndex.reusableListing}). The metadata of a transcript is + * cached by `(path, mtime, size)` in an LRU and read on a miss from one head + * and tail read (the titles by the SDK's rule, see {@link sdkTitles}; the + * facts, see {@link scanTranscriptFile}) and the title sidecar. Misses are + * read in parallel batches. * * Order is the last user activity descending, then session id: the time of * the last real user prompt (`lastPromptAt`), else `updatedAt`. `updatedAt` @@ -20,7 +21,6 @@ import * as fs from "node:fs/promises"; import * as path from "node:path"; -import type { SDKSessionInfo } from "@anthropic-ai/claude-agent-sdk"; import { sanitizeTitle } from "../session-titles.js"; import { effectiveTitle, visibleTitle } from "./archive-title.js"; import { @@ -66,6 +66,10 @@ const READ_BATCH_SIZE = 16; const MAX_CWD_PROBES_PER_DIR = 64; /** How long a directory that gave no cwd is not probed again. */ const NO_CWD_RETRY_MS = 60_000; +/** How long the pages after a list reuse its enumeration. */ +const LISTING_REUSE_MS = 30_000; +/** Enumerations kept for the next page of their list. */ +const MAX_REUSABLE_LISTINGS = 8; /** Which sessions a list holds by archive state: `unarchived` the * unarchived ones only, `archived` the archived ones only, `all` both in one @@ -135,10 +139,60 @@ export type ListQuery = { after?: ListCursor; }; -export type GetSessionInfo = ( - sessionId: string, - options: { dir?: string }, -) => Promise; +/** The transcripts of a list scope as one enumeration found them. */ +type Listing = { + /** The scope paths; undefined for every project. */ + paths: string[] | undefined; + /** The directories read: the projects root, then each project directory. */ + dirs: string[]; + /** The names each of `dirs` had, as {@link DirListings} returned them. */ + names: string[][]; + /** Newest mtime first. */ + candidates: TranscriptCandidate[]; + /** When it was enumerated. */ + at: number; +}; + +/** The key of the page of `query` that follows `cursor`. */ +function pageKey(query: ListQuery, cursor: ListCursor): string { + return JSON.stringify([ + query.cwd ?? null, + query.includeWorktrees ?? false, + cursor.orderAtMs, + cursor.sessionId, + ]); +} + +/** The candidates of each session. */ +function copiesBySession( + candidates: readonly TranscriptCandidate[], +): Map { + const copies = new Map(); + for (const candidate of candidates) { + const known = copies.get(candidate.sessionId); + if (known) known.push(candidate); + else copies.set(candidate.sessionId, [candidate]); + } + return copies; +} + +/** `candidates` as they are on disk now; one that is gone or empty is left + * out. */ +async function restat(candidates: readonly TranscriptCandidate[]): Promise { + const stats = await statFiles(candidates.map(({ filePath }) => filePath)); + return candidates.flatMap((candidate, i) => { + const now = stats[i]; + if (!now?.isFile() || now.size === 0) return []; + if ( + now.mtimeMs === candidate.mtimeMs && + now.size === candidate.size && + now.ino === candidate.ino + ) { + return [candidate]; + } + return [{ ...candidate, mtimeMs: now.mtimeMs, size: now.size, ino: now.ino }]; + }); +} type CacheEntry = { mtimeMs: number; @@ -346,8 +400,16 @@ export class SessionIndex { /** Directories whose transcripts gave no cwd, by the directory mtime and * when that was learned. */ private readonly dirsWithoutCwd = new Map(); + /** The enumeration of each list with a next page, by {@link pageKey}. */ + private readonly nextPages = new Lru(MAX_REUSABLE_LISTINGS); - constructor(private readonly getSessionInfo: GetSessionInfo) {} + /** Transcripts read so far (metadata cache misses). */ + private reads = 0; + + /** How many transcripts the index has read, for tests and diagnostics. */ + get transcriptsRead(): number { + return this.reads; + } /** The paths whose sessions a list of `cwd` shows: `cwd`, and with * `includeWorktrees` the same subdirectory of every other existing @@ -418,14 +480,21 @@ export class SessionIndex { * every copy of a session: a list shows the newest one that has a row * (see {@link collectRows}), as the SDK does. */ async enumerateFiles(paths?: readonly string[]): Promise { + return (await this.enumerate(paths)).candidates; + } + + /** {@link enumerateFiles}, with the directory listings it was made from. */ + private async enumerate( + paths?: readonly string[], + ): Promise<{ dirs: string[]; names: string[][]; candidates: TranscriptCandidate[] }> { const root = projectsRoot(); + const rootNames = await this.listings.names(root); const dirs: { dirName: string; projectPath?: string }[] = paths ? await this.projectDirs(paths) - : (await this.listings.names(root)).map((dirName) => ({ dirName })); + : rootNames.map((dirName) => ({ dirName })); // All directories' names first, then one bounded run of stats. - const listed = await this.listings.namesOfAll( - dirs.map(({ dirName }) => path.join(root, dirName)), - ); + const dirPaths = dirs.map(({ dirName }) => path.join(root, dirName)); + const listed = await this.listings.namesOfAll(dirPaths); const files = dirs.flatMap(({ dirName, projectPath }, i) => listed[i]!.filter((name) => name.endsWith(".jsonl") && isSessionId(name.slice(0, -6))).map( (name) => ({ name, filePath: path.join(root, dirName, name), dirName, projectPath }), @@ -446,7 +515,7 @@ export class SessionIndex { ino: fileStats.ino, }); }); - return all; + return { dirs: [root, ...dirPaths], names: [rootNames, ...listed], candidates: all }; } /** @@ -524,11 +593,10 @@ export class SessionIndex { /** One page of rows, plus whether more rows follow the page. */ async list(query: ListQuery): Promise<{ rows: IndexRow[]; hasMore: boolean }> { - const paths = query.cwd - ? await this.listedPaths(query.cwd, query.includeWorktrees ?? false) - : undefined; const after = query.after; - const enumerated = (await this.enumerateFiles(paths)).sort(byMtimeDescending); + const reused = after ? await this.reusableListing(query, after) : undefined; + const listing = reused ?? (await this.freshListing(query)); + const { paths, candidates: enumerated } = listing; // A page after a cursor skips, without reading them, the sessions whose // newest transcript has a cached order key before the cursor. The // archive state comes from the title, so the archive filter applies once @@ -548,9 +616,47 @@ export class SessionIndex { archived: query.archived, after, wanted: query.limit + 1, - unread: (read) => enumerated.filter((candidate) => !read.has(candidate)), + unread: (read) => enumerated.filter((candidate) => !read.has(candidate.filePath)), + restat: reused !== undefined, }); - return { rows: rows.slice(0, query.limit), hasMore: rows.length > query.limit }; + const page = rows.slice(0, query.limit); + const hasMore = rows.length > query.limit; + const last = page.at(-1); + if (hasMore && last) this.nextPages.set(pageKey(query, last), listing); + return { rows: page, hasMore }; + } + + /** Enumerates the scope of `query`, newest mtime first. */ + private async freshListing(query: ListQuery): Promise { + const at = Date.now(); + const paths = query.cwd + ? await this.listedPaths(query.cwd, query.includeWorktrees ?? false) + : undefined; + const { dirs, names, candidates } = await this.enumerate(paths); + return { paths, dirs, names, candidates: candidates.sort(byMtimeDescending), at }; + } + + /** + * The enumeration of the earlier page of `query` that ended at `cursor`, + * while it is recent and no directory of it has changed its entries: a + * transcript was neither added nor removed (see {@link DirListings}, which + * returns the same names until then). Which transcripts the page reads is + * stat'ed again (see {@link collectRows}), so a page made from it is as + * fresh as one from a new enumeration: a transcript that grew since moves + * up, out of the pages after the cursor, or keeps its place. A page of + * thousands of transcripts then stats some hundreds, not all of them. + */ + private async reusableListing( + query: ListQuery, + cursor: ListCursor, + ): Promise { + const key = pageKey(query, cursor); + const listing = this.nextPages.peek(key); + if (!listing) return undefined; + this.nextPages.delete(key); + if (Date.now() - listing.at > LISTING_REUSE_MS) return undefined; + const names = await this.listings.namesOfAll(listing.dirs); + return names.every((current, i) => current === listing.names[i]) ? listing : undefined; } /** @@ -571,7 +677,7 @@ export class SessionIndex { inScope: scopeOf(paths), archived: "all", wanted: Infinity, - unread: (read) => siblings().filter((candidate) => !read.has(candidate)), + unread: (read) => siblings().filter((candidate) => !read.has(candidate.filePath)), }); } @@ -624,6 +730,12 @@ export class SessionIndex { * row, as the SDK list does. With a finite `wanted`, stops once `wanted` * rows are certain to come first (a candidate's order key is at most its * mtime, so `candidates` must be in mtime order then). + * + * With `restat`, `candidates` are an earlier enumeration: each is stat'ed + * again just before it is read, together with the other copies of its + * session, any of which may be its newest one now. Their mtimes still + * bound the order keys: a transcript that changed since either kept its + * order key (no new prompt) or got one newer than any listed before. */ private async collectRows( candidates: readonly TranscriptCandidate[], @@ -633,8 +745,10 @@ export class SessionIndex { archived: ArchivedFilter; after?: ListCursor; wanted: number; - /** The transcripts not read, which may supply a sibling's cwd. */ - unread: (read: ReadonlySet) => readonly TranscriptCandidate[]; + /** The transcripts not read (by path), which may supply a sibling's + * cwd. */ + unread: (read: ReadonlySet) => readonly TranscriptCandidate[]; + restat?: boolean; }, ): Promise { const { inScope, archived, after, wanted } = options; @@ -668,7 +782,9 @@ export class SessionIndex { } pending = left; }; - const read = new Set(); + const read = new Set(); + /** With `restat`: the copies of each session. */ + const copies = options.restat ? copiesBySession(candidates) : undefined; let index = 0; while (index < candidates.length) { if (bySession.size >= wanted) { @@ -678,11 +794,18 @@ export class SessionIndex { break; } } - const batch = candidates.slice(index, index + READ_BATCH_SIZE); - index += batch.length; + const next = candidates.slice(index, index + READ_BATCH_SIZE); + index += next.length; + const batch = copies + ? await restat( + [...new Set(next.flatMap(({ sessionId }) => copies.get(sessionId)!))].filter( + ({ filePath }) => !read.has(filePath), + ), + ) + : next; const resolved = await Promise.all( batch.map(async (candidate) => { - read.add(candidate); + read.add(candidate.filePath); const metadata = await this.metadataOf(candidate); if (!metadata || (await this.continuedElsewhere(candidate, metadata))) return undefined; return { candidate, metadata }; @@ -704,6 +827,7 @@ export class SessionIndex { await this.learnDirCwds( new Set(pending.map(({ candidate }) => candidate.dirName)), options.unread(read), + options.restat ?? false, ); await settlePending(); } @@ -733,10 +857,11 @@ export class SessionIndex { /** Reads the unread transcripts of `dirNames` until each directory has a * known cwd, at most {@link MAX_CWD_PROBES_PER_DIR} per directory. A * directory that gives none is not read again for a while, unless it - * changes. */ + * changes. `restat` as in {@link collectRows}. */ private async learnDirCwds( dirNames: ReadonlySet, unread: readonly TranscriptCandidate[], + restatFirst: boolean, ): Promise { const root = projectsRoot(); const now = Date.now(); @@ -761,8 +886,9 @@ export class SessionIndex { }); while (remaining.length > 0) { remaining = remaining.filter((candidate) => !this.dirCwds.has(candidate.dirName)); - const batch = remaining.slice(0, READ_BATCH_SIZE); - remaining = remaining.slice(batch.length); + const next = remaining.slice(0, READ_BATCH_SIZE); + remaining = remaining.slice(next.length); + const batch = restatFirst ? await restat(next) : next; await Promise.all(batch.map((candidate) => this.metadataOf(candidate))); } for (const [dirName, mtimeMs] of mtimes) { @@ -839,6 +965,7 @@ export class SessionIndex { candidate: TranscriptCandidate, previous?: PreviousScan, ): Promise { + this.reads++; const headTail = await readHeadTail(candidate.filePath, candidate.size); if (isSidechainTranscript(headTail.head)) return null; const facts = await scanTranscriptFile( @@ -857,7 +984,7 @@ export class SessionIndex { ...(facts.tailCwd ? pathAndAncestors(facts.tailCwd) : []), ]); if (fileCwd) this.dirCwds.set(candidate.dirName, fileCwd); - const sdk = await this.titlesOf(candidate, headTail, fileCwd); + const sdk = await this.titlesOf(candidate, headTail); // No title at all: the SDK does not list it either. if (!sdk.summary) return null; // Archived by its name alone: the agent name, else the custom title. @@ -882,32 +1009,17 @@ export class SessionIndex { } /** - * The titles the SDK reports for the listed transcript (`customTitle` and - * `summary` of `getSessionInfo`). The SDK reads the first copy of the - * session that its search finds; its answer is used only when that copy is - * the listed file (same size and mtime), else the same titles are taken - * from the listed file itself (see {@link sdkTitles}). + * The titles the SDK gives the listed transcript (`customTitle` and + * `summary` of its `getSessionInfo`), by the SDK's rule from the head and + * tail read here and the sidecar (see {@link sdkTitles}). Not through the + * SDK itself: its lookup of the transcript reads the whole project + * directory on every call, which made each page of a large project cost a + * listing of every transcript per row, and it may read another copy of the + * session than the listed one. */ - private async titlesOf( - candidate: TranscriptCandidate, - headTail: HeadTail, - fileCwd: string | undefined, - ): Promise { - const dir = [fileCwd, candidate.projectPath].find( - (cwd) => cwd !== undefined && isExactProjectDir(candidate.dirName, cwd), - ); - const info = await this.getSessionInfo(candidate.sessionId, dir ? { dir } : {}).catch( - () => undefined, - ); - if ( - info && - info.fileSize === candidate.size && - info.lastModified === Math.trunc(candidate.mtimeMs) - ) { - return { customTitle: info.customTitle, summary: info.summary }; - } + private async titlesOf(candidate: TranscriptCandidate, headTail: HeadTail): Promise { const sidecar = - tailCustomTitle(headTail.tail) === undefined + tailCustomTitle(headTail) === undefined ? await readSidecarTitle(candidate.filePath, candidate.sessionId) : undefined; return sdkTitles(headTail, sidecar); diff --git a/src/session-index/transcript-scan.ts b/src/session-index/transcript-scan.ts index 6feed7f5..fe6d795a 100644 --- a/src/session-index/transcript-scan.ts +++ b/src/session-index/transcript-scan.ts @@ -56,7 +56,15 @@ export type TranscriptFacts = { hasMessages: boolean; }; -export type HeadTail = { head: string; tail: string }; +export type HeadTail = { + head: string; + /** The last 64 KB from their first whole line on; the whole file when it + * fits in the head. */ + tail: string; + /** The last 64 KB as the SDK searches them, with the cut line before + * `tail`; absent when that is `tail`. */ + sdkTail?: string; +}; /** Facts that an earlier scan may have found before the tail window. */ const INHERITED_FACTS = ["model", "tailCwd", "lastTurnEndedAt"] as const; @@ -69,14 +77,18 @@ export type PreviousScan = { size: number } & Pick< /** The last `window` bytes of a file of `size` bytes, without the first, * cut line. The whole file when it fits. */ -async function readTail(handle: fs.FileHandle, size: number, window: number): Promise { +async function readTail( + handle: fs.FileHandle, + size: number, + window: number, +): Promise<{ raw: string; tail: string }> { const length = Math.min(window, size); const buffer = Buffer.allocUnsafe(length); const { bytesRead } = await handle.read(buffer, 0, length, size - length); const raw = buffer.toString("utf8", 0, bytesRead); - if (length === size) return raw; + if (length === size) return { raw, tail: raw }; const newline = raw.indexOf("\n"); - return newline >= 0 ? raw.slice(newline + 1) : ""; + return { raw, tail: newline >= 0 ? raw.slice(newline + 1) : "" }; } /** The first and the last 64 KB of the first `size` bytes of `filePath`: @@ -98,7 +110,8 @@ export async function readHeadTailOf(handle: fs.FileHandle, size: number): Promi const first = await handle.read(buffer, 0, Math.min(CHUNK_SIZE, size), 0); const head = buffer.toString("utf8", 0, first.bytesRead); if (size <= CHUNK_SIZE) return { head, tail: head }; - return { head, tail: await readTail(handle, size, CHUNK_SIZE) }; + const { raw, tail } = await readTail(handle, size, CHUNK_SIZE); + return raw === tail ? { head, tail } : { head, tail, sdkTail: raw }; } /** @@ -464,8 +477,8 @@ export function isSidechainTranscript(head: string): boolean { } /** The last `customTitle` of the tail, as the SDK reads it. */ -export function tailCustomTitle(tail: string): string | undefined { - return lastField(tail, "customTitle"); +export function tailCustomTitle({ tail, sdkTail }: HeadTail): string | undefined { + return lastField(sdkTail ?? tail, "customTitle"); } /** The last non-blank top-level `agentName` of the records of `text`, the @@ -500,11 +513,15 @@ export type SdkTitles = { /** * The titles the SDK gives a transcript, from its own head and tail, in the * SDK's order: custom title (tail, the sidecar, head), AI title, then - * {@link generatedTitle}. For a transcript that the SDK `getSessionInfo` - * does not read (another copy of the session comes first in its search). - * The sidecar title counts only when the tail has no custom title. + * {@link generatedTitle}: what the SDK `getSessionInfo` reports for this + * file, without its lookup of the file. The sidecar title counts only when + * the tail has no custom title. */ -export function sdkTitles({ head, tail }: HeadTail, sidecarTitle?: string): SdkTitles { +export function sdkTitles( + { head, tail: wholeLines, sdkTail }: HeadTail, + sidecarTitle?: string, +): SdkTitles { + const tail = sdkTail ?? wholeLines; const customTitle = lastField(tail, "customTitle") ?? sidecarTitle ?? diff --git a/src/tests/session-index-units.test.ts b/src/tests/session-index-units.test.ts index b04b308c..addd13e2 100644 --- a/src/tests/session-index-units.test.ts +++ b/src/tests/session-index-units.test.ts @@ -837,7 +837,7 @@ describe("session id lookup", () => { await fs.mkdir(path.join(dir, id), { recursive: true }); const file = path.join(dir, `${id}.jsonl`); await fs.writeFile(file, lines(user("hi", "2026-01-01T00:00:00.000Z")) + "\n"); - const index = new SessionIndex(async () => undefined); + const index = new SessionIndex(); expect(await index.findTranscripts(id)).toEqual([file]); expect(await index.scanSession(id)).toEqual({ transcripts: [{ filePath: file, size: expect.any(Number) }], diff --git a/src/tests/session-index.test.ts b/src/tests/session-index.test.ts index 610a743d..a255fa7d 100644 --- a/src/tests/session-index.test.ts +++ b/src/tests/session-index.test.ts @@ -21,6 +21,8 @@ import type { SessionInfo } from "@agentclientprotocol/sdk"; import { encodeProjectPath } from "../session-index/project-dirs.js"; import { archiveInsteadOfDelete, + archiveTo, + renameTo, SessionIndexService, writeCustomTitleSidecar, } from "../session-index/service.js"; @@ -168,6 +170,10 @@ async function indexAgent() { return created; } +/** The transcripts the session index of `agent` has read so far. */ +const transcriptsRead = (agent: ClaudeAcpAgent): number => + (agent as any).sessionIndex.service.index.transcriptsRead; + /** The `custom-title` and `agent-name` records that title a session. */ const titleRecords = (sessionId: string, title: string) => [ { type: "custom-title", customTitle: title, sessionId }, @@ -335,11 +341,11 @@ describe("session/list of a sessionIndex client", () => { const { agent } = await indexAgent(); const page = await agent.listSessions({ cwd: workspace, _meta: listMeta({ limit: 5 }) }); expect(page.sessions).toHaveLength(5); - expect(vi.mocked(getSessionInfo).mock.calls.length).toBeLessThanOrEqual(16); + const read = transcriptsRead(agent); + expect(read).toBeLessThanOrEqual(16); // A second list reads nothing again. - vi.mocked(getSessionInfo).mockClear(); await agent.listSessions({ cwd: workspace, _meta: listMeta({ limit: 5 }) }); - expect(getSessionInfo).not.toHaveBeenCalled(); + expect(transcriptsRead(agent)).toBe(read); }); it("lists unarchived sessions by default, archived ones only, or all in one order", async () => { @@ -1624,13 +1630,12 @@ describe("_session/list/subscribe", () => { // Past the events of these writes, which a new watch may still get. await new Promise((resolve) => setTimeout(resolve, 1000)); const { subscriptions, sent, rowsRead, service } = slowFirstRows(0, { rescanMs: 300 }); - vi.mocked(getSessionInfo).mockClear(); await subscriptions.subscribe(workspace); // Past the rescan after the watchers opened, and a few periodic ones. await new Promise((resolve) => setTimeout(resolve, 1500)); expect(rowsRead).toEqual([]); expect(sent).toEqual([]); - expect(vi.mocked(getSessionInfo)).not.toHaveBeenCalled(); + expect(service.index.transcriptsRead).toBe(0); const files = await service.index.enumerateFiles([workspace]); expect(files).toHaveLength(12); expect(files.filter((file) => service.index.isRead(file))).toEqual([]); @@ -2567,12 +2572,11 @@ describe("cwd recovery in a directory without any cwd", () => { await writeTranscript({ cwd: other, recordCwd: null, lastMessageAt: base - 3600_000 - i }); } const { agent } = await indexAgent(); - vi.mocked(getSessionInfo).mockClear(); await agent.listSessions({ _meta: listMeta({ limit: 2 }) }); - expect(vi.mocked(getSessionInfo).mock.calls.length).toBeLessThanOrEqual(16 + 64); - vi.mocked(getSessionInfo).mockClear(); + const read = transcriptsRead(agent); + expect(read).toBeLessThanOrEqual(16 + 64); await agent.listSessions({ _meta: listMeta({ limit: 2 }) }); - expect(getSessionInfo).not.toHaveBeenCalled(); + expect(transcriptsRead(agent)).toBe(read); }); }); @@ -2677,14 +2681,6 @@ describe("list metadata of the listed copy", () => { const page = await agent.listSessions({}); expect(page.sessions.map((s) => s.title)).toEqual(["Listed copy"]); }); - - it("keeps a row whose copy the SDK does not find", async () => { - const session = await writeTranscript({}); - vi.mocked(getSessionInfo).mockResolvedValueOnce(undefined); - const { agent } = await indexAgent(); - const page = await agent.listSessions({ cwd: workspace }); - expect(page.sessions.map((s) => [s.sessionId, s.title])).toEqual([[session.id, "Fix it"]]); - }); }); describe("a continued session", () => { @@ -2790,8 +2786,6 @@ describe("title of a listed copy that starts with a slash command", () => { .map((entry) => JSON.stringify(entry)) .join("\n") + "\n", ); - // The SDK does not find this copy: the title comes from the file. - vi.mocked(getSessionInfo).mockResolvedValueOnce(undefined); const { agent } = await indexAgent(); const page = await agent.listSessions({ cwd: workspace }); expect(page.sessions.map((s) => s.title)).toEqual([ @@ -3565,7 +3559,6 @@ describe("session index cost", () => { const base = Date.parse("2026-08-01T00:00:00Z"); for (let i = 0; i < 120; i++) await writeTranscript({ lastMessageAt: base - i * 60_000 }); const { agent } = await indexAgent(); - vi.mocked(getSessionInfo).mockClear(); let cursor: string | undefined; let rows = 0; do { @@ -3579,7 +3572,7 @@ describe("session index cost", () => { } while (cursor); expect(rows).toBe(120); // Pages after a cursor skip the transcripts cached before it. - expect(vi.mocked(getSessionInfo).mock.calls.length).toBeLessThanOrEqual(120); + expect(transcriptsRead(agent)).toBeLessThanOrEqual(120); }); it("reads only the changed transcript for a subscription", async () => { @@ -3594,12 +3587,11 @@ describe("session index cost", () => { }); await service.subscribeList(workspace); await settle(); - vi.mocked(getSessionInfo).mockClear(); + const read = service.index.transcriptsRead; await fs.appendFile(sessions[3]!.file, promptRecord(sessions[3]!.id, "Next") + "\n"); expect(await waitFor(() => changes.length > 0)).toBe(true); - // At most the changed one (it may have been read already, for its - // creation just before subscribe). - for (const [id] of vi.mocked(getSessionInfo).mock.calls) expect(id).toBe(sessions[3]!.id); + // At most the changed one. + expect(service.index.transcriptsRead - read).toBeLessThanOrEqual(1); service.dispose(); }); }); @@ -4103,3 +4095,252 @@ describe("archive of a long session", () => { expect(lastPromptAt(await list("archived"))).toBe(new Date(at - 1000).toISOString()); }); }); + +describe("list subscriptions after this connection's own title changes", () => { + it("push a rename, archive, unarchive and delete at once, without the watcher", async () => { + const session = await writeTranscript({ prompt: "Old title" }); + const changes: ListChanges[] = []; + const service = new SessionIndexService({ + notifyListChanges: async (change) => { + changes.push(change); + }, + // The watcher's events and rescans would come only after the test. + listSubscriptionTiming: { + debounceMs: 60_000, + maxWaitMs: 60_000, + rescanMs: 60_000, + minSessionIntervalMs: 0, + }, + logError: () => {}, + }); + await service.subscribeList(workspace); + // Past the rescan that follows the opening of the watchers. + await new Promise((resolve) => setTimeout(resolve, 1300)); + changes.length = 0; + const last = () => + changes.flatMap((change) => change.sessions.filter((s) => s.sessionId === session.id)).at(-1); + const soon = (check: () => boolean) => waitFor(check, 500); + + await service.retitle(session.id, renameTo("New title"), { sidecar: "always" }); + expect(await soon(() => last()?.title === "New title")).toBe(true); + await service.retitle(session.id, archiveTo(true), { sidecar: "existing" }); + expect(await soon(() => airRow(last()!).archived === true)).toBe(true); + await service.retitle(session.id, archiveTo(false), { sidecar: "existing" }); + expect(await soon(() => airRow(last()!).archived === false)).toBe(true); + expect(last()!.title).toBe("New title"); + // A rename to the title it has changes nothing, and sends nothing. + const sent = changes.length; + await service.retitle(session.id, renameTo("New title"), { sidecar: "always" }); + await new Promise((resolve) => setTimeout(resolve, 300)); + expect(changes).toHaveLength(sent); + + await service.delete(session.id, false); + expect(await soon(() => changes.some((change) => change.removed.includes(session.id)))).toBe( + true, + ); + service.dispose(); + }); + + it("push a rename made by another process through the watcher", async () => { + const session = await writeTranscript({ prompt: "Old title" }); + const changes: ListChanges[] = []; + const service = new SessionIndexService({ + notifyListChanges: async (change) => { + changes.push(change); + }, + logError: () => {}, + }); + await service.subscribeList(workspace); + await settle(); + await fs.appendFile( + session.file, + titleRecords(session.id, "Renamed elsewhere") + .map((record) => JSON.stringify(record)) + .join("\n") + "\n", + ); + await writeCustomTitleSidecar(session.file, "Renamed elsewhere"); + expect( + await waitFor(() => + changes.some((change) => + change.sessions.some( + (s) => s.sessionId === session.id && s.title === "Renamed elsewhere", + ), + ), + ), + ).toBe(true); + service.dispose(); + }); +}); + +describe("the pages of one list", () => { + const base = Date.parse("2026-07-01T00:00:00Z"); + const projectDir = () => path.join(configDir, "projects", encodeProjectPath(workspace)); + + /** Transcripts one minute apart, newest first, in directories that + * changed long enough ago for their listings to be kept. */ + async function project(count: number) { + const sessions: { id: string; file: string }[] = []; + for (let i = 0; i < count; i++) { + sessions.push(await writeTranscript({ lastMessageAt: base - i * 60_000 })); + } + const past = (Date.now() - 60_000) / 1000; + await fs.utimes(projectDir(), past, past); + await fs.utimes(path.dirname(projectDir()), past, past); + return sessions; + } + + async function drain( + agent: ClaudeAcpAgent, + limit: number, + between?: (page: number) => Promise, + ) { + const pages: string[][] = []; + let cursor: string | undefined; + do { + const page = await agent.listSessions({ cwd: workspace, cursor, _meta: listMeta({ limit }) }); + pages.push(page.sessions.map((s) => s.sessionId)); + cursor = page.nextCursor ?? undefined; + if (cursor) await between?.(pages.length); + } while (cursor); + return pages; + } + + const enumerations = (agent: ClaudeAcpAgent) => + vi.spyOn((agent as any).sessionIndex.service.index, "enumerate"); + + it("enumerates the transcripts once for all its pages", async () => { + const sessions = await project(45); + const { agent } = await indexAgent(); + const enumerate = enumerations(agent); + const pages = await drain(agent, 10); + expect(pages.flat()).toEqual(sessions.map(({ id }) => id)); + expect(enumerate).toHaveBeenCalledTimes(1); + // A new list enumerates again, and pages as before. + expect(await drain(agent, 10)).toEqual(pages); + expect(enumerate).toHaveBeenCalledTimes(2); + }); + + it("is as fresh after a cursor as a new enumeration", async () => { + const sessions = await project(30); + const { agent } = await indexAgent(); + const enumerate = enumerations(agent); + const deleted = sessions[25]!; + const prompted = sessions[22]!; + const answered = sessions[28]!; + const pages = await drain(agent, 10, async (page) => { + if (page !== 1) return; + // Neither adds nor removes a transcript: the enumeration is reused. + await fs.truncate(deleted.file, 0); + await fs.appendFile(prompted.file, promptRecord(prompted.id, "Again") + "\n"); + await fs.appendFile(answered.file, assistantRecord(answered.id, "end_turn") + "\n"); + }); + expect(enumerate).toHaveBeenCalledTimes(1); + const listed = pages.flat(); + // Emptied: gone. Prompted: moved before the cursor, out of later pages. + expect(listed).not.toContain(deleted.id); + expect(listed).not.toContain(prompted.id); + // Answered without a prompt: in its place, with its new turn end. + expect(listed.indexOf(answered.id)).toBe(listed.indexOf(sessions[27]!.id) + 1); + const fresh = await agent.listSessions({ cwd: workspace, _meta: listMeta({ limit: 50 }) }); + expect(fresh.sessions[0]!.sessionId).toBe(prompted.id); + expect(fresh.sessions.map((s) => s.sessionId).filter((id) => id !== prompted.id)).toEqual( + listed, + ); + }); + + it("reads every copy of a session that a later page reaches", async () => { + const sessions = await project(30); + // A session with a copy among the second page and an older one in + // another project, which gets a new prompt after the first page. + const id = randomUUID(); + await writeTranscript({ sessionId: id, lastMessageAt: base - 14.5 * 60_000 }); + const older = await writeTranscript({ + sessionId: id, + cwd: path.join(workspace, "other"), + lastMessageAt: base - 90 * 60_000, + }); + const past = (Date.now() - 60_000) / 1000; + await fs.utimes(projectDir(), past, past); + await fs.utimes(path.dirname(older.file), past, past); + await fs.utimes(path.dirname(projectDir()), past, past); + const { agent } = await indexAgent(); + const enumerate = enumerations(agent); + const pages: string[][] = []; + let cursor: string | undefined; + do { + const page = await agent.listSessions({ cursor, _meta: listMeta({ limit: 10 }) }); + pages.push(page.sessions.map((s) => s.sessionId)); + cursor = page.nextCursor ?? undefined; + if (pages.length === 1) { + await fs.appendFile(older.file, promptRecord(id, "Resumed") + "\n"); + } + } while (cursor); + expect(enumerate).toHaveBeenCalledTimes(1); + // Its newest copy now comes before the first page's cursor. + expect(pages.flat()).toEqual(sessions.map((session) => session.id)); + const fresh = await agent.listSessions({ _meta: listMeta({ limit: 1 }) }); + expect(fresh.sessions.map((s) => [s.sessionId, s.cwd])).toEqual([ + [id, path.join(workspace, "other")], + ]); + }); + + it("enumerates again once a transcript is added or removed", async () => { + const sessions = await project(30); + const { agent } = await indexAgent(); + const enumerate = enumerations(agent); + let added: { id: string } | undefined; + const pages = await drain(agent, 10, async (page) => { + if (page === 1) { + // An older session copied in: listed in its place. + added = await writeTranscript({ lastMessageAt: base - 25.5 * 60_000 }); + } else if (page === 2) { + await fs.rm(sessions[29]!.file); + } + }); + expect(enumerate).toHaveBeenCalledTimes(3); + const listed = pages.flat(); + expect(listed.indexOf(added!.id)).toBe(listed.indexOf(sessions[25]!.id) + 1); + expect(listed).not.toContain(sessions[29]!.id); + expect(listed).toHaveLength(30); + }); +}); + +describe("titles by the SDK's rule", () => { + it("finds a custom title whose record starts before the tail window", async () => { + const id = randomUUID(); + // Past the head window. + let filler = assistantRecord(id, null); + while (filler.length < 70_000) filler += "\n" + assistantRecord(id, null); + // The cut line of the last 64 KB, which the SDK still searches. + const title = JSON.stringify({ + type: "custom-title", + padding: "p".repeat(10_000), + customTitle: "Across the boundary", + sessionId: id, + }); + // About 60 KB after it: the window starts within its padding. + let tail = assistantRecord(id, null); + while (tail.length < 60_000) tail += "\n" + assistantRecord(id, null); + const session = await writeTranscript({ + sessionId: id, + prompt: "First prompt", + trailer: [], + }); + await fs.appendFile(session.file, `${filler}\n${title}\n${tail}\n`); + const bytes = await fs.readFile(session.file); + const window = bytes.length - 64 * 1024; + const start = bytes.indexOf(title); + expect(start).toBeLessThan(window); + expect(start + title.indexOf('"customTitle"')).toBeGreaterThan(window); + expect(start).toBeGreaterThan(64 * 1024); + const sdk = await vi + .importActual( + "@anthropic-ai/claude-agent-sdk", + ) + .then((module) => module.getSessionInfo(id, { dir: workspace })); + expect(sdk!.summary).toBe("Across the boundary"); + const { agent } = await indexAgent(); + const page = await agent.listSessions({ cwd: workspace }); + expect(page.sessions.map((s) => s.title)).toEqual(["Across the boundary"]); + }); +});