jerry/claude-agent-acp
1
0
Fork 0
mirror of https://github.com/agentclientprotocol/claude-agent-acp.git synced 2026-10-27 12:40:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: push renamed sessions to list subscribers and reuse one listing across pages (#1293)

Browse Source 
Two findings of the 2026-10-10 night perf run on a 5,000-session home
(findings 8 and 9).

## Rename push (finding 9)

The run measured "no push within 10 s" after `_session/rename`. The
cause was the harness, not the adapter. The bench renames the newest
session to `"Night renamed"`, and an earlier bench run on the same home
had already given it that title. The transcript shows that pair of title
records twice. A rename to the current title appends nothing, so there
was nothing to push. On a fresh home, the same bench gets the push. The
hypotheses checked:

- The metadata cache is not updated by a rename. The appended records
change the transcript's size, and the watcher re-reads it.
- The sidecar lies outside the non-recursive watch, but so does every
rename. Each one also appends to the transcript, which is watched.

Still, the push depended on `fs.watch` plus its 150 ms quiet period. Now
a rename, archive, unarchive or delete through this connection publishes
at once. `ListSubscriptions.sessionWritten` stats the session's
transcripts and reads its row, the way a transcript event would, without
waiting. The event that follows finds the row unchanged and sends
nothing. Renames by other processes still arrive through the watcher.
The 1 s per-session interval of a subscription still applies.

## Listing every page (finding 8)

Profiling `SessionIndex.list` showed where a page's time goes. The full
enumeration and stat of 5,000 transcripts takes about 20 ms per page.
Most of the remaining ~380 ms was the SDK `getSessionInfo`, called once
per row read: its lookup runs a `readdir` of the whole project directory
(5,000 entries) on every call.

- A metadata read no longer calls the SDK. Titles come from the head and
tail the index already reads, by the same rule (`sdkTitles`). The tail
is searched as the SDK searches it, cut first line included.
- The pages after a cursor reuse the list's enumeration. It is keyed by
the cursor and kept for at most 30 s, at most 8 lists. It is dropped as
soon as a directory of the scope adds or removes an entry, through the
`DirListings` identity check. Such a page stats only the transcripts it
reads (each with its session's other copies) right before reading them.
A transcript that changed since either keeps its order key (no new
prompt) or moves up before the cursor, so the page is as fresh as one
from a new enumeration. The first page always enumerates.

## Measurements

Real adapter (`dist/index.js`), driven as AIR drives it by
`acp-perf-harness/air-ui/night/acp-adapter-bench.mjs`. Fixture: 5,000
sessions from `gen-night-data.mjs`.

| | before | after |
|---|---|---|
| Listing every page (25 × 200) | 10.1–16.3 s | 0.92–1.16 s |
| First page (51) | 156–472 ms | 46–62 ms |
| Show more (50) | 117–283 ms | 13–23 ms |
| Push after archive | ~170 ms | 3–4 ms |
| Push after a rename | ~170 ms | 3–5 ms |

## Tests

- A rename, archive, unarchive and delete each push within 500 ms while
the watcher's quiet period is 60 s (fails without the fix). A rename by
another process is pushed through the watcher.
- Paging:
  - one enumeration for all pages;
- between pages, a transcript emptied, one with a new prompt, and one
with only a new answer give the same pages as a new enumeration;
- an older copy of a session that got a prompt hides it on later pages;
  - adding or removing a transcript enumerates again.
- A custom title whose record starts before the tail window matches the
SDK.
- Read-count assertions use the new `SessionIndex.transcriptsRead`
counter instead of SDK call counts.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
nikita-ashihmin
2026-10-10 13:06:40 +04:00
committed by GitHub
parent 96b6b4f478
commit 92abb8fd4d
7 changed files with 505 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/session-index/first-prompt.ts
View File

@@ -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.

24
src/session-index/list-subscriptions.ts
View File

@@ -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);

22
src/session-index/service.ts
View File

@@ -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<string |
const { size } = await fs.stat(filePath);
const headTail = await readHeadTail(filePath, size);
const sidecar =
tailCustomTitle(headTail.tail) === undefined
tailCustomTitle(headTail) === undefined
? await readSidecarTitle(filePath, sessionId)
: undefined;
const { title, archived } = effectiveTitle(
@@ -426,7 +423,6 @@ export type RetitleOptions = {
};
export type SessionIndexDeps = {
getSessionInfo?: GetSessionInfo;
deleteSession?: (sessionId: string) => Promise<void>;
registry?: LiveSessionRegistry;
now?: () => number;
@@ -434,6 +430,11 @@ export type SessionIndexDeps = {
ownSessionState?: (sessionId: string) => OwnSessionState | undefined;
/** Sends `_session/list/changes`. */
notifyListChanges?: (changes: ListChanges) => Promise<void>;
/** 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<void>;
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);
}

228
src/session-index/session-index.ts
View File

@@ -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<SDKSessionInfo | undefined>;
/** 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<string, TranscriptCandidate[]> {
const copies = new Map<string, TranscriptCandidate[]>();
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<TranscriptCandidate[]> {
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<string, { mtimeMs: number; at: number }>();
/** The enumeration of each list with a next page, by {@link pageKey}. */
private readonly nextPages = new Lru<string, Listing>(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<TranscriptCandidate[]> {
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<Listing> {
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<Listing | undefined> {
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<TranscriptCandidate>) => readonly TranscriptCandidate[];
/** The transcripts not read (by path), which may supply a sibling's
* cwd. */
unread: (read: ReadonlySet<string>) => readonly TranscriptCandidate[];
restat?: boolean;
},
): Promise<IndexRow[]> {
const { inScope, archived, after, wanted } = options;
@@ -668,7 +782,9 @@ export class SessionIndex {
}
pending = left;
};
const read = new Set<TranscriptCandidate>();
const read = new Set<string>();
/** 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<string>,
unread: readonly TranscriptCandidate[],
restatFirst: boolean,
): Promise<void> {
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<TranscriptMetadata | null> {
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<SdkTitles> {
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<SdkTitles> {
const sidecar =
tailCustomTitle(headTail.tail) === undefined
tailCustomTitle(headTail) === undefined
? await readSidecarTitle(candidate.filePath, candidate.sessionId)
: undefined;
return sdkTitles(headTail, sidecar);

39
src/session-index/transcript-scan.ts
View File

@@ -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<string> {
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 ??

2
src/tests/session-index-units.test.ts
View File

@@ -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) }],

291
src/tests/session-index.test.ts
View File

@@ -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<void>,
) {
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<typeof import("@anthropic-ai/claude-agent-sdk")>(
"@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"]);
});
});