sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(agent): simplify session storage composition

Browse source
This commit is contained in:
Christian Klotz 2026-08-02 00:08:18 +03:00
parent 1a7e1faa1f
commit 8823891f3d
No known key found for this signature in database
16 changed files with 538 additions and 498 deletions
Download patch file Download diff file Expand all files Collapse all files

96
packages/agent/docs/harness.md
View file

@ -1996,8 +1996,9 @@ Rules, both scopes:
## 14. Session stores
`Session` is the only opened-session object. `SessionRepository` borrows a
caller-owned `SessionStore`, opens readers, and constructs `Session` objects.
The in-memory store is the simplest built-in option:
caller-owned, faceted `SessionStore` and constructs `Session` objects from
canonical metadata. No separate opened reader exists. The in-memory store is
the simplest built-in option:
```ts
await using store = createInMemorySessionStore();
@ -2007,71 +2008,62 @@ const session = await repository.create({});
```
Use `createJsonlSessionStore({ fs, sessionsRoot })` for filesystem persistence;
the same scanning search composes with it. Omitting `search` is valid, but
`repository.search()` then returns no hits. SQLite support is provided
separately by `createSqliteSessionStore()` and should use
`createSqliteSessionSearch(options)` against the same canonical database. Drain
all session or harness operations before the store leaves
scope; repositories and sessions do not dispose it.
the same scanning search composes with it. Omitting `search` is valid. SQLite
support is provided separately by `createSqliteSessionStore()` and may compose
with `createSqliteSessionSearch(options)` against the same canonical database.
A custom store implements `SessionStore`:
A custom store implements one lifecycle facet and one entry facet:
```ts
interface SessionReader<TMetadata extends SessionMetadata = SessionMetadata> {
readonly metadata: TMetadata;
readHead(): Promise<SessionHead>;
readEntry(id: string): Promise<SessionTreeEntry | undefined>;
readEntries(options?: SessionEntryCursorOptions): Promise<readonly SessionTreeEntry[]>;
readPathToRootOrCompaction(
leafId: string | null,
): Promise<readonly SessionTreeEntry[]>;
interface SessionCatalog<TMetadata, TCreateOptions, TListOptions> {
create(options: TCreateOptions): Promise<TMetadata>;
open(metadata: TMetadata): Promise<TMetadata>;
list(options?: TListOptions): Promise<TMetadata[]>;
delete(metadata: TMetadata): Promise<void>;
fork(source: TMetadata, options: TCreateOptions, selection: SessionForkSelection): Promise<TMetadata>;
}
type SessionForkSelection =
| { kind: "all" }
| { kind: "before_user_message"; entryId: string }
| { kind: "through_entry"; entryId: string };
interface SessionEntries<TMetadata> {
readHead(metadata: TMetadata): Promise<SessionHead>;
readEntry(metadata: TMetadata, id: string): Promise<SessionTreeEntry | undefined>;
readEntries(metadata: TMetadata, options?: SessionEntryCursorOptions): Promise<readonly SessionTreeEntry[]>;
findEntriesOnBranch(metadata: TMetadata, query: SessionBranchQuery & { start: string | null }): Promise<readonly SessionTreeEntry[]>;
readPathToRootOrCompaction(metadata: TMetadata, leafId: string | null): Promise<readonly SessionTreeEntry[]>;
append(metadata: TMetadata, entry: SessionTreeEntry): Promise<void>;
getLabel(metadata: TMetadata, id: string): Promise<string | undefined>;
getName(metadata: TMetadata): Promise<string | undefined>;
getStats(metadata: TMetadata): Promise<SessionStats>;
}
interface SessionStore<
TMetadata extends SessionMetadata = SessionMetadata,
TCreateOptions extends SessionCreateOptions = SessionCreateOptions,
TListOptions = void,
> extends AsyncDisposable {
create(options: TCreateOptions): Promise<SessionReader<TMetadata>>;
load(metadata: TMetadata): Promise<SessionReader<TMetadata>>;
list(options?: TListOptions): Promise<TMetadata[]>;
appendEntry(metadata: TMetadata, entry: SessionTreeEntry): Promise<void>;
delete(metadata: TMetadata): Promise<void>;
fork(
source: TMetadata,
options: TCreateOptions,
selection: SessionForkSelection,
): Promise<SessionReader<TMetadata>>;
interface SessionStore<TMetadata, TCreateOptions, TListOptions> extends AsyncDisposable {
readonly sessions: SessionCatalog<TMetadata, TCreateOptions, TListOptions>;
readonly entries: SessionEntries<TMetadata>;
}
```
`create()` and `load()` return a `SessionReader` with canonical metadata, head,
entry, range, and branch-path reads. Reader lifetimes belong to the store.
`appendEntry()` receives a complete entry and must preserve append order and
entry-id uniqueness. `SessionRepository` converts public fork options into a
`SessionForkSelection`; `fork()` validates and executes that selection inside
the owning store, allowing backend-native copies without repository-level
history materialization.
Disposal rejects new operations and reader calls, drains accepted writes, and
releases owned resources. Store implementations know nothing about harness
operations, queues, or recovery.
`Session.buildContext()` reads only the active path. Complete history is read on
demand by `getEntries()` and by derived queries that currently require full
hydration, such as session statistics, labels, and names.
Expose custom implementations through a factory returning `SessionStore` rather
than exporting the concrete class.
Lifecycle methods return canonical metadata. Entry methods receive that
metadata, allowing each backend to route to cached state or a native connection
without another public handle. Fork selection remains in the owning store,
allowing backend-native copies. Derived reads let Memory and JSONL maintain
in-memory projections while SQLite uses materialized rows and indexes.
`createSessionRepository({ onCommit })` optionally awaits an API-user callback
after canonical creates, forks, deletes, and appends. The callback decides
whether secondary-system failures propagate or are handled as best effort; a
rejection cannot roll back the already committed primary write.
Drain all session or harness operations before the store leaves scope.
Repositories and sessions do not dispose it. Store implementations know nothing
about harness operations, queues, search backends, or recovery. Expose custom
implementations through a factory returning `SessionStore` rather than exporting
the concrete class.
Contract, all stores:
- One total append order (`seq`) across session and harness entries. Harness entries and leaf records carry `ref`; session entries do not (membership derives from parent linkage).
- An append is durable when its promise resolves; events fire after.
- Entry ids are unique per session, enforced at append.
- Reader methods return immutable entry arrays; callers cannot mutate stored state.
- Entry methods return immutable entry arrays; callers cannot mutate stored state.
- `readHead()` rejects `invalid_session` when its non-null leaf does not reference a canonical entry.
- One writer per *session*, enforced by the serving layer; SQLite additionally rejects concurrent writers itself. This is per session, not per store: one SQLite database can host many sessions, all writable concurrently — each through its own single live harness. The same applies to a directory of JSONL files.

12
packages/agent/src/harness/session/fork.ts
View file

@ -1,6 +1,14 @@
import type { SessionForkOptions, SessionForkSelection, SessionReader, SessionTreeEntry } from "../types.ts";
import type { SessionForkOptions, SessionForkSelection, SessionTreeEntry } from "../types.ts";
import { SessionError } from "../types.ts";
type MaybePromise<T> = T | Promise<T>;
export interface SessionForkEntrySource {
readEntry(id: string): MaybePromise<SessionTreeEntry | undefined>;
readEntries(): MaybePromise<readonly SessionTreeEntry[]>;
readPathToRootOrCompaction(leafId: string | null): MaybePromise<readonly SessionTreeEntry[]>;
}
export function createSessionForkSelection(options: SessionForkOptions): SessionForkSelection {
if (!options.entryId) return { kind: "all" };
return (options.position ?? "before") === "at"
@ -9,7 +17,7 @@ export function createSessionForkSelection(options: SessionForkOptions): Session
}
export async function readSessionEntriesForFork(
reader: SessionReader,
reader: SessionForkEntrySource,
selection: SessionForkSelection,
): Promise<readonly SessionTreeEntry[]> {
if (selection.kind === "all") return reader.readEntries();

88
packages/agent/src/harness/session/jsonl-store.ts
View file

@ -3,8 +3,9 @@ import type {
JsonlSessionCreateOptions,
JsonlSessionListOptions,
JsonlSessionMetadata,
SessionCatalog,
SessionEntries,
SessionForkSelection,
SessionReader,
SessionStore,
SessionTreeEntry,
} from "../types.ts";
@ -12,7 +13,7 @@ import { SessionError, toError } from "../types.ts";
import { readSessionEntriesForFork } from "./fork.ts";
import { KeyedOperationQueue } from "./keyed-operation-queue.ts";
import { createSessionId, createTimestamp, getFileSystemResultOrThrow } from "./repository.ts";
import { createSessionEntryCollectionReader, SessionEntryCollection } from "./session-entry-collection.ts";
import { SessionEntryCollection } from "./session-entry-collection.ts";
export interface JsonlSessionStoreOptions {
fs: JsonlSessionStoreFileSystem;
@ -211,6 +212,28 @@ class JsonlSessionStore
private disposed = false;
private disposePromise: Promise<void> | undefined;
readonly sessions: SessionCatalog<JsonlSessionMetadata, JsonlSessionCreateOptions, JsonlSessionListOptions> = {
create: (options) => this.create(options),
open: (metadata) => this.open(metadata),
list: (options) => this.list(options),
delete: (metadata) => this.delete(metadata),
fork: (source, options, selection) => this.fork(source, options, selection),
};
readonly entries: SessionEntries<JsonlSessionMetadata> = {
readHead: (metadata) => this.readCollection(metadata, (entries) => entries.readHead()),
readEntry: (metadata, id) => this.readCollection(metadata, (entries) => entries.readEntry(id)),
readEntries: (metadata, options) => this.readCollection(metadata, (entries) => entries.readEntries(options)),
findEntriesOnBranch: (metadata, query) =>
this.readCollection(metadata, (entries) => entries.findEntriesOnBranch(query)),
readPathToRootOrCompaction: (metadata, leafId) =>
this.readCollection(metadata, (entries) => entries.readPathToRootOrCompaction(leafId)),
append: (metadata, entry) => this.appendEntry(metadata, entry),
getLabel: (metadata, id) => this.readCollection(metadata, (entries) => entries.getLabel(id)),
getName: (metadata) => this.readCollection(metadata, (entries) => entries.getName()),
getStats: (metadata) => this.readCollection(metadata, (entries) => entries.getStats()),
};
constructor(options: JsonlSessionStoreOptions) {
this.fs = options.fs;
this.sessionsRootInput = options.sessionsRoot;
@ -219,7 +242,7 @@ class JsonlSessionStore
});
}
create(options: JsonlSessionCreateOptions): Promise<SessionReader<JsonlSessionMetadata>> {
private create(options: JsonlSessionCreateOptions): Promise<JsonlSessionMetadata> {
this.assertOpen();
const descriptor = createDocumentDescriptor(options);
return this.operations.enqueue(descriptor.operationKey, () =>
@ -227,12 +250,12 @@ class JsonlSessionStore
);
}
load(metadata: JsonlSessionMetadata): Promise<SessionReader<JsonlSessionMetadata>> {
private open(metadata: JsonlSessionMetadata): Promise<JsonlSessionMetadata> {
this.assertOpen();
return this.operations.enqueue(this.operationKey(metadata), () => this.loadDocument(metadata));
}
private async loadDocument(metadata: JsonlSessionMetadata): Promise<SessionReader<JsonlSessionMetadata>> {
private async loadDocument(metadata: JsonlSessionMetadata): Promise<JsonlSessionMetadata> {
if (
!getFileSystemResultOrThrow(await this.fs.exists(metadata.path), `Failed to check session ${metadata.path}`)
) {
@ -242,10 +265,10 @@ class JsonlSessionStore
const entries = this.entryCollectionsByPath.get(metadata.path);
if (entries) entries.replace(document.entries);
else this.entryCollectionsByPath.set(metadata.path, new SessionEntryCollection(document.entries));
return this.reader(document.metadata);
return document.metadata;
}
list(options: JsonlSessionListOptions = {}): Promise<JsonlSessionMetadata[]> {
private list(options: JsonlSessionListOptions = {}): Promise<JsonlSessionMetadata[]> {
this.assertOpen();
return this.operations.enqueueBarrier(() => this.listSessions(options));
}
@ -265,7 +288,7 @@ class JsonlSessionStore
return sessions.sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime());
}
appendEntry(metadata: JsonlSessionMetadata, entry: SessionTreeEntry): Promise<void> {
private appendEntry(metadata: JsonlSessionMetadata, entry: SessionTreeEntry): Promise<void> {
this.assertOpen();
return this.operations.enqueue(this.operationKey(metadata), async () => {
if (
@ -287,7 +310,7 @@ class JsonlSessionStore
});
}
delete(metadata: JsonlSessionMetadata): Promise<void> {
private delete(metadata: JsonlSessionMetadata): Promise<void> {
this.assertOpen();
return this.operations.enqueue(this.operationKey(metadata), async () => {
getFileSystemResultOrThrow(
@ -299,11 +322,11 @@ class JsonlSessionStore
});
}
fork(
private fork(
source: JsonlSessionMetadata,
options: JsonlSessionCreateOptions,
selection: SessionForkSelection,
): Promise<SessionReader<JsonlSessionMetadata>> {
): Promise<JsonlSessionMetadata> {
this.assertOpen();
const descriptor = createDocumentDescriptor(options);
const sourceEntries = this.operations.enqueue(this.operationKey(source), async () => {
@ -314,10 +337,7 @@ class JsonlSessionStore
const entries = this.entryCollectionsByPath.get(source.path);
if (entries) entries.replace(document.entries);
else this.entryCollectionsByPath.set(source.path, new SessionEntryCollection(document.entries));
return readSessionEntriesForFork(
createSessionEntryCollectionReader(document.metadata, this.entryCollectionsByPath.get(source.path)!),
selection,
);
return readSessionEntriesForFork(this.entryCollectionsByPath.get(source.path)!, selection);
});
return this.operations.enqueue(descriptor.operationKey, async () =>
this.createDocument(
@ -352,7 +372,7 @@ class JsonlSessionStore
parentSessionPath: string | undefined,
metadata: Record<string, unknown> | undefined,
entries: readonly SessionTreeEntry[],
): Promise<SessionReader<JsonlSessionMetadata>> {
): Promise<JsonlSessionMetadata> {
const dir = await this.getSessionDir(options.cwd);
getFileSystemResultOrThrow(
await this.fs.createDir(dir, { recursive: true }),
@ -378,40 +398,12 @@ class JsonlSessionStore
getFileSystemResultOrThrow(await this.fs.writeFile(path, content), `Failed to create session ${path}`);
this.entryCollectionsByPath.set(path, new SessionEntryCollection(entries));
this.operationKeysByPath.set(path, descriptor.operationKey);
return this.reader(metadataFromHeader(header, path));
return metadataFromHeader(header, path);
}
private reader(metadata: JsonlSessionMetadata): SessionReader<JsonlSessionMetadata> {
const operationKey = this.operationKey(metadata);
return {
metadata,
readHead: () => {
this.assertOpen();
return this.operations.enqueue(operationKey, () => this.entryCollection(metadata.path).readHead());
},
readEntry: (id) => {
this.assertOpen();
return this.operations.enqueue(operationKey, () => this.entryCollection(metadata.path).readEntry(id));
},
readEntries: (options) => {
this.assertOpen();
return this.operations.enqueue(operationKey, () =>
this.entryCollection(metadata.path).readEntries(options),
);
},
findEntriesOnBranch: (query) => {
this.assertOpen();
return this.operations.enqueue(operationKey, () =>
this.entryCollection(metadata.path).findEntriesOnBranch(query),
);
},
readPathToRootOrCompaction: (leafId) => {
this.assertOpen();
return this.operations.enqueue(operationKey, () =>
this.entryCollection(metadata.path).readPathToRootOrCompaction(leafId),
);
},
};
private readCollection<T>(metadata: JsonlSessionMetadata, read: (entries: SessionEntryCollection) => T): Promise<T> {
this.assertOpen();
return this.operations.enqueue(this.operationKey(metadata), () => read(this.entryCollection(metadata.path)));
}
private entryCollection(path: string): SessionEntryCollection {

119
packages/agent/src/harness/session/memory-store.ts
View file

@ -1,9 +1,16 @@
import type { SessionForkSelection, SessionMetadata, SessionReader, SessionStore, SessionTreeEntry } from "../types.ts";
import type {
SessionCatalog,
SessionEntries,
SessionForkSelection,
SessionMetadata,
SessionStore,
SessionTreeEntry,
} from "../types.ts";
import { SessionError } from "../types.ts";
import { readSessionEntriesForFork } from "./fork.ts";
import { KeyedOperationQueue } from "./keyed-operation-queue.ts";
import { createSessionId, createTimestamp } from "./repository.ts";
import { createSessionEntryCollectionReader, SessionEntryCollection } from "./session-entry-collection.ts";
import { SessionEntryCollection } from "./session-entry-collection.ts";
export type InMemorySessionCreateOptions = { id?: string };
@ -13,12 +20,33 @@ interface InMemorySessionState {
}
class InMemorySessionStore implements SessionStore<SessionMetadata, InMemorySessionCreateOptions, void> {
private readonly sessions = new Map<string, InMemorySessionState>();
private readonly states = new Map<string, InMemorySessionState>();
private readonly operations = new KeyedOperationQueue<string>();
private disposed = false;
private disposePromise: Promise<void> | undefined;
create(options: InMemorySessionCreateOptions = {}): Promise<SessionReader<SessionMetadata>> {
readonly sessions: SessionCatalog<SessionMetadata, InMemorySessionCreateOptions, void> = {
create: (options = {}) => this.create(options),
open: (metadata) => this.open(metadata),
list: () => this.list(),
delete: (metadata) => this.delete(metadata),
fork: (source, options, selection) => this.fork(source, options, selection),
};
readonly entries: SessionEntries<SessionMetadata> = {
readHead: (metadata) => this.read(metadata, (entries) => entries.readHead()),
readEntry: (metadata, id) => this.read(metadata, (entries) => entries.readEntry(id)),
readEntries: (metadata, options) => this.read(metadata, (entries) => entries.readEntries(options)),
findEntriesOnBranch: (metadata, query) => this.read(metadata, (entries) => entries.findEntriesOnBranch(query)),
readPathToRootOrCompaction: (metadata, leafId) =>
this.read(metadata, (entries) => entries.readPathToRootOrCompaction(leafId)),
append: (metadata, entry) => this.append(metadata, entry),
getLabel: (metadata, id) => this.read(metadata, (entries) => entries.getLabel(id)),
getName: (metadata) => this.read(metadata, (entries) => entries.getName()),
getStats: (metadata) => this.read(metadata, (entries) => entries.getStats()),
};
private create(options: InMemorySessionCreateOptions): Promise<SessionMetadata> {
this.assertOpen();
const id = options.id ?? createSessionId();
return this.operations.enqueue(id, () => {
@ -26,57 +54,57 @@ class InMemorySessionStore implements SessionStore<SessionMetadata, InMemorySess
metadata: { id, createdAt: createTimestamp() },
entries: new SessionEntryCollection(),
};
this.sessions.set(state.metadata.id, state);
return this.reader(state);
this.states.set(id, state);
return state.metadata;
});
}
load(metadata: SessionMetadata): Promise<SessionReader<SessionMetadata>> {
private open(metadata: SessionMetadata): Promise<SessionMetadata> {
this.assertOpen();
return this.operations.enqueue(metadata.id, () => this.reader(this.getState(metadata)));
return this.operations.enqueue(metadata.id, () => this.getState(metadata).metadata);
}
list(): Promise<SessionMetadata[]> {
private list(): Promise<SessionMetadata[]> {
this.assertOpen();
return this.operations.enqueueBarrier(() => [...this.sessions.values()].map((state) => state.metadata));
return this.operations.enqueueBarrier(() => [...this.states.values()].map((state) => state.metadata));
}
appendEntry(metadata: SessionMetadata, entry: SessionTreeEntry): Promise<void> {
private delete(metadata: SessionMetadata): Promise<void> {
this.assertOpen();
return this.operations.enqueue(metadata.id, () => {
const state = this.getState(metadata);
state.entries.append(entry);
this.states.delete(metadata.id);
});
}
delete(metadata: SessionMetadata): Promise<void> {
this.assertOpen();
return this.operations.enqueue(metadata.id, () => {
this.sessions.delete(metadata.id);
});
}
fork(
private fork(
source: SessionMetadata,
options: InMemorySessionCreateOptions,
selection: SessionForkSelection,
): Promise<SessionReader<SessionMetadata>> {
): Promise<SessionMetadata> {
this.assertOpen();
const id = options.id ?? createSessionId();
const sourceEntries = this.operations.enqueue(source.id, () => {
const sourceState = this.getState(source);
return readSessionEntriesForFork(
createSessionEntryCollectionReader(sourceState.metadata, sourceState.entries),
selection,
);
});
const sourceEntries = this.operations.enqueue(source.id, () =>
readSessionEntriesForFork(this.getState(source).entries, selection),
);
return this.operations.enqueue(id, async () => {
const state: InMemorySessionState = {
metadata: { id, createdAt: createTimestamp() },
entries: new SessionEntryCollection(await sourceEntries),
};
this.sessions.set(state.metadata.id, state);
return this.reader(state);
this.states.set(id, state);
return state.metadata;
});
}
private read<T>(metadata: SessionMetadata, read: (entries: SessionEntryCollection) => T): Promise<T> {
this.assertOpen();
return this.operations.enqueue(metadata.id, () => read(this.getState(metadata).entries));
}
private append(metadata: SessionMetadata, entry: SessionTreeEntry): Promise<void> {
this.assertOpen();
return this.operations.enqueue(metadata.id, () => {
this.getState(metadata).entries.append(entry);
});
}
@ -93,37 +121,10 @@ class InMemorySessionStore implements SessionStore<SessionMetadata, InMemorySess
}
private getState(metadata: SessionMetadata): InMemorySessionState {
const state = this.sessions.get(metadata.id);
const state = this.states.get(metadata.id);
if (!state) throw new SessionError("not_found", `Session not found: ${metadata.id}`);
return state;
}
private reader(state: InMemorySessionState): SessionReader<SessionMetadata> {
const reader = createSessionEntryCollectionReader(state.metadata, state.entries);
return {
metadata: reader.metadata,
readHead: () => {
this.assertOpen();
return this.operations.enqueue(state.metadata.id, () => reader.readHead());
},
readEntry: (id) => {
this.assertOpen();
return this.operations.enqueue(state.metadata.id, () => reader.readEntry(id));
},
readEntries: (options) => {
this.assertOpen();
return this.operations.enqueue(state.metadata.id, () => reader.readEntries(options));
},
findEntriesOnBranch: (query) => {
this.assertOpen();
return this.operations.enqueue(state.metadata.id, () => reader.findEntriesOnBranch(query));
},
readPathToRootOrCompaction: (leafId) => {
this.assertOpen();
return this.operations.enqueue(state.metadata.id, () => reader.readPathToRootOrCompaction(leafId));
},
};
}
}
export function createInMemorySessionStore(): SessionStore<SessionMetadata, InMemorySessionCreateOptions, void> {

27
packages/agent/src/harness/session/repository.ts
View file

@ -2,6 +2,7 @@ import { uuidv7 } from "@earendil-works/pi-ai";
import {
type FileError,
type Result,
type SessionCommitHandler,
type SessionCreateOptions,
SessionError,
type SessionForkOptions,
@ -12,7 +13,7 @@ import {
type SessionTreeEntry,
} from "../types.ts";
import { createSessionForkSelection } from "./fork.ts";
import { createSessionFromReader, type Session, type SessionContextBuildOptions } from "./session.ts";
import { createSession, type Session, type SessionContextBuildOptions } from "./session.ts";
export function createSessionId(): string {
return uuidv7();
@ -30,31 +31,38 @@ export class SessionRepository<
private readonly store: SessionStore<TMetadata, TCreateOptions, TListOptions>;
private readonly searchStore: SessionSearch<TMetadata> | null;
private readonly contextBuildOptions: SessionContextBuildOptions;
private readonly onCommit: SessionCommitHandler<TMetadata> | undefined;
constructor(options: {
store: SessionStore<TMetadata, TCreateOptions, TListOptions>;
search?: SessionSearch<TMetadata> | null;
contextBuildOptions?: SessionContextBuildOptions;
onCommit?: SessionCommitHandler<TMetadata>;
}) {
this.store = options.store;
this.searchStore = options.search ?? null;
this.contextBuildOptions = options.contextBuildOptions ?? {};
this.onCommit = options.onCommit;
}
async create(options: TCreateOptions): Promise<Session<TMetadata>> {
return createSessionFromReader(this.store, await this.store.create(options), this.contextBuildOptions);
const metadata = await this.store.sessions.create(options);
await this.onCommit?.({ kind: "session_created", metadata });
return createSession(this.store.entries, metadata, this.contextBuildOptions, this.onCommit);
}
async open(metadata: TMetadata): Promise<Session<TMetadata>> {
return createSessionFromReader(this.store, await this.store.load(metadata), this.contextBuildOptions);
const canonical = await this.store.sessions.open(metadata);
return createSession(this.store.entries, canonical, this.contextBuildOptions, this.onCommit);
}
async list(options?: TListOptions): Promise<TMetadata[]> {
return await this.store.list(options);
return await this.store.sessions.list(options);
}
async delete(metadata: TMetadata): Promise<void> {
await this.store.delete(metadata);
await this.store.sessions.delete(metadata);
await this.onCommit?.({ kind: "session_deleted", metadata });
}
async fork(source: TMetadata, options: SessionForkOptions & TCreateOptions): Promise<Session<TMetadata>> {
@ -62,11 +70,9 @@ export class SessionRepository<
const createOptions = { ...options };
delete createOptions.entryId;
delete createOptions.position;
return createSessionFromReader(
this.store,
await this.store.fork(source, createOptions, selection),
this.contextBuildOptions,
);
const metadata = await this.store.sessions.fork(source, createOptions, selection);
await this.onCommit?.({ kind: "session_forked", metadata, source });
return createSession(this.store.entries, metadata, this.contextBuildOptions, this.onCommit);
}
async search(options: Parameters<SessionSearch<TMetadata>["search"]>[0]): Promise<SessionSearchHit<TMetadata>[]> {
@ -82,6 +88,7 @@ export function createSessionRepository<
store: SessionStore<TMetadata, TCreateOptions, TListOptions>;
search?: SessionSearch<TMetadata> | null;
contextBuildOptions?: SessionContextBuildOptions;
onCommit?: SessionCommitHandler<TMetadata>;
}): SessionRepository<TMetadata, TCreateOptions, TListOptions> {
return new SessionRepository(options);
}

19
packages/agent/src/harness/session/search-backend.ts
View file

@ -1,15 +1,20 @@
import type {
SessionMetadata,
SessionReader,
SessionSearch,
SessionSearchHit,
SessionSearchOptions,
SessionTreeEntry,
} from "../types.ts";
import { findSessionEntryMatches } from "./repository.ts";
type SessionSearchSource<TMetadata extends SessionMetadata> = {
load(metadata: TMetadata): Promise<SessionReader<TMetadata>>;
list(): Promise<TMetadata[]>;
sessions: {
open(metadata: TMetadata): Promise<TMetadata>;
list(): Promise<TMetadata[]>;
};
entries: {
readEntries(metadata: TMetadata): Promise<readonly SessionTreeEntry[]>;
};
};
/** Searches canonical sessions directly and therefore has no index to maintain. */
@ -22,11 +27,13 @@ class ScanningSessionSearch<TMetadata extends SessionMetadata = SessionMetadata>
async search(options: SessionSearchOptions): Promise<SessionSearchHit<TMetadata>[]> {
const hits: SessionSearchHit<TMetadata>[] = [];
for (const metadata of await this.source.list()) {
for (const metadata of await this.source.sessions.list()) {
const cwd = (metadata as { cwd?: unknown }).cwd;
if (options.cwd !== undefined && cwd !== options.cwd) continue;
const reader = await this.source.load(metadata);
hits.push(...findSessionEntryMatches(reader.metadata, await reader.readEntries(), options.text));
const canonical = await this.source.sessions.open(metadata);
hits.push(
...findSessionEntryMatches(canonical, await this.source.entries.readEntries(canonical), options.text),
);
}
return hits;
}

91
packages/agent/src/harness/session/session-entry-collection.ts
View file

@ -2,17 +2,64 @@ import type {
SessionBranchQuery,
SessionEntryCursorOptions,
SessionHead,
SessionMetadata,
SessionReader,
SessionStats,
SessionTreeEntry,
} from "../types.ts";
import { SessionError } from "../types.ts";
interface SessionEntryProjection {
name: string | undefined;
labelsById: Map<string, string>;
stats: SessionStats;
}
function createProjection(): SessionEntryProjection {
return {
name: undefined,
labelsById: new Map(),
stats: { messageCount: 0, cachedTokens: 0, uncachedTokens: 0, totalTokens: 0, costTotal: 0 },
};
}
function applyProjection(projection: SessionEntryProjection, entry: SessionTreeEntry): void {
if (entry.type === "session_info") {
projection.name = entry.name?.trim() || undefined;
} else if (entry.type === "label") {
const label = entry.label?.trim();
if (label) projection.labelsById.set(entry.targetId, label);
else projection.labelsById.delete(entry.targetId);
}
if (entry.type === "message") projection.stats.messageCount += 1;
const usage =
entry.type === "message"
? entry.message.role === "assistant"
? entry.message.usage
: undefined
: entry.type === "compaction" || entry.type === "branch_summary"
? entry.usage
: undefined;
if (
!usage ||
typeof usage.input !== "number" ||
typeof usage.output !== "number" ||
typeof usage.cacheRead !== "number" ||
typeof usage.cacheWrite !== "number" ||
typeof usage.cost?.total !== "number"
) {
return;
}
projection.stats.cachedTokens += usage.cacheRead;
projection.stats.uncachedTokens += usage.input + usage.cacheWrite;
projection.stats.totalTokens += usage.input + usage.output + usage.cacheRead + usage.cacheWrite;
projection.stats.costTotal += usage.cost.total;
}
/** Ordered entries and derived indexes for an array-backed session store. */
export class SessionEntryCollection {
private entries: SessionTreeEntry[] = [];
private byId = new Map<string, SessionTreeEntry>();
private leafId: string | null = null;
private projection = createProjection();
constructor(entries: readonly SessionTreeEntry[] = []) {
this.replace(entries);
@ -29,11 +76,13 @@ export class SessionEntryCollection {
this.entries.push(entry);
this.byId.set(entry.id, entry);
this.leafId = entry.type === "leaf" ? entry.targetId : entry.id;
applyProjection(this.projection, entry);
}
replace(entries: readonly SessionTreeEntry[]): void {
const nextEntries = [...entries];
const nextById = new Map<string, SessionTreeEntry>();
const nextProjection = createProjection();
let nextLeafId: string | null = null;
for (const entry of nextEntries) {
if (nextById.has(entry.id)) {
@ -41,10 +90,12 @@ export class SessionEntryCollection {
}
nextById.set(entry.id, entry);
nextLeafId = entry.type === "leaf" ? entry.targetId : entry.id;
applyProjection(nextProjection, entry);
}
this.entries = nextEntries;
this.byId = nextById;
this.leafId = nextLeafId;
this.projection = nextProjection;
}
readHead(): SessionHead {
@ -101,6 +152,18 @@ export class SessionEntryCollection {
return query.limit === undefined ? entries : entries.slice(0, query.limit);
}
getLabel(id: string): string | undefined {
return this.projection.labelsById.get(id);
}
getName(): string | undefined {
return this.projection.name;
}
getStats(): SessionStats {
return { ...this.projection.stats };
}
readPathToRootOrCompaction(requestedLeafId: string | null): readonly SessionTreeEntry[] {
if (requestedLeafId === null) return [];
const path: SessionTreeEntry[] = [];
@ -122,27 +185,3 @@ export class SessionEntryCollection {
return path.reverse();
}
}
export function createSessionEntryCollectionReader<TMetadata extends SessionMetadata>(
metadata: TMetadata,
entries: SessionEntryCollection,
): SessionReader<TMetadata> {
return {
metadata,
async readHead() {
return entries.readHead();
},
async readEntry(id) {
return entries.readEntry(id);
},
async readEntries(options) {
return entries.readEntries(options);
},
async findEntriesOnBranch(query) {
return entries.findEntriesOnBranch(query);
},
async readPathToRootOrCompaction(leafId) {
return entries.readPathToRootOrCompaction(leafId);
},
};
}

162
packages/agent/src/harness/session/session.ts
View file

@ -12,13 +12,13 @@ import type {
MessageEntry,
ModelChangeEntry,
SessionBranchQuery,
SessionCommitHandler,
SessionContext,
SessionEntries,
SessionEntryCursorOptions,
SessionInfoEntry,
SessionMetadata,
SessionReader,
SessionStats,
SessionStore,
SessionTreeEntry,
ThinkingLevelChangeEntry,
} from "../types.ts";
@ -150,129 +150,29 @@ export function buildSessionContext(
return { ...state, messages };
}
class HydratedSessionState {
private readonly entries: SessionTreeEntry[];
private readonly labelsById = new Map<string, string>();
constructor(entries: readonly SessionTreeEntry[]) {
this.entries = [...entries];
for (const entry of this.entries) this.applyProjection(entry);
}
getLabel(id: string): string | undefined {
return this.labelsById.get(id);
}
getSessionName(): string | undefined {
for (let i = this.entries.length - 1; i >= 0; i--) {
const entry = this.entries[i]!;
if (entry.type === "session_info") return entry.name?.trim() || undefined;
}
return undefined;
}
getSessionStats(): SessionStats {
let messageCount = 0;
let cachedTokens = 0;
let uncachedTokens = 0;
let totalTokens = 0;
let costTotal = 0;
for (const entry of this.entries) {
if (entry.type === "message") messageCount += 1;
const usage =
entry.type === "message"
? entry.message.role === "assistant"
? entry.message.usage
: undefined
: entry.type === "compaction" || entry.type === "branch_summary"
? entry.usage
: undefined;
if (
!usage ||
typeof usage.input !== "number" ||
typeof usage.output !== "number" ||
typeof usage.cacheRead !== "number" ||
typeof usage.cacheWrite !== "number" ||
typeof usage.cost?.total !== "number"
)
continue;
cachedTokens += usage.cacheRead;
uncachedTokens += usage.input + usage.cacheWrite;
totalTokens += usage.input + usage.output + usage.cacheRead + usage.cacheWrite;
costTotal += usage.cost.total;
}
return { messageCount, cachedTokens, uncachedTokens, totalTokens, costTotal };
}
private applyProjection(entry: SessionTreeEntry): void {
if (entry.type !== "label") return;
const label = entry.label?.trim();
if (label) this.labelsById.set(entry.targetId, label);
else this.labelsById.delete(entry.targetId);
}
}
export interface Session<TMetadata extends SessionMetadata = SessionMetadata> {
getMetadata(): Promise<TMetadata>;
getLeafId(): Promise<string | null>;
getEntry(id: string): Promise<SessionTreeEntry | undefined>;
getEntries(options?: SessionEntryCursorOptions): Promise<SessionTreeEntry[]>;
getBranch(fromId?: string | null): Promise<SessionTreeEntry[]>;
findEntriesOnBranch(query?: SessionBranchQuery): Promise<SessionTreeEntry[]>;
findEntryOnBranch(query?: SessionBranchQuery): Promise<SessionTreeEntry | undefined>;
buildContextEntries(options?: SessionContextBuildOptions): Promise<SessionTreeEntry[]>;
buildContext(options?: SessionContextBuildOptions): Promise<SessionContext>;
getLabel(id: string): Promise<string | undefined>;
getSessionStats(): Promise<SessionStats>;
getSessionName(): Promise<string | undefined>;
appendMessage(message: AgentMessage): Promise<string>;
appendThinkingLevelChange(thinkingLevel: string): Promise<string>;
appendModelChange(provider: string, modelId: string): Promise<string>;
appendActiveToolsChange(activeToolNames: string[]): Promise<string>;
appendCompaction<T = unknown>(
summary: string,
firstKeptEntryId: string | undefined,
tokensBefore: number,
details?: T,
fromHook?: boolean,
usage?: Usage,
retainedTail?: AgentMessage[],
): Promise<string>;
appendCustomEntry(customType: string, data?: unknown): Promise<string>;
appendCustomMessageEntry<T = unknown>(
customType: string,
content: string | (TextContent | ImageContent)[],
display: boolean,
details?: T,
): Promise<string>;
appendLabel(targetId: string, label: string | undefined): Promise<string>;
appendSessionName(name: string): Promise<string>;
moveTo(
entryId: string | null,
summary?: { summary: string; details?: unknown; usage?: Usage; fromHook?: boolean },
): Promise<string | undefined>;
}
class StoreSession<TMetadata extends SessionMetadata = SessionMetadata> implements Session<TMetadata> {
private readonly store: Pick<SessionStore<TMetadata>, "appendEntry">;
private readonly reader: SessionReader<TMetadata>;
export class Session<TMetadata extends SessionMetadata = SessionMetadata> {
private readonly entries: SessionEntries<TMetadata>;
private readonly metadata: TMetadata;
private hydratedStatePromise: Promise<HydratedSessionState> | undefined;
private leafId: string | null;
private readonly contextBuildOptions: SessionContextBuildOptions;
private readonly onEntryCommitted: ((entry: SessionTreeEntry) => Promise<void>) | undefined;
private appendTail: Promise<void> = Promise.resolve();
/** @internal Construct sessions through SessionRepository. */
constructor(
store: Pick<SessionStore<TMetadata>, "appendEntry">,
reader: SessionReader<TMetadata>,
entries: SessionEntries<TMetadata>,
metadata: TMetadata,
leafId: string | null,
contextBuildOptions: SessionContextBuildOptions = {},
onCommit?: SessionCommitHandler<TMetadata>,
) {
this.store = store;
this.reader = reader;
this.metadata = reader.metadata;
this.entries = entries;
this.metadata = metadata;
this.leafId = leafId;
this.contextBuildOptions = contextBuildOptions;
this.onEntryCommitted = onCommit
? (entry) => onCommit({ kind: "entry_appended", metadata: this.metadata, entry })
: undefined;
}
async getMetadata(): Promise<TMetadata> {
@ -282,19 +182,21 @@ class StoreSession<TMetadata extends SessionMetadata = SessionMetadata> implemen
return this.leafId;
}
async getEntry(id: string): Promise<SessionTreeEntry | undefined> {
return this.reader.readEntry(id);
return this.entries.readEntry(this.metadata, id);
}
async getEntries(options?: SessionEntryCursorOptions): Promise<SessionTreeEntry[]> {
return [...(await this.reader.readEntries(options))];
return [...(await this.entries.readEntries(this.metadata, options))];
}
async getBranch(fromId?: string | null): Promise<SessionTreeEntry[]> {
return [...(await this.reader.readPathToRootOrCompaction(fromId === undefined ? this.leafId : fromId))];
return [
...(await this.entries.readPathToRootOrCompaction(this.metadata, fromId === undefined ? this.leafId : fromId)),
];
}
async findEntriesOnBranch(query: SessionBranchQuery = {}): Promise<SessionTreeEntry[]> {
return [
...(await this.reader.findEntriesOnBranch({
...(await this.entries.findEntriesOnBranch(this.metadata, {
...query,
start: query.start === undefined ? this.leafId : query.start,
})),
@ -324,18 +226,13 @@ class StoreSession<TMetadata extends SessionMetadata = SessionMetadata> implemen
}
async getLabel(id: string): Promise<string | undefined> {
return (await this.getHydratedState()).getLabel(id);
return this.entries.getLabel(this.metadata, id);
}
async getSessionStats(): Promise<SessionStats> {
return (await this.getHydratedState()).getSessionStats();
return this.entries.getStats(this.metadata);
}
async getSessionName(): Promise<string | undefined> {
return (await this.getHydratedState()).getSessionName();
}
private async getHydratedState(): Promise<HydratedSessionState> {
this.hydratedStatePromise ??= this.reader.readEntries().then((entries) => new HydratedSessionState(entries));
return this.hydratedStatePromise;
return this.entries.getName(this.metadata);
}
private async createEntryId(): Promise<string> {
@ -355,9 +252,9 @@ class StoreSession<TMetadata extends SessionMetadata = SessionMetadata> implemen
parentId: this.leafId,
timestamp: new Date().toISOString(),
});
await this.store.appendEntry(this.metadata, entry);
await this.entries.append(this.metadata, entry);
this.leafId = entry.type === "leaf" ? entry.targetId : entry.id;
this.hydratedStatePromise = undefined;
await this.onEntryCommitted?.(entry);
return entry;
});
this.appendTail = operation.then(
@ -535,10 +432,11 @@ class StoreSession<TMetadata extends SessionMetadata = SessionMetadata> implemen
}
/** @internal Construct sessions only through SessionRepository. */
export async function createSessionFromReader<TMetadata extends SessionMetadata>(
store: Pick<SessionStore<TMetadata>, "appendEntry">,
reader: SessionReader<TMetadata>,
export async function createSession<TMetadata extends SessionMetadata>(
entries: SessionEntries<TMetadata>,
metadata: TMetadata,
contextBuildOptions: SessionContextBuildOptions = {},
onCommit?: SessionCommitHandler<TMetadata>,
): Promise<Session<TMetadata>> {
return new StoreSession(store, reader, (await reader.readHead()).leafId, contextBuildOptions);
return new Session(entries, metadata, (await entries.readHead(metadata)).leafId, contextBuildOptions, onCommit);
}

51
packages/agent/src/harness/types.ts
View file

@ -555,15 +555,32 @@ export interface SessionHead {
leafId: string | null;
}
/** Canonical reads for one stored session. Its lifetime is owned by the containing {@link SessionStore}. */
export interface SessionReader<TMetadata extends SessionMetadata = SessionMetadata> {
readonly metadata: TMetadata;
export interface SessionCatalog<
TMetadata extends SessionMetadata = SessionMetadata,
TCreateOptions extends SessionCreateOptions = SessionCreateOptions,
TListOptions = void,
> {
create(options: TCreateOptions): Promise<TMetadata>;
open(metadata: TMetadata): Promise<TMetadata>;
list(options?: TListOptions): Promise<TMetadata[]>;
delete(metadata: TMetadata): Promise<void>;
fork(source: TMetadata, options: TCreateOptions, selection: SessionForkSelection): Promise<TMetadata>;
}
export interface SessionEntries<TMetadata extends SessionMetadata = SessionMetadata> {
/** Rejects with `invalid_session` when a non-null active leaf does not reference a stored entry. */
readHead(): Promise<SessionHead>;
readEntry(id: string): Promise<SessionTreeEntry | undefined>;
readEntries(options?: SessionEntryCursorOptions): Promise<readonly SessionTreeEntry[]>;
findEntriesOnBranch(query: SessionBranchQuery & { start: string | null }): Promise<readonly SessionTreeEntry[]>;
readPathToRootOrCompaction(leafId: string | null): Promise<readonly SessionTreeEntry[]>;
readHead(metadata: TMetadata): Promise<SessionHead>;
readEntry(metadata: TMetadata, id: string): Promise<SessionTreeEntry | undefined>;
readEntries(metadata: TMetadata, options?: SessionEntryCursorOptions): Promise<readonly SessionTreeEntry[]>;
findEntriesOnBranch(
metadata: TMetadata,
query: SessionBranchQuery & { start: string | null },
): Promise<readonly SessionTreeEntry[]>;
readPathToRootOrCompaction(metadata: TMetadata, leafId: string | null): Promise<readonly SessionTreeEntry[]>;
append(metadata: TMetadata, entry: SessionTreeEntry): Promise<void>;
getLabel(metadata: TMetadata, id: string): Promise<string | undefined>;
getName(metadata: TMetadata): Promise<string | undefined>;
getStats(metadata: TMetadata): Promise<SessionStats>;
}
/** Owns persistence and resources shared by all sessions in a repository. */
@ -572,14 +589,20 @@ export interface SessionStore<
TCreateOptions extends SessionCreateOptions = SessionCreateOptions,
TListOptions = void,
> extends AsyncDisposable {
create(options: TCreateOptions): Promise<SessionReader<TMetadata>>;
load(metadata: TMetadata): Promise<SessionReader<TMetadata>>;
list(options?: TListOptions): Promise<TMetadata[]>;
appendEntry(metadata: TMetadata, entry: SessionTreeEntry): Promise<void>;
delete(metadata: TMetadata): Promise<void>;
fork(source: TMetadata, options: TCreateOptions, selection: SessionForkSelection): Promise<SessionReader<TMetadata>>;
readonly sessions: SessionCatalog<TMetadata, TCreateOptions, TListOptions>;
readonly entries: SessionEntries<TMetadata>;
}
export type SessionCommit<TMetadata extends SessionMetadata = SessionMetadata> =
| { kind: "session_created"; metadata: TMetadata }
| { kind: "session_forked"; metadata: TMetadata; source: TMetadata }
| { kind: "session_deleted"; metadata: TMetadata }
| { kind: "entry_appended"; metadata: TMetadata; entry: SessionTreeEntry };
export type SessionCommitHandler<TMetadata extends SessionMetadata = SessionMetadata> = (
commit: SessionCommit<TMetadata>,
) => Promise<void>;
export interface JsonlSessionCreateOptions extends SessionCreateOptions {
cwd: string;
parentSessionPath?: string;

1
packages/agent/src/index.ts
View file

@ -47,6 +47,7 @@ export {
type ContextEntryTransform,
type CustomEntryContextMessageProjector,
defaultContextEntryTransform,
Session,
type SessionContextBuildOptions,
sessionEntryToContextMessages,
} from "./harness/session/session.ts";

19
packages/agent/test/harness/agent-harness.test.ts
View file

@ -102,17 +102,16 @@ async function createBlockingSession(expectedWrites: number): Promise<{
const releaseWrites = deferred();
let writesStarted = 0;
const blockingStore: ReturnType<typeof createInMemorySessionStore> = {
create: (options) => source.create(options),
load: (metadata) => source.load(metadata),
list: (options) => source.list(options),
async appendEntry(metadata, entry) {
writesStarted++;
if (writesStarted === expectedWrites) allWritesStarted.resolve();
await releaseWrites.promise;
await source.appendEntry(metadata, entry);
sessions: source.sessions,
entries: {
...source.entries,
async append(metadata, entry) {
writesStarted++;
if (writesStarted === expectedWrites) allWritesStarted.resolve();
await releaseWrites.promise;
await source.entries.append(metadata, entry);
},
},
delete: (metadata) => source.delete(metadata),
fork: (metadata, options, selection) => source.fork(metadata, options, selection),
[Symbol.asyncDispose]: () => source[Symbol.asyncDispose](),
};
const session = await createSessionRepository({ store: blockingStore }).create({});

20
packages/agent/test/harness/branch-query.test.ts
View file

@ -104,8 +104,8 @@ describe("bounded session branch queries", () => {
it("rejects corrupt parent chains in array-backed stores", async () => {
const store = createInMemorySessionStore();
ownedStores.push(store);
const reader = await store.create({ id: "corrupt-memory" });
await store.appendEntry(reader.metadata, {
const metadata = await store.sessions.create({ id: "corrupt-memory" });
await store.entries.append(metadata, {
type: "message",
id: "orphan",
parentId: "missing-parent",
@ -114,30 +114,34 @@ describe("bounded session branch queries", () => {
});
expect(
(await reader.findEntriesOnBranch({ start: "orphan", stopAtId: "orphan" })).map((entry) => entry.id),
(await store.entries.findEntriesOnBranch(metadata, { start: "orphan", stopAtId: "orphan" })).map(
(entry) => entry.id,
),
).toEqual(["orphan"]);
expect(
(await reader.findEntriesOnBranch({ start: "orphan", stopAtType: "message" })).map((entry) => entry.id),
(await store.entries.findEntriesOnBranch(metadata, { start: "orphan", stopAtType: "message" })).map(
(entry) => entry.id,
),
).toEqual(["orphan"]);
await expect(reader.findEntriesOnBranch({ start: "orphan" })).rejects.toMatchObject({
await expect(store.entries.findEntriesOnBranch(metadata, { start: "orphan" })).rejects.toMatchObject({
code: "invalid_session",
message: "Entry missing-parent not found",
});
await store.appendEntry(reader.metadata, {
await store.entries.append(metadata, {
type: "message",
id: "cycle-a",
parentId: "cycle-b",
timestamp: "2026-01-01T00:00:01.000Z",
message: createUserMessage("a"),
});
await store.appendEntry(reader.metadata, {
await store.entries.append(metadata, {
type: "message",
id: "cycle-b",
parentId: "cycle-a",
timestamp: "2026-01-01T00:00:02.000Z",
message: createUserMessage("b"),
});
await expect(reader.findEntriesOnBranch({ start: "cycle-b" })).rejects.toMatchObject({
await expect(store.entries.findEntriesOnBranch(metadata, { start: "cycle-b" })).rejects.toMatchObject({
code: "invalid_session",
message: "Session branch contains a cycle at cycle-b",
});

210
packages/agent/test/harness/repo.test.ts
View file

@ -7,7 +7,7 @@ import {
type InMemorySessionCreateOptions,
} from "../../src/harness/session/memory-store.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import type { SessionForkSelection, SessionMetadata, SessionReader, SessionStore } from "../../src/harness/types.ts";
import type { SessionCommit, SessionForkSelection, SessionMetadata, SessionStore } from "../../src/harness/types.ts";
import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts";
afterEach(() => {
@ -154,37 +154,36 @@ function createCountingInMemorySessionStore(): {
readPathCount: 0,
forkSelections: [],
};
const countReads = (reader: SessionReader<SessionMetadata>): SessionReader<SessionMetadata> => ({
metadata: reader.metadata,
readHead: () => {
counter.readHeadCount += 1;
return reader.readHead();
},
readEntry: (id) => reader.readEntry(id),
readEntries: (options) => {
counter.readEntriesCount += 1;
return reader.readEntries(options);
},
findEntriesOnBranch: (query) => reader.findEntriesOnBranch(query),
readPathToRootOrCompaction: (leafId) => {
counter.readPathCount += 1;
return reader.readPathToRootOrCompaction(leafId);
},
});
return {
counter,
store: {
create: (options) => source.create(options),
async load(metadata) {
counter.loadCount += 1;
return countReads(await source.load(metadata));
sessions: {
create: (options) => source.sessions.create(options),
async open(metadata) {
counter.loadCount += 1;
return source.sessions.open(metadata);
},
list: (options) => source.sessions.list(options),
delete: (metadata) => source.sessions.delete(metadata),
fork: (metadata, options, selection) => {
counter.forkSelections.push(selection);
return source.sessions.fork(metadata, options, selection);
},
},
list: (options) => source.list(options),
appendEntry: (metadata, entry) => source.appendEntry(metadata, entry),
delete: (metadata) => source.delete(metadata),
fork: (metadata, options, selection) => {
counter.forkSelections.push(selection);
return source.fork(metadata, options, selection);
entries: {
...source.entries,
readHead: (metadata) => {
counter.readHeadCount += 1;
return source.entries.readHead(metadata);
},
readEntries: (metadata, options) => {
counter.readEntriesCount += 1;
return source.entries.readEntries(metadata, options);
},
readPathToRootOrCompaction: (metadata, leafId) => {
counter.readPathCount += 1;
return source.entries.readPathToRootOrCompaction(metadata, leafId);
},
},
[Symbol.asyncDispose]: () => source[Symbol.asyncDispose](),
},
@ -221,6 +220,56 @@ describe("InMemorySessionStore", () => {
await expect(repo.open(metadata)).rejects.toThrow("Session not found: session-1");
});
it("fans committed changes out through the consumer callback", async () => {
const commits: SessionCommit[] = [];
const repo = createSessionRepository({
store: createInMemorySessionStore(),
onCommit: async (commit) => {
commits.push(commit);
},
});
const session = await repo.create({ id: "session-1" });
const metadata = await session.getMetadata();
const entryId = await session.appendMessage(createUserMessage("one"));
const fork = await repo.fork(metadata, { id: "session-2" });
await repo.delete(await fork.getMetadata());
expect(commits.map((commit) => commit.kind)).toEqual([
"session_created",
"entry_appended",
"session_forked",
"session_deleted",
]);
expect(commits[1]).toMatchObject({ kind: "entry_appended", entry: { id: entryId } });
});
it("leaves fan-out failure policy to the consumer", async () => {
const required = await createSessionRepository({
store: createInMemorySessionStore(),
onCommit: async (commit) => {
if (commit.kind === "entry_appended") throw new Error("index unavailable");
},
}).create({ id: "required" });
await expect(required.appendMessage(createUserMessage("required"))).rejects.toThrow("index unavailable");
await expect(required.getEntries()).resolves.toHaveLength(1);
const failures: Error[] = [];
const bestEffort = await createSessionRepository({
store: createInMemorySessionStore(),
onCommit: async (commit) => {
if (commit.kind !== "entry_appended") return;
try {
throw new Error("index unavailable");
} catch (error) {
if (error instanceof Error) failures.push(error);
else throw error;
}
},
}).create({ id: "best-effort" });
await expect(bestEffort.appendMessage(createUserMessage("best effort"))).resolves.toBeTypeOf("string");
expect(failures).toHaveLength(1);
});
it("delegates full-session fork selection without opening the source", async () => {
const { store, counter } = createCountingInMemorySessionStore();
const repo = createSessionRepository({ store });
@ -232,8 +281,8 @@ describe("InMemorySessionStore", () => {
expect((await fork.getEntries()).map((entry) => entry.id)).toHaveLength(1);
expect(counter).toMatchObject({
loadCount: 0,
readHeadCount: 0,
readEntriesCount: 0,
readHeadCount: 2,
readEntriesCount: 1,
readPathCount: 0,
forkSelections: [{ kind: "all" }],
});
@ -260,7 +309,7 @@ describe("InMemorySessionStore", () => {
await opened.buildContext();
expect(counter).toMatchObject({ loadCount: 1, readHeadCount: 1, readEntriesCount: 0, readPathCount: 1 });
expect(counter).toMatchObject({ loadCount: 1, readHeadCount: 2, readEntriesCount: 0, readPathCount: 1 });
});
it("rejects repository operations and session writes after store disposal", async () => {
@ -304,12 +353,12 @@ describe("JsonlSessionStore", () => {
env.writtenPaths.add(sourcePath);
const store = createJsonlSessionStore({ fs: env, sessionsRoot: root });
const options = { cwd: "/tmp/target", id: "shared-id" };
const first = store.create(options);
const first = store.sessions.create(options);
await env.firstWriteStarted.promise;
const second =
secondOperation === "create"
? store.create(options)
: store.fork(
? store.sessions.create(options)
: store.sessions.fork(
{
id: "source",
createdAt: "2025-01-01T00:00:00.000Z",
@ -337,24 +386,24 @@ describe("JsonlSessionStore", () => {
it("encodes custom session IDs used in filenames", async () => {
const root = createTempDir();
const store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const snapshot = await store.create({ cwd: root, id: "../unsafe\\id" });
const snapshot = await store.sessions.create({ cwd: root, id: "../unsafe\\id" });
expect(snapshot.metadata.path).toContain("..%2Funsafe%5Cid.jsonl");
expect(existsSync(snapshot.metadata.path)).toBe(true);
expect(snapshot.path).toContain("..%2Funsafe%5Cid.jsonl");
expect(existsSync(snapshot.path)).toBe(true);
});
it("allows appends to different sessions to run concurrently", async () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const store = createJsonlSessionStore({ fs: env, sessionsRoot: root });
const first = await store.create({ cwd: "/tmp/first", id: "first" });
const second = await store.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.metadata.path);
const secondBlock = env.blockAppend(second.metadata.path);
const first = await store.sessions.create({ cwd: "/tmp/first", id: "first" });
const second = await store.sessions.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.path);
const secondBlock = env.blockAppend(second.path);
const firstAppend = store.appendEntry(first.metadata, createEntry("first-entry"));
const firstAppend = store.entries.append(first, createEntry("first-entry"));
await firstBlock.started.promise;
const secondAppend = store.appendEntry(second.metadata, createEntry("second-entry"));
const secondAppend = store.entries.append(second, createEntry("second-entry"));
const secondStartedBeforeFirstFinished = await settlesBeforeNextTurn(secondBlock.started.promise);
firstBlock.release.resolve();
@ -369,12 +418,10 @@ describe("JsonlSessionStore", () => {
const store = createJsonlSessionStore({ fs: env, sessionsRoot: root });
const snapshots = [];
for (let index = 0; index < 6; index++) {
snapshots.push(await store.create({ cwd: `/tmp/session-${index}`, id: `session-${index}` }));
snapshots.push(await store.sessions.create({ cwd: `/tmp/session-${index}`, id: `session-${index}` }));
}
const blocks = snapshots.map((snapshot) => env.blockAppend(snapshot.metadata.path));
const appends = snapshots.map((snapshot, index) =>
store.appendEntry(snapshot.metadata, createEntry(`entry-${index}`)),
);
const blocks = snapshots.map((snapshot) => env.blockAppend(snapshot.path));
const appends = snapshots.map((snapshot, index) => store.entries.append(snapshot, createEntry(`entry-${index}`)));
await Promise.all(blocks.slice(0, 4).map((block) => block.started.promise));
const fifthStartedWithoutCapacity = await settlesBeforeNextTurn(blocks[4]!.started.promise);
const sixthStartedWithoutCapacity = await settlesBeforeNextTurn(blocks[5]!.started.promise);
@ -390,13 +437,13 @@ describe("JsonlSessionStore", () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const store = createJsonlSessionStore({ fs: env, sessionsRoot: root, maxConcurrentOperations: 1 });
const first = await store.create({ cwd: "/tmp/first", id: "first" });
const second = await store.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.metadata.path);
const secondBlock = env.blockAppend(second.metadata.path);
const firstAppend = store.appendEntry(first.metadata, createEntry("first-entry"));
const first = await store.sessions.create({ cwd: "/tmp/first", id: "first" });
const second = await store.sessions.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.path);
const secondBlock = env.blockAppend(second.path);
const firstAppend = store.entries.append(first, createEntry("first-entry"));
await firstBlock.started.promise;
const secondAppend = store.appendEntry(second.metadata, createEntry("second-entry"));
const secondAppend = store.entries.append(second, createEntry("second-entry"));
const secondStartedBeforeFirstFinished = await settlesBeforeNextTurn(secondBlock.started.promise);
firstBlock.release.resolve();
@ -426,13 +473,13 @@ describe("JsonlSessionStore", () => {
sessionsRoot: root,
maxConcurrentOperations: 1,
});
const first = await store.create({ cwd: "/tmp/first", id: "first" });
const second = await store.create({ cwd: "/tmp/second", id: "second" });
const first = await store.sessions.create({ cwd: "/tmp/first", id: "first" });
const second = await store.sessions.create({ cwd: "/tmp/second", id: "second" });
const duplicate = createEntry("duplicate-entry");
await store.appendEntry(first.metadata, duplicate);
await store.entries.append(first, duplicate);
const failure = store.appendEntry(first.metadata, duplicate);
const nextSession = store.appendEntry(second.metadata, createEntry("next-entry"));
const failure = store.entries.append(first, duplicate);
const nextSession = store.entries.append(second, createEntry("next-entry"));
await expect(failure).rejects.toThrow("Entry duplicate-entry already exists");
await expect(nextSession).resolves.toBeUndefined();
@ -442,19 +489,20 @@ describe("JsonlSessionStore", () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const store = createJsonlSessionStore({ fs: env, sessionsRoot: root });
const snapshot = await store.create({ cwd: root, id: "session" });
const block = env.blockAppend(snapshot.metadata.path);
const snapshot = await store.sessions.create({ cwd: root, id: "session" });
const block = env.blockAppend(snapshot.path);
const firstAppend = store.appendEntry(snapshot.metadata, createEntry("first-entry"));
const firstAppend = store.entries.append(snapshot, createEntry("first-entry"));
await block.started.promise;
const secondAppend = store.appendEntry(snapshot.metadata, createEntry("second-entry"));
const secondAppend = store.entries.append(snapshot, createEntry("second-entry"));
await new Promise<void>((resolve) => setImmediate(resolve));
const appendCountWhileFirstWasBlocked = env.appendCounts.get(snapshot.metadata.path);
const appendCountWhileFirstWasBlocked = env.appendCounts.get(snapshot.path);
block.release.resolve();
await Promise.all([firstAppend, secondAppend]);
expect(appendCountWhileFirstWasBlocked).toBe(1);
expect((await (await store.load(snapshot.metadata)).readEntries()).map((entry) => entry.id)).toEqual([
await store.sessions.open(snapshot);
expect((await store.entries.readEntries(snapshot)).map((entry) => entry.id)).toEqual([
"first-entry",
"second-entry",
]);
@ -464,16 +512,16 @@ describe("JsonlSessionStore", () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const store = createJsonlSessionStore({ fs: env, sessionsRoot: root });
const first = await store.create({ cwd: "/tmp/first", id: "first" });
const second = await store.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.metadata.path);
const secondBlock = env.blockAppend(second.metadata.path);
const first = await store.sessions.create({ cwd: "/tmp/first", id: "first" });
const second = await store.sessions.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.path);
const secondBlock = env.blockAppend(second.path);
const listBlock = env.blockNextList();
const firstAppend = store.appendEntry(first.metadata, createEntry("first-entry"));
const firstAppend = store.entries.append(first, createEntry("first-entry"));
await firstBlock.started.promise;
const list = store.list();
const secondAppend = store.appendEntry(second.metadata, createEntry("second-entry"));
const list = store.sessions.list();
const secondAppend = store.entries.append(second, createEntry("second-entry"));
const listStartedBeforeFirstFinished = await settlesBeforeNextTurn(listBlock.started.promise);
firstBlock.release.resolve();
@ -494,12 +542,12 @@ describe("JsonlSessionStore", () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const store = createJsonlSessionStore({ fs: env, sessionsRoot: root });
const first = await store.create({ cwd: "/tmp/first", id: "first" });
const second = await store.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.metadata.path);
const secondBlock = env.blockAppend(second.metadata.path);
const firstAppend = store.appendEntry(first.metadata, createEntry("first-entry"));
const secondAppend = store.appendEntry(second.metadata, createEntry("second-entry"));
const first = await store.sessions.create({ cwd: "/tmp/first", id: "first" });
const second = await store.sessions.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.path);
const secondBlock = env.blockAppend(second.path);
const firstAppend = store.entries.append(first, createEntry("first-entry"));
const secondAppend = store.entries.append(second, createEntry("second-entry"));
await Promise.all([firstBlock.started.promise, secondBlock.started.promise]);
let disposed = false;
@ -584,8 +632,8 @@ describe("JsonlSessionStore", () => {
const root = createTempDir();
const store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const repo = createSessionRepository({ store });
const reader = await store.create({ cwd: root, id: "session-1" });
await store.appendEntry(reader.metadata, {
const metadata = await store.sessions.create({ cwd: root, id: "session-1" });
await store.entries.append(metadata, {
type: "leaf",
id: "leaf",
parentId: null,
@ -593,7 +641,7 @@ describe("JsonlSessionStore", () => {
timestamp: "2026-01-01T00:00:00.000Z",
});
await expect(repo.open(reader.metadata)).rejects.toMatchObject({
await expect(repo.open(metadata)).rejects.toMatchObject({
code: "invalid_session",
message: "Entry missing not found",
});

22
packages/agent/test/harness/session-backends.test.ts
View file

@ -140,7 +140,7 @@ describe("JsonlSessionStore", () => {
it("enforces entry uniqueness and does not recreate deleted files", async () => {
const root = createTempDir();
const store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const snapshot = await store.create({ cwd: root, id: "session-1" });
const snapshot = await store.sessions.create({ cwd: root, id: "session-1" });
const entry = {
type: "message" as const,
id: "entry-1",
@ -148,20 +148,18 @@ describe("JsonlSessionStore", () => {
timestamp: "2026-01-01T00:00:00.000Z",
message: createUserMessage("one"),
};
await store.appendEntry(snapshot.metadata, entry);
await expect(store.appendEntry(snapshot.metadata, entry)).rejects.toThrow("Entry entry-1 already exists");
await store.delete(snapshot.metadata);
await expect(store.appendEntry(snapshot.metadata, { ...entry, id: "entry-2" })).rejects.toThrow(
"Session not found",
);
expect(existsSync(snapshot.metadata.path)).toBe(false);
await store.entries.append(snapshot, entry);
await expect(store.entries.append(snapshot, entry)).rejects.toThrow("Entry entry-1 already exists");
await store.sessions.delete(snapshot);
await expect(store.entries.append(snapshot, { ...entry, id: "entry-2" })).rejects.toThrow("Session not found");
expect(existsSync(snapshot.path)).toBe(false);
});
it("scopes entry uniqueness to the session path", async () => {
const root = createTempDir();
const store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const first = await store.create({ cwd: "/tmp/first", id: "shared-session-id" });
const second = await store.create({ cwd: "/tmp/second", id: "shared-session-id" });
const first = await store.sessions.create({ cwd: "/tmp/first", id: "shared-session-id" });
const second = await store.sessions.create({ cwd: "/tmp/second", id: "shared-session-id" });
const entry = {
type: "message" as const,
id: "shared-entry-id",
@ -170,8 +168,8 @@ describe("JsonlSessionStore", () => {
message: createUserMessage("one"),
};
await store.appendEntry(first.metadata, entry);
await expect(store.appendEntry(second.metadata, entry)).resolves.toBeUndefined();
await store.entries.append(first, entry);
await expect(store.entries.append(second, entry)).resolves.toBeUndefined();
});
it("rejects non-object header metadata", async () => {

82
packages/storage/sqlite-node/src/sqlite/session-store.ts
View file

@ -1,6 +1,7 @@
import type {
SessionCatalog,
SessionEntries,
SessionForkSelection,
SessionReader,
SessionStore,
SessionTreeEntry,
} from "@earendil-works/pi-agent-core";
@ -73,13 +74,35 @@ class SqliteSessionStore
private readonly operations = new SerialOperationQueue();
private readonly writers = new Map<string, SqliteSessionConnection>();
readonly sessions: SessionCatalog<SqliteSessionMetadata, SqliteSessionCreateOptions, SqliteSessionListOptions> = {
create: (options) => this.create(options),
open: (metadata) => this.open(metadata),
list: (options) => this.list(options),
delete: (metadata) => this.delete(metadata),
fork: (source, options, selection) => this.fork(source, options, selection),
};
readonly entries: SessionEntries<SqliteSessionMetadata> = {
readHead: (metadata) => this.read(metadata, (connection) => connection.readHead()),
readEntry: (metadata, id) => this.read(metadata, (connection) => connection.readEntry(id)),
readEntries: (metadata, options) => this.read(metadata, (connection) => connection.readEntries(options)),
findEntriesOnBranch: (metadata, query) =>
this.read(metadata, (connection) => connection.findEntriesOnBranch(query)),
readPathToRootOrCompaction: (metadata, leafId) =>
this.read(metadata, (connection) => connection.readPathToRootOrCompaction(leafId)),
append: (metadata, entry) => this.appendEntry(metadata, entry),
getLabel: (metadata, id) => this.read(metadata, (connection) => connection.getLabel(id)),
getName: (metadata) => this.read(metadata, (connection) => connection.getName()),
getStats: (metadata) => this.read(metadata, (connection) => connection.getStats()),
};
constructor(options: SqliteSessionStoreOptions) {
this.env = options.env;
this.sqlite = options.sqlite;
this.databasePathInput = options.databasePath;
}
create(options: SqliteSessionCreateOptions): Promise<SessionReader<SqliteSessionMetadata>> {
private create(options: SqliteSessionCreateOptions): Promise<SqliteSessionMetadata> {
this.assertOpen();
return this.operations.enqueue(async () => {
const db = await this.getDatabase();
@ -94,16 +117,16 @@ class SqliteSessionStore
);
const metadata = connection.metadata;
this.writers.set(metadata.id, connection);
return this.reader(connection);
return metadata;
});
}
load(metadata: SqliteSessionMetadata): Promise<SessionReader<SqliteSessionMetadata>> {
private open(metadata: SqliteSessionMetadata): Promise<SqliteSessionMetadata> {
this.assertOpen();
return this.operations.enqueue(() => this.loadSession(metadata));
}
private async loadSession(metadata: SqliteSessionMetadata): Promise<SessionReader<SqliteSessionMetadata>> {
private async loadSession(metadata: SqliteSessionMetadata): Promise<SqliteSessionMetadata> {
if (
!getFileSystemResultOrThrow(await this.env.exists(metadata.path), `Failed to check database ${metadata.path}`)
) {
@ -112,10 +135,10 @@ class SqliteSessionStore
const connection =
this.writers.get(metadata.id) ?? (await SqliteSessionConnection.open(await this.getDatabase(), metadata));
this.writers.set(metadata.id, connection);
return this.reader(connection);
return connection.metadata;
}
list(options: SqliteSessionListOptions = {}): Promise<SqliteSessionMetadata[]> {
private list(options: SqliteSessionListOptions = {}): Promise<SqliteSessionMetadata[]> {
this.assertOpen();
return this.operations.enqueue(() => this.listSessions(options));
}
@ -138,7 +161,7 @@ class SqliteSessionStore
return rows.map((row) => rowToMetadata(row, path));
}
appendEntry(metadata: SqliteSessionMetadata, entry: SessionTreeEntry): Promise<void> {
private appendEntry(metadata: SqliteSessionMetadata, entry: SessionTreeEntry): Promise<void> {
this.assertOpen();
return this.operations.enqueue(async () => {
const connection =
@ -148,7 +171,7 @@ class SqliteSessionStore
});
}
delete(metadata: SqliteSessionMetadata): Promise<void> {
private delete(metadata: SqliteSessionMetadata): Promise<void> {
this.assertOpen();
return this.operations.enqueue(async () => {
const db = await this.getDatabase();
@ -166,11 +189,11 @@ class SqliteSessionStore
});
}
fork(
private fork(
source: SqliteSessionMetadata,
options: SqliteSessionCreateOptions,
selection: SessionForkSelection,
): Promise<SessionReader<SqliteSessionMetadata>> {
): Promise<SqliteSessionMetadata> {
this.assertOpen();
return this.operations.enqueue(async () => {
const db = await this.getDatabase();
@ -189,7 +212,7 @@ class SqliteSessionStore
});
const metadata = connection.metadata;
this.writers.set(metadata.id, connection);
return this.reader(connection);
return metadata;
});
}
@ -214,30 +237,17 @@ class SqliteSessionStore
if (this.disposed) throw new SessionError("storage", "SQLite session store is disposed");
}
private reader(connection: SqliteSessionConnection): SessionReader<SqliteSessionMetadata> {
return {
metadata: connection.metadata,
readHead: () => {
this.assertOpen();
return this.operations.enqueue(() => connection.readHead());
},
readEntry: (id) => {
this.assertOpen();
return this.operations.enqueue(() => connection.readEntry(id));
},
readEntries: (options) => {
this.assertOpen();
return this.operations.enqueue(() => connection.readEntries(options));
},
findEntriesOnBranch: (query) => {
this.assertOpen();
return this.operations.enqueue(() => connection.findEntriesOnBranch(query));
},
readPathToRootOrCompaction: (leafId) => {
this.assertOpen();
return this.operations.enqueue(() => connection.readPathToRootOrCompaction(leafId));
},
};
private read<T>(
metadata: SqliteSessionMetadata,
read: (connection: SqliteSessionConnection) => Promise<T>,
): Promise<T> {
this.assertOpen();
return this.operations.enqueue(async () => {
const connection =
this.writers.get(metadata.id) ?? (await SqliteSessionConnection.open(await this.getDatabase(), metadata));
this.writers.set(metadata.id, connection);
return read(connection);
});
}
private async getDatabasePath(): Promise<string> {

17
packages/storage/sqlite-node/src/sqlite/storage/index.ts
View file

@ -1,7 +1,7 @@
import type {
SessionBranchQuery,
SessionEntryCursorOptions,
SessionReader,
SessionStats,
SessionTreeEntry,
} from "@earendil-works/pi-agent-core";
import { SessionError, toError } from "@earendil-works/pi-agent-core";
@ -74,7 +74,7 @@ async function loadSqliteSession(
};
}
export class SqliteSessionConnection implements SessionReader<SqliteSessionMetadata> {
export class SqliteSessionConnection {
private readonly db: SqliteDatabase;
readonly metadata: SqliteSessionMetadata;
private byId: Map<string, SessionTreeEntry>;
@ -332,6 +332,19 @@ export class SqliteSessionConnection implements SessionReader<SqliteSessionMetad
);
}
async getLabel(id: string): Promise<string | undefined> {
return this.materializedState.labelsById.get(id);
}
async getName(): Promise<string | undefined> {
return this.materializedState.name;
}
async getStats(): Promise<SessionStats> {
const { messageCount, cachedTokens, uncachedTokens, totalTokens, costTotal } = this.materializedState;
return { messageCount, cachedTokens, uncachedTokens, totalTokens, costTotal };
}
async readHead(): Promise<{ leafId: string | null }> {
const row = await this.db
.prepare(