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): compose session storage through repositories

Browse source
This commit is contained in:
Christian Klotz 2026-08-02 12:31:27 +03:00
parent a4c7cb025d
commit f07af611d3
No known key found for this signature in database
28 changed files with 549 additions and 556 deletions
Download patch file Download diff file Expand all files Collapse all files

6
packages/agent/CHANGELOG.md
View file

@ -4,13 +4,11 @@
### Breaking Changes
- `SessionRepository.search()` now throws a `SessionError` with code `unsupported` when no search implementation is configured instead of returning no hits.
- Changed `Session` into the sole opened-session aggregate and replaced `SessionRepo` and concrete per-session persistence classes with a non-owning `SessionRepository`, caller-owned `SessionCollection`, and coherent per-session `SessionStorage`. Create collections with `createInMemorySessionCollection()` or `createJsonlSessionCollection()`, compose them with `createSessionRepository({ collection, search: createScanningSessionSearch(collection) })`, and dispose the collection after draining harness and session work.
- `Session` instances are now created by `SessionRepository`; direct construction from an independently supplied collection and snapshot was removed.
- Replaced per-session `SessionReader` objects and repository-level append routing with coherent `SessionStorage` implementations that own both reads and writes.
- Replaced backend-specific `SessionRepo` implementations and separately exposed persistence classes with resource-owning `SessionRepository` implementations backed by coherent per-session `SessionStorage` objects. Use `new InMemorySessionRepository()` or `new JsonlSessionRepository()`, and implement `SessionRepository` directly for custom backends.
### Added
- Added independently composable `SessionSearch`, including scanning search over any `SessionRepository`.
- Added bounded `Session.findEntriesOnBranch()` and `findEntryOnBranch()` queries with explicit traversal, filtering, ordering, and limit options.
## [0.83.0] - 2026-07-29

2
packages/agent/docs/agent-harness.md
View file

@ -87,7 +87,7 @@ The package exports `createReadTool()`, `createWriteTool()`, `createEditTool()`,
The session contains persisted entries only. Session reads return persisted state and do not include queued writes.
`Session.buildContextEntries()` returns the compaction-aware entry sequence used for model context construction. `Session.buildContext()` derives runtime state from the full active branch, then projects those context entries to `AgentMessage[]`. Custom entries are omitted from model context by default; applications can pass `entryProjectors` through `createSessionRepository({ collection, contextBuildOptions })` or `buildContext()` to project selected custom entries into messages. Applications can also pass stacked `entryTransforms`, which run after the default compaction transform, to filter or reorder context entries before projection.
`Session.buildContextEntries()` returns the compaction-aware entry sequence used for model context construction. `Session.buildContext()` derives runtime state from the full active branch, then projects those context entries to `AgentMessage[]`. Custom entries are omitted from model context by default; applications can pass `entryProjectors` through a built-in repository's `contextBuildOptions` or `buildContext()` to project selected custom entries into messages. Applications can also pass stacked `entryTransforms`, which run after the default compaction transform, to filter or reorder context entries before projection.
Session storage implementations must persist leaf changes as `leaf` entries. `setLeafId()` is not an in-memory-only cursor update; it appends a durable entry whose `targetId` is the active tree leaf or `null` for root. Reopening storage must reconstruct the current leaf from the latest persisted leaf-affecting entry.

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

@ -1993,31 +1993,47 @@ Rules, both scopes:
- Persisted config derives from the copied tree via the usual branch point queries; `main` sits at the fork point (`scope: "branch"`) or the source's `main` leaf (`scope: "tree"`).
- **Threads are refs first.** A platform thread sharing one source of truth with its channel is a ref in the same session (section 6), not a fork. Fork when a *separate* session is wanted: subagents, exports, clones. Whether a thread becomes a ref, a fork, or a fresh session with platform backlog as prompt-time context is application policy; all three are supported.
## 14. Session collections
## 14. Session repositories
`Session` is the public opened-session facade. `SessionRepository` borrows a
caller-owned `SessionCollection`, opens complete per-session `SessionStorage`
implementations, and constructs `Session` objects.
The in-memory collection is the simplest built-in option:
`SessionRepository` is the single backend extension contract and owns all shared
storage resources. `Session` is the opened-session facade over one coherent
`SessionStorage`. Built-in repositories require no assembly:
```ts
await using collection = createInMemorySessionCollection();
const search = createScanningSessionSearch(collection);
const repository = createSessionRepository({ collection, search });
await using repository = new InMemorySessionRepository();
const session = await repository.create({});
const search = createScanningSessionSearch(repository);
```
Use `createJsonlSessionCollection({ fs, sessionsRoot })` for filesystem persistence;
the same scanning search composes with it. Omitting `search` is valid, but
`repository.search()` then throws a `SessionError` with code `unsupported`.
SQLite support is provided
separately by `createSqliteSessionCollection()` and should use
`createSqliteSessionSearch(options)` against the same canonical database. Drain
all session or harness operations before the collection leaves
scope; repositories and sessions do not dispose it.
Use `new JsonlSessionRepository({ fs, sessionsRoot })` for filesystem persistence.
Construct `SqliteSessionRepository` separately for SQLite support.
`SessionSearch` is an independent derived capability: use
`createScanningSessionSearch(repository)` or `createSqliteSessionSearch(options)`
directly rather than attaching search to the repository. Drain all session or
harness operations before the repository leaves scope.
A custom backend implements `SessionCollection`; its lifecycle methods return one
coherent `SessionStorage` for each opened session:
A custom backend implements `SessionRepository` directly:
```ts
interface SessionRepository<
TMetadata extends SessionMetadata = SessionMetadata,
TCreateOptions extends SessionCreateOptions = SessionCreateOptions,
TListOptions = void,
> extends AsyncDisposable {
create(options: TCreateOptions): Promise<Session<TMetadata>>;
open(metadata: TMetadata): Promise<Session<TMetadata>>;
list(options?: TListOptions): Promise<TMetadata[]>;
delete(metadata: TMetadata): Promise<void>;
fork(
source: TMetadata,
options: SessionForkOptions & TCreateOptions,
): Promise<Session<TMetadata>>;
}
```
Repository implementations use `createSession(storage, contextBuildOptions)` to
wrap each created or opened storage connection. The complete per-session storage
contract is:
```ts
interface SessionStorage<TMetadata extends SessionMetadata = SessionMetadata> {
@ -2034,56 +2050,27 @@ interface SessionStorage<TMetadata extends SessionMetadata = SessionMetadata> {
getName(): Promise<string | undefined>;
getStats(): Promise<SessionStats>;
}
type SessionForkSelection =
| { kind: "all" }
| { kind: "before_user_message"; entryId: string }
| { kind: "through_entry"; entryId: string };
interface SessionCollection<
TMetadata extends SessionMetadata = SessionMetadata,
TCreateOptions extends SessionCreateOptions = SessionCreateOptions,
TListOptions = void,
> extends AsyncDisposable {
create(options: TCreateOptions): Promise<SessionStorage<TMetadata>>;
open(metadata: TMetadata): Promise<SessionStorage<TMetadata>>;
list(options?: TListOptions): Promise<TMetadata[]>;
delete(metadata: TMetadata): Promise<void>;
fork(
source: TMetadata,
options: TCreateOptions,
selection: SessionForkSelection,
): Promise<SessionStorage<TMetadata>>;
}
```
Storage lifetimes belong to the collection. `SessionStorage.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 collection, allowing backend-native copies without repository-level
history materialization.
Disposal rejects new operations and storage calls, drains accepted writes, and
releases owned resources. Collection implementations know nothing about harness
operations, queues, or recovery.
`SessionStorage.appendEntry()` receives a complete entry and must preserve append
order and entry-id uniqueness. `createSessionForkSelection()` and
`readSessionEntriesForFork()` provide shared fork selection and validation while
allowing repository-native copies without generic history materialization.
Repositories own storage lifetimes: disposal rejects new operations and storage
calls, drains accepted writes, and releases shared resources.
`Session.buildContext()` reads only the active path. Built-in Memory and JSONL
storage maintain eager in-memory indexes; SQLite serves branch and projection
queries from native indexes and materialized state.
Custom collections return complete `SessionStorage` implementations. Memory and
JSONL share a private array-backed index; SQLite implements the same contract
with native queries and materialized projections.
Expose custom implementations through a factory returning `SessionCollection` rather
than exporting the concrete class.
Contract, all repositories:
Contract, all collections:
- 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.
- Storage 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 collection: 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.
- One writer per *session*, enforced by the serving layer; SQLite additionally rejects concurrent writers itself. This is per session, not per repository: 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.
### JSONL
@ -2371,10 +2358,10 @@ Current implementation (what is being replaced or wrapped):
5. `packages/agent/src/agent-loop.ts` — monolithic loop to split into step primitives.
6. `packages/agent/src/agent.ts` — stateful wrapper: queues, continuation, abort, settlement.
7. `packages/agent/src/harness/agent-harness.ts` — the harness this design replaces.
8. `packages/agent/src/harness/types.ts` — entry union, session collection contract, event/hook types.
8. `packages/agent/src/harness/types.ts` — entry union, session storage contract, event/hook types.
9. `packages/agent/src/harness/session/session.ts` — Session, context build, entry creation.
10. `packages/agent/src/harness/session/jsonl-collection.ts` — JSONL session collection and v3 document codec.
11. `packages/agent/src/harness/session/memory-collection.ts` — in-memory session collection.
10. `packages/agent/src/harness/session/jsonl-repo.ts` — JSONL session repository and v3 document codec.
11. `packages/agent/src/harness/session/memory-repo.ts` — in-memory session repository.
12. `packages/agent/src/harness/messages.ts` — defaultConvertToLlm and message helpers.
12a. `packages/ai/src/utils/transform-messages.ts` — orphaned-tool-call healing; the adjacency backstop referenced in sections 5 and 13.
13. `packages/agent/src/harness/compaction/compaction.ts` — preparation, split-turn generation, retry.
@ -2388,7 +2375,7 @@ SQLite backend:
18. `packages/storage/sqlite-node/src/sqlite/storage/branch-entries.ts` — active-branch materialization.
19. `packages/storage/sqlite-node/src/sqlite/storage/session-materialized.ts` — stats/labels/config projections.
20. `packages/storage/sqlite-node/src/sqlite/migrations/001_initial.sql` and `migrations.ts` — schema and migration mechanism.
21. `packages/storage/sqlite-node/src/sqlite/session-collection.ts` — create/open/fork.
21. `packages/storage/sqlite-node/src/sqlite/repo.ts` — create/open/fork.
Behavioral tests (compatibility requirements):
@ -2396,5 +2383,5 @@ Behavioral tests (compatibility requirements):
23. `packages/agent/test/agent.test.ts`
24. `packages/agent/test/harness/agent-harness.test.ts`
25. `packages/agent/test/harness/session.test.ts`
26. `packages/agent/test/harness/session-collections.test.ts`
26. `packages/agent/test/harness/session-backends.test.ts`
27. `packages/agent/test/harness/sqlite-migrations.test.ts`

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

@ -1,32 +0,0 @@
import type { SessionForkOptions, SessionForkSelection, SessionTreeEntry } from "../types.ts";
import { SessionError } from "../types.ts";
type MaybePromise<T> = T | Promise<T>;
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"
? { kind: "through_entry", entryId: options.entryId }
: { kind: "before_user_message", entryId: options.entryId };
}
/** @internal Shared fork selection validation for built-in collections. */
export async function readSessionEntriesForFork(
source: SessionForkEntrySource,
selection: SessionForkSelection,
): Promise<readonly SessionTreeEntry[]> {
if (selection.kind === "all") return source.readEntries();
const target = await source.readEntry(selection.entryId);
if (!target) throw new SessionError("invalid_fork_target", `Entry ${selection.entryId} not found`);
if (selection.kind === "through_entry") return source.readPathToRootOrCompaction(target.id);
if (target.type !== "message" || target.message.role !== "user") {
throw new SessionError("invalid_fork_target", `Entry ${selection.entryId} is not a user message`);
}
return source.readPathToRootOrCompaction(target.parentId);
}

79
packages/agent/src/harness/session/jsonl-collection.ts → packages/agent/src/harness/session/jsonl-repo.ts
View file

@ -3,24 +3,31 @@ import type {
JsonlSessionCreateOptions,
JsonlSessionListOptions,
JsonlSessionMetadata,
SessionCollection,
SessionForkOptions,
SessionForkSelection,
SessionStorage,
SessionTreeEntry,
} from "../types.ts";
import { SessionError, toError } from "../types.ts";
import { ArraySessionIndex } from "./array-session-index.ts";
import { readSessionEntriesForFork } from "./fork.ts";
import { KeyedOperationQueue } from "./keyed-operation-queue.ts";
import { createSessionId, createTimestamp, getFileSystemResultOrThrow } from "./repository.ts";
import {
createSessionForkSelection,
createSessionId,
createTimestamp,
getFileSystemResultOrThrow,
readSessionEntriesForFork,
type SessionRepository,
} from "./repository.ts";
import { createSession, type Session, type SessionContextBuildOptions } from "./session.ts";
export interface JsonlSessionCollectionOptions {
fs: JsonlSessionCollectionFileSystem;
export interface JsonlSessionBackendOptions {
fs: JsonlSessionRepositoryFileSystem;
sessionsRoot: string;
/** Maximum active operations across session keys. Defaults to 4. */
maxConcurrentOperations?: number;
}
export type JsonlSessionCollectionFileSystem = Pick<
export type JsonlSessionRepositoryFileSystem = Pick<
FileSystem,
| "absolutePath"
| "joinPath"
@ -199,10 +206,8 @@ function createDocumentDescriptor(options: JsonlSessionCreateOptions): SessionDo
};
}
class JsonlSessionCollection
implements SessionCollection<JsonlSessionMetadata, JsonlSessionCreateOptions, JsonlSessionListOptions>
{
private readonly fs: JsonlSessionCollectionFileSystem;
export class JsonlSessionBackend {
private readonly fs: JsonlSessionRepositoryFileSystem;
private readonly sessionsRootInput: string;
private sessionsRoot: string | undefined;
private readonly entryIndexesByPath = new Map<string, ArraySessionIndex>();
@ -211,7 +216,7 @@ class JsonlSessionCollection
private disposed = false;
private disposePromise: Promise<void> | undefined;
constructor(options: JsonlSessionCollectionOptions) {
constructor(options: JsonlSessionBackendOptions) {
this.fs = options.fs;
this.sessionsRootInput = options.sessionsRoot;
this.operations = new KeyedOperationQueue({
@ -340,7 +345,7 @@ class JsonlSessionCollection
}
private assertOpen(): void {
if (this.disposed) throw new SessionError("storage", "JSONL session collection is disposed");
if (this.disposed) throw new SessionError("storage", "JSONL session repository is disposed");
}
private operationKey(metadata: JsonlSessionMetadata): string {
@ -433,8 +438,50 @@ class JsonlSessionCollection
}
}
export function createJsonlSessionCollection(
options: JsonlSessionCollectionOptions,
): SessionCollection<JsonlSessionMetadata, JsonlSessionCreateOptions, JsonlSessionListOptions> {
return new JsonlSessionCollection(options);
export interface JsonlSessionRepositoryOptions extends JsonlSessionBackendOptions {
contextBuildOptions?: SessionContextBuildOptions;
}
export class JsonlSessionRepository
implements SessionRepository<JsonlSessionMetadata, JsonlSessionCreateOptions, JsonlSessionListOptions>
{
private readonly backend: JsonlSessionBackend;
private readonly contextBuildOptions: SessionContextBuildOptions;
constructor(options: JsonlSessionRepositoryOptions) {
const { contextBuildOptions, ...backendOptions } = options;
this.backend = new JsonlSessionBackend(backendOptions);
this.contextBuildOptions = contextBuildOptions ?? {};
}
async create(options: JsonlSessionCreateOptions): Promise<Session<JsonlSessionMetadata>> {
return createSession(await this.backend.create(options), this.contextBuildOptions);
}
async open(metadata: JsonlSessionMetadata): Promise<Session<JsonlSessionMetadata>> {
return createSession(await this.backend.open(metadata), this.contextBuildOptions);
}
async list(options?: JsonlSessionListOptions): Promise<JsonlSessionMetadata[]> {
return await this.backend.list(options);
}
async delete(metadata: JsonlSessionMetadata): Promise<void> {
await this.backend.delete(metadata);
}
async fork(
source: JsonlSessionMetadata,
options: SessionForkOptions & JsonlSessionCreateOptions,
): Promise<Session<JsonlSessionMetadata>> {
const { entryId: _entryId, position: _position, ...createOptions } = options;
return createSession(
await this.backend.fork(source, createOptions, createSessionForkSelection(options)),
this.contextBuildOptions,
);
}
async [Symbol.asyncDispose](): Promise<void> {
await this.backend[Symbol.asyncDispose]();
}
}

66
packages/agent/src/harness/session/memory-collection.ts → packages/agent/src/harness/session/memory-repo.ts
View file

@ -1,5 +1,5 @@
import type {
SessionCollection,
SessionForkOptions,
SessionForkSelection,
SessionMetadata,
SessionStorage,
@ -7,9 +7,15 @@ import type {
} from "../types.ts";
import { SessionError } from "../types.ts";
import { ArraySessionIndex } from "./array-session-index.ts";
import { readSessionEntriesForFork } from "./fork.ts";
import { KeyedOperationQueue } from "./keyed-operation-queue.ts";
import { createSessionId, createTimestamp } from "./repository.ts";
import {
createSessionForkSelection,
createSessionId,
createTimestamp,
readSessionEntriesForFork,
type SessionRepository,
} from "./repository.ts";
import { createSession, type Session, type SessionContextBuildOptions } from "./session.ts";
export type InMemorySessionCreateOptions = { id?: string };
@ -18,7 +24,7 @@ interface InMemorySessionState {
entries: ArraySessionIndex;
}
class InMemorySessionCollection implements SessionCollection<SessionMetadata, InMemorySessionCreateOptions, void> {
export class InMemorySessionBackend {
private readonly sessions = new Map<string, InMemorySessionState>();
private readonly operations = new KeyedOperationQueue<string>();
private disposed = false;
@ -109,7 +115,7 @@ class InMemorySessionCollection implements SessionCollection<SessionMetadata, In
}
private assertOpen(): void {
if (this.disposed) throw new SessionError("storage", "In-memory session collection is disposed");
if (this.disposed) throw new SessionError("storage", "In-memory session repository is disposed");
}
private getState(metadata: SessionMetadata): InMemorySessionState {
@ -119,10 +125,48 @@ class InMemorySessionCollection implements SessionCollection<SessionMetadata, In
}
}
export function createInMemorySessionCollection(): SessionCollection<
SessionMetadata,
InMemorySessionCreateOptions,
void
> {
return new InMemorySessionCollection();
export interface InMemorySessionRepositoryOptions {
contextBuildOptions?: SessionContextBuildOptions;
}
export class InMemorySessionRepository
implements SessionRepository<SessionMetadata, InMemorySessionCreateOptions, void>
{
private readonly backend = new InMemorySessionBackend();
private readonly contextBuildOptions: SessionContextBuildOptions;
constructor(options: InMemorySessionRepositoryOptions = {}) {
this.contextBuildOptions = options.contextBuildOptions ?? {};
}
async create(options: InMemorySessionCreateOptions = {}): Promise<Session<SessionMetadata>> {
return createSession(await this.backend.create(options), this.contextBuildOptions);
}
async open(metadata: SessionMetadata): Promise<Session<SessionMetadata>> {
return createSession(await this.backend.open(metadata), this.contextBuildOptions);
}
async list(): Promise<SessionMetadata[]> {
return await this.backend.list();
}
async delete(metadata: SessionMetadata): Promise<void> {
await this.backend.delete(metadata);
}
async fork(
source: SessionMetadata,
options: SessionForkOptions & InMemorySessionCreateOptions,
): Promise<Session<SessionMetadata>> {
const { entryId: _entryId, position: _position, ...createOptions } = options;
return createSession(
await this.backend.fork(source, createOptions, createSessionForkSelection(options)),
this.contextBuildOptions,
);
}
async [Symbol.asyncDispose](): Promise<void> {
await this.backend[Symbol.asyncDispose]();
}
}

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

@ -2,16 +2,14 @@ import { uuidv7 } from "@earendil-works/pi-ai";
import {
type FileError,
type Result,
type SessionCollection,
type SessionCreateOptions,
SessionError,
type SessionForkOptions,
type SessionForkSelection,
type SessionMetadata,
type SessionSearch,
type SessionSearchHit,
type SessionTreeEntry,
} from "../types.ts";
import { createSessionForkSelection } from "./fork.ts";
import { createSession, type Session, type SessionContextBuildOptions } from "./session.ts";
import type { Session } from "./session.ts";
export function createSessionId(): string {
return uuidv7();
@ -21,65 +19,16 @@ export function createTimestamp(): string {
return new Date().toISOString();
}
export class SessionRepository<
export interface SessionRepository<
TMetadata extends SessionMetadata = SessionMetadata,
TCreateOptions extends SessionCreateOptions = SessionCreateOptions,
TListOptions = void,
> {
private readonly collection: SessionCollection<TMetadata, TCreateOptions, TListOptions>;
private readonly sessionSearch: SessionSearch<TMetadata> | undefined;
private readonly contextBuildOptions: SessionContextBuildOptions;
constructor(options: {
collection: SessionCollection<TMetadata, TCreateOptions, TListOptions>;
search?: SessionSearch<TMetadata>;
contextBuildOptions?: SessionContextBuildOptions;
}) {
this.collection = options.collection;
this.sessionSearch = options.search;
this.contextBuildOptions = options.contextBuildOptions ?? {};
}
async create(options: TCreateOptions): Promise<Session<TMetadata>> {
return createSession(await this.collection.create(options), this.contextBuildOptions);
}
async open(metadata: TMetadata): Promise<Session<TMetadata>> {
return createSession(await this.collection.open(metadata), this.contextBuildOptions);
}
async list(options?: TListOptions): Promise<TMetadata[]> {
return await this.collection.list(options);
}
async delete(metadata: TMetadata): Promise<void> {
await this.collection.delete(metadata);
}
async fork(source: TMetadata, options: SessionForkOptions & TCreateOptions): Promise<Session<TMetadata>> {
const selection = createSessionForkSelection(options);
const createOptions = { ...options };
delete createOptions.entryId;
delete createOptions.position;
return createSession(await this.collection.fork(source, createOptions, selection), this.contextBuildOptions);
}
async search(options: Parameters<SessionSearch<TMetadata>["search"]>[0]): Promise<SessionSearchHit<TMetadata>[]> {
if (!this.sessionSearch) throw new SessionError("unsupported", "Session search is not configured");
return await this.sessionSearch.search(options);
}
}
export function createSessionRepository<
TMetadata extends SessionMetadata = SessionMetadata,
TCreateOptions extends SessionCreateOptions = SessionCreateOptions,
TListOptions = void,
>(options: {
collection: SessionCollection<TMetadata, TCreateOptions, TListOptions>;
search?: SessionSearch<TMetadata>;
contextBuildOptions?: SessionContextBuildOptions;
}): SessionRepository<TMetadata, TCreateOptions, TListOptions> {
return new SessionRepository(options);
> extends AsyncDisposable {
create(options: TCreateOptions): Promise<Session<TMetadata>>;
open(metadata: TMetadata): Promise<Session<TMetadata>>;
list(options?: TListOptions): Promise<TMetadata[]>;
delete(metadata: TMetadata): Promise<void>;
fork(source: TMetadata, options: SessionForkOptions & TCreateOptions): Promise<Session<TMetadata>>;
}
export function getFileSystemResultOrThrow<TValue>(result: Result<TValue, FileError>, message: string): TValue {
@ -89,3 +38,34 @@ export function getFileSystemResultOrThrow<TValue>(result: Result<TValue, FileEr
}
return result.value;
}
type MaybePromise<T> = T | Promise<T>;
interface SessionForkEntrySource {
readEntry(id: string): MaybePromise<SessionTreeEntry | undefined>;
readEntries(): MaybePromise<readonly SessionTreeEntry[]>;
readPathToRootOrCompaction(leafId: string | null): MaybePromise<readonly SessionTreeEntry[]>;
}
/** @internal Transitional fork selection shared by built-in repositories. */
export function createSessionForkSelection(options: SessionForkOptions): SessionForkSelection {
if (!options.entryId) return { kind: "all" };
return (options.position ?? "before") === "at"
? { kind: "through_entry", entryId: options.entryId }
: { kind: "before_user_message", entryId: options.entryId };
}
/** @internal Shared fork selection validation for built-in repositories. */
export async function readSessionEntriesForFork(
source: SessionForkEntrySource,
selection: SessionForkSelection,
): Promise<readonly SessionTreeEntry[]> {
if (selection.kind === "all") return source.readEntries();
const target = await source.readEntry(selection.entryId);
if (!target) throw new SessionError("invalid_fork_target", `Entry ${selection.entryId} not found`);
if (selection.kind === "through_entry") return source.readPathToRootOrCompaction(target.id);
if (target.type !== "message" || target.message.role !== "user") {
throw new SessionError("invalid_fork_target", `Entry ${selection.entryId} is not a user message`);
}
return source.readPathToRootOrCompaction(target.parentId);
}

31
packages/agent/src/harness/session/scanning-session-search.ts → packages/agent/src/harness/session/search.ts
View file

@ -1,21 +1,23 @@
import type {
SessionCreateOptions,
SessionMetadata,
SessionSearch,
SessionSearchHit,
SessionSearchOptions,
SessionStorage,
} from "../types.ts";
import type { SessionRepository } from "./repository.ts";
import type { Session } from "./session.ts";
type SessionSearchSource<TMetadata extends SessionMetadata> = {
open(metadata: TMetadata): Promise<SessionStorage<TMetadata>>;
type ScanningSessionSearchSource<TMetadata extends SessionMetadata> = {
list(): Promise<TMetadata[]>;
open(metadata: TMetadata): Promise<Session<TMetadata>>;
};
/** Searches canonical sessions directly and therefore has no index to maintain. */
class ScanningSessionSearch<TMetadata extends SessionMetadata = SessionMetadata> implements SessionSearch<TMetadata> {
private readonly source: SessionSearchSource<TMetadata>;
private readonly source: ScanningSessionSearchSource<TMetadata>;
constructor(source: SessionSearchSource<TMetadata>) {
constructor(source: ScanningSessionSearchSource<TMetadata>) {
this.source = source;
}
@ -26,24 +28,21 @@ class ScanningSessionSearch<TMetadata extends SessionMetadata = SessionMetadata>
for (const metadata of await this.source.list()) {
const cwd = (metadata as { cwd?: unknown }).cwd;
if (options.cwd !== undefined && cwd !== options.cwd) continue;
const storage = await this.source.open(metadata);
for (const entry of await storage.readEntries()) {
const session = await this.source.open(metadata);
for (const entry of await session.getEntries()) {
const payload = JSON.stringify(entry);
if (!payload.toLowerCase().includes(normalizedText)) continue;
hits.push({
metadata: storage.metadata,
entryId: entry.id,
timestamp: entry.timestamp,
snippet: payload,
});
hits.push({ metadata, entryId: entry.id, timestamp: entry.timestamp, snippet: payload });
}
}
return hits;
}
}
export function createScanningSessionSearch<TMetadata extends SessionMetadata>(
source: SessionSearchSource<TMetadata>,
): SessionSearch<TMetadata> {
export function createScanningSessionSearch<
TMetadata extends SessionMetadata,
TCreateOptions extends SessionCreateOptions,
TListOptions,
>(source: Pick<SessionRepository<TMetadata, TCreateOptions, TListOptions>, "list" | "open">): SessionSearch<TMetadata> {
return new ScanningSessionSearch(source);
}

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

@ -421,7 +421,7 @@ export class Session<TMetadata extends SessionMetadata = SessionMetadata> {
}
}
/** @internal Construct sessions only through SessionRepository. */
/** Wraps an opened storage connection for a SessionRepository implementation. */
export async function createSession<TMetadata extends SessionMetadata>(
storage: SessionStorage<TMetadata>,
contextBuildOptions: SessionContextBuildOptions = {},

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

@ -225,7 +225,6 @@ export class BranchSummaryError extends Error {
export type SessionErrorCode =
| "not_found"
| "unsupported"
| "invalid_session"
| "invalid_entry"
| "invalid_fork_target"
@ -556,7 +555,7 @@ export interface SessionHead {
leafId: string | null;
}
/** Complete storage contract for one opened session. Its lifetime is owned by its {@link SessionCollection}. */
/** Complete storage contract for one opened session. Its lifetime is owned by its repository. */
export interface SessionStorage<TMetadata extends SessionMetadata = SessionMetadata> {
readonly metadata: TMetadata;
/** Rejects with `invalid_session` when a non-null active leaf does not reference a stored entry. */
@ -571,23 +570,6 @@ export interface SessionStorage<TMetadata extends SessionMetadata = SessionMetad
getStats(): Promise<SessionStats>;
}
/** Owns the collection lifecycle and resources shared by sessions in a repository. */
export interface SessionCollection<
TMetadata extends SessionMetadata = SessionMetadata,
TCreateOptions extends SessionCreateOptions = SessionCreateOptions,
TListOptions = void,
> extends AsyncDisposable {
create(options: TCreateOptions): Promise<SessionStorage<TMetadata>>;
open(metadata: TMetadata): Promise<SessionStorage<TMetadata>>;
list(options?: TListOptions): Promise<TMetadata[]>;
delete(metadata: TMetadata): Promise<void>;
fork(
source: TMetadata,
options: TCreateOptions,
selection: SessionForkSelection,
): Promise<SessionStorage<TMetadata>>;
}
export interface JsonlSessionCreateOptions extends SessionCreateOptions {
cwd: string;
parentSessionPath?: string;

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

@ -29,23 +29,25 @@ export {
} from "./harness/compaction/compaction.ts";
export * from "./harness/messages.ts";
export * from "./harness/prompt-templates.ts";
export { readSessionEntriesForFork } from "./harness/session/fork.ts";
export {
createJsonlSessionCollection,
type JsonlSessionCollectionFileSystem,
type JsonlSessionCollectionOptions,
} from "./harness/session/jsonl-collection.ts";
JsonlSessionRepository,
type JsonlSessionRepositoryFileSystem,
type JsonlSessionRepositoryOptions,
loadJsonlSessionMetadata,
} from "./harness/session/jsonl-repo.ts";
export {
createInMemorySessionCollection,
type InMemorySessionCreateOptions,
} from "./harness/session/memory-collection.ts";
InMemorySessionRepository,
type InMemorySessionRepositoryOptions,
} from "./harness/session/memory-repo.ts";
export * from "./harness/session/repository.ts";
export * from "./harness/session/scanning-session-search.ts";
export * from "./harness/session/search.ts";
export {
buildContextEntries,
buildSessionContext,
type ContextEntryTransform,
type CustomEntryContextMessageProjector,
createSession,
defaultContextEntryTransform,
type SessionContextBuildOptions,
sessionEntryToContextMessages,

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

@ -11,9 +11,8 @@ import { getModel } from "@earendil-works/pi-ai/compat";
import { describe, expect, it } from "vitest";
import { AgentHarness } from "../../src/harness/agent-harness.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { createInMemorySessionCollection } from "../../src/harness/session/memory-collection.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import type { Session } from "../../src/harness/session/session.ts";
import { InMemorySessionBackend } from "../../src/harness/session/memory-repo.ts";
import { createSession, type Session } from "../../src/harness/session/session.ts";
import type { AgentHarnessTool, PromptTemplate, Skill } from "../../src/harness/types.ts";
import type { AgentMessage, AgentTool } from "../../src/types.ts";
import { calculateTool, createCalculateToolWithUsage } from "../utils/calculate.ts";
@ -97,7 +96,7 @@ async function createBlockingSession(expectedWrites: number): Promise<{
releaseWrites: ReturnType<typeof deferred>;
getEntries(): ReturnType<Session["getEntries"]>;
}> {
const source = createInMemorySessionCollection();
const source = new InMemorySessionBackend();
const allWritesStarted = deferred();
const releaseWrites = deferred();
let writesStarted = 0;
@ -110,15 +109,18 @@ async function createBlockingSession(expectedWrites: number): Promise<{
await storage.appendEntry(entry);
},
});
const blockingCollection: ReturnType<typeof createInMemorySessionCollection> = {
const blockingBackend: Pick<
InMemorySessionBackend,
"create" | "open" | "list" | "delete" | "fork" | typeof Symbol.asyncDispose
> = {
create: async (options) => blockWrites(await source.create(options)),
open: async (metadata) => blockWrites(await source.open(metadata)),
list: (options) => source.list(options),
list: () => source.list(),
delete: (metadata) => source.delete(metadata),
fork: async (metadata, options, selection) => blockWrites(await source.fork(metadata, options, selection)),
[Symbol.asyncDispose]: () => source[Symbol.asyncDispose](),
};
const session = await createSessionRepository({ collection: blockingCollection }).create({});
const session = await createSession(await blockingBackend.create({}));
return {
session,
allWritesStarted,

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

@ -1,17 +1,16 @@
import { join } from "node:path";
import { afterEach, describe, expect, it } from "vitest";
import { createNodeSqliteFactory, createSqliteSessionCollection } from "../../../storage/sqlite-node/src/index.ts";
import { createNodeSqliteFactory, SqliteSessionRepository } from "../../../storage/sqlite-node/src/index.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { createJsonlSessionCollection } from "../../src/harness/session/jsonl-collection.ts";
import { createInMemorySessionCollection } from "../../src/harness/session/memory-collection.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import { JsonlSessionRepository } from "../../src/harness/session/jsonl-repo.ts";
import { InMemorySessionBackend, InMemorySessionRepository } from "../../src/harness/session/memory-repo.ts";
import type { Session } from "../../src/harness/session/session.ts";
import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts";
const ownedCollections: AsyncDisposable[] = [];
const ownedRepositories: AsyncDisposable[] = [];
afterEach(async () => {
for (const collection of ownedCollections.splice(0)) await collection[Symbol.asyncDispose]();
for (const repository of ownedRepositories.splice(0)) await repository[Symbol.asyncDispose]();
});
async function verifyBranchQueries(session: Session): Promise<{ tail: string; fullPath: string[] }> {
@ -90,9 +89,8 @@ async function verifyBranchQueries(session: Session): Promise<{ tail: string; fu
describe("bounded session branch queries", () => {
it("provides identical in-memory query semantics", async () => {
const collection = createInMemorySessionCollection();
ownedCollections.push(collection);
const repo = createSessionRepository({ collection });
const repo = new InMemorySessionRepository();
ownedRepositories.push(repo);
const session = await repo.create({ id: "memory" });
const expected = await verifyBranchQueries(session);
const reopened = await repo.open(await session.getMetadata());
@ -102,9 +100,9 @@ describe("bounded session branch queries", () => {
});
it("rejects corrupt parent chains in array-backed readers", async () => {
const collection = createInMemorySessionCollection();
ownedCollections.push(collection);
const storage = await collection.create({ id: "corrupt-memory" });
const backend = new InMemorySessionBackend();
ownedRepositories.push(backend);
const storage = await backend.create({ id: "corrupt-memory" });
await storage.appendEntry({
type: "message",
id: "orphan",
@ -145,9 +143,8 @@ describe("bounded session branch queries", () => {
it("provides identical JSONL query semantics", async () => {
const root = createTempDir();
const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
ownedCollections.push(collection);
const repo = createSessionRepository({ collection });
const repo = new JsonlSessionRepository({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
ownedRepositories.push(repo);
const session = await repo.create({ id: "jsonl", cwd: root });
const expected = await verifyBranchQueries(session);
const reopened = await repo.open(await session.getMetadata());
@ -160,13 +157,12 @@ describe("bounded session branch queries", () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({
const repo = new SqliteSessionRepository({
env: new NodeExecutionEnv({ cwd: root }),
sqlite,
databasePath,
});
ownedCollections.push(collection);
const repo = createSessionRepository({ collection });
ownedRepositories.push(repo);
const session = await repo.create({ id: "bounded-sqlite", cwd: root });
const rootId = await session.appendMessage(createUserMessage("root"));
const middleId = await session.appendMessage(createAssistantMessage("middle"));
@ -226,13 +222,12 @@ describe("bounded session branch queries", () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({
const repo = new SqliteSessionRepository({
env: new NodeExecutionEnv({ cwd: root }),
sqlite,
databasePath,
});
ownedCollections.push(collection);
const repo = createSessionRepository({ collection });
ownedRepositories.push(repo);
const session = await repo.create({ id: "invalid-filtered-sqlite", cwd: root });
await session.appendMessage(createUserMessage("root"));
const customId = await session.appendCustomEntry("note", { value: 1 });
@ -269,13 +264,12 @@ describe("bounded session branch queries", () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({
const repo = new SqliteSessionRepository({
env: new NodeExecutionEnv({ cwd: root }),
sqlite,
databasePath,
});
ownedCollections.push(collection);
const repo = createSessionRepository({ collection });
ownedRepositories.push(repo);
const session = await repo.create({ id: "bounded-corrupt-sqlite", cwd: root });
const rootId = await session.appendMessage(createUserMessage("root"));
const childId = await session.appendMessage(createAssistantMessage("child"));
@ -324,13 +318,12 @@ describe("bounded session branch queries", () => {
it("provides identical SQLite query semantics", async () => {
const root = createTempDir();
const collection = createSqliteSessionCollection({
const repo = new SqliteSessionRepository({
env: new NodeExecutionEnv({ cwd: root }),
sqlite: createNodeSqliteFactory(),
databasePath: join(root, "sessions.sqlite"),
});
ownedCollections.push(collection);
const repo = createSessionRepository({ collection });
ownedRepositories.push(repo);
const session = await repo.create({ id: "sqlite", cwd: root });
const expected = await verifyBranchQueries(session);
const reopened = await repo.open(await session.getMetadata());

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

@ -1,18 +1,10 @@
import { existsSync, writeFileSync } from "node:fs";
import { afterEach, describe, expect, it, vi } from "vitest";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { createJsonlSessionCollection } from "../../src/harness/session/jsonl-collection.ts";
import {
createInMemorySessionCollection,
type InMemorySessionCreateOptions,
} from "../../src/harness/session/memory-collection.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import type {
SessionCollection,
SessionForkSelection,
SessionMetadata,
SessionStorage,
} from "../../src/harness/types.ts";
import { JsonlSessionBackend, JsonlSessionRepository } from "../../src/harness/session/jsonl-repo.ts";
import { InMemorySessionBackend, InMemorySessionRepository } from "../../src/harness/session/memory-repo.ts";
import { createSession } from "../../src/harness/session/session.ts";
import type { SessionForkSelection, SessionMetadata, SessionStorage } from "../../src/harness/types.ts";
import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts";
afterEach(() => {
@ -139,7 +131,7 @@ class BlockingAppendEnv extends NodeExecutionEnv {
}
}
interface SessionCollectionReadCounter {
interface SessionBackendReadCounter {
openCount: number;
readHeadCount: number;
readEntriesCount: number;
@ -147,12 +139,12 @@ interface SessionCollectionReadCounter {
forkSelections: SessionForkSelection[];
}
function createCountingInMemorySessionCollection(): {
collection: SessionCollection<SessionMetadata, InMemorySessionCreateOptions, void>;
counter: SessionCollectionReadCounter;
function createCountingInMemorySessionBackend(): {
backend: Pick<InMemorySessionBackend, "create" | "open" | "list" | "delete" | "fork" | typeof Symbol.asyncDispose>;
counter: SessionBackendReadCounter;
} {
const source = createInMemorySessionCollection();
const counter: SessionCollectionReadCounter = {
const source = new InMemorySessionBackend();
const counter: SessionBackendReadCounter = {
openCount: 0,
readHeadCount: 0,
readEntriesCount: 0,
@ -182,13 +174,13 @@ function createCountingInMemorySessionCollection(): {
});
return {
counter,
collection: {
backend: {
create: (options) => source.create(options),
async open(metadata) {
counter.openCount += 1;
return countReads(await source.open(metadata));
},
list: (options) => source.list(options),
list: () => source.list(),
delete: (metadata) => source.delete(metadata),
fork: (metadata, options, selection) => {
counter.forkSelections.push(selection);
@ -199,18 +191,9 @@ function createCountingInMemorySessionCollection(): {
};
}
describe("InMemorySessionCollection", () => {
it("rejects search when no search implementation is configured", async () => {
const repo = createSessionRepository({ collection: createInMemorySessionCollection() });
await expect(repo.search({ text: "needle" })).rejects.toMatchObject({
code: "unsupported",
message: "Session search is not configured",
});
});
describe("InMemorySessionBackend", () => {
it("opens, deletes, and forks by metadata", async () => {
const repo = createSessionRepository({ collection: createInMemorySessionCollection() });
const repo = new InMemorySessionRepository();
const session = await repo.create({ id: "session-1" });
const metadata = await session.getMetadata();
const user1 = await session.appendMessage(createUserMessage("one"));
@ -239,12 +222,13 @@ describe("InMemorySessionCollection", () => {
});
it("delegates full-session fork selection without opening the source", async () => {
const { collection, counter } = createCountingInMemorySessionCollection();
const repo = createSessionRepository({ collection });
const source = await repo.create({ id: "session-1" });
const { backend, counter } = createCountingInMemorySessionBackend();
const source = await createSession(await backend.create({ id: "session-1" }));
await source.appendMessage(createUserMessage("one"));
const fork = await repo.fork(await source.getMetadata(), { id: "session-2" });
const fork = await createSession(
await backend.fork(await source.getMetadata(), { id: "session-2" }, { kind: "all" }),
);
expect((await fork.getEntries()).map((entry) => entry.id)).toHaveLength(1);
expect(counter).toMatchObject({
@ -257,9 +241,8 @@ describe("InMemorySessionCollection", () => {
});
it("retains the opened aggregate instead of reloading for scoped reads", async () => {
const { collection, counter } = createCountingInMemorySessionCollection();
const repo = createSessionRepository({ collection });
const session = await repo.create({ id: "session-1" });
const { backend, counter } = createCountingInMemorySessionBackend();
const session = await createSession(await backend.create({ id: "session-1" }));
const entryId = await session.appendMessage(createUserMessage("one"));
await session.getMetadata();
@ -269,43 +252,40 @@ describe("InMemorySessionCollection", () => {
});
it("builds context from the branch storage without loading complete history", async () => {
const { collection, counter } = createCountingInMemorySessionCollection();
const repo = createSessionRepository({ collection });
const created = await repo.create({ id: "session-1" });
const { backend, counter } = createCountingInMemorySessionBackend();
const created = await createSession(await backend.create({ id: "session-1" }));
await created.appendMessage(createUserMessage("one"));
const opened = await repo.open(await created.getMetadata());
const opened = await createSession(await backend.open(await created.getMetadata()));
await opened.buildContext();
expect(counter).toMatchObject({ openCount: 1, readHeadCount: 1, readEntriesCount: 0, readPathCount: 1 });
});
it("rejects repository operations and session writes after collection disposal", async () => {
const collection = createInMemorySessionCollection();
const repo = createSessionRepository({ collection });
it("rejects repository operations and session writes after disposal", async () => {
const repo = new InMemorySessionRepository();
const session = await repo.create({ id: "session-1" });
await collection[Symbol.asyncDispose]();
await repo[Symbol.asyncDispose]();
await expect(repo.list()).rejects.toThrow("In-memory session collection is disposed");
await expect(repo.list()).rejects.toThrow("In-memory session repository is disposed");
await expect(session.appendMessage(createUserMessage("late"))).rejects.toThrow(
"In-memory session collection is disposed",
"In-memory session repository is disposed",
);
});
it("supports lexical ownership with await using", async () => {
let listDisposedCollection: (() => Promise<SessionMetadata[]>) | undefined;
let listDisposedRepository: (() => Promise<SessionMetadata[]>) | undefined;
{
await using collection = createInMemorySessionCollection();
const repository = createSessionRepository({ collection });
listDisposedCollection = () => repository.list();
await using repository = new InMemorySessionRepository();
listDisposedRepository = () => repository.list();
await repository.create({ id: "session-1" });
}
await expect(listDisposedCollection!()).rejects.toThrow("In-memory session collection is disposed");
await expect(listDisposedRepository!()).rejects.toThrow("In-memory session repository is disposed");
});
});
describe("JsonlSessionCollection", () => {
describe("JsonlSessionBackend", () => {
it.each(["create", "fork"] as const)(
"serializes conflicting create and %s destinations",
async (secondOperation) => {
@ -319,14 +299,14 @@ describe("JsonlSessionCollection", () => {
`${JSON.stringify({ type: "session", version: 3, id: "source", timestamp: "2025-01-01T00:00:00.000Z", cwd: "/tmp/source" })}\n`,
);
env.writtenPaths.add(sourcePath);
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: root });
const options = { cwd: "/tmp/target", id: "shared-id" };
const first = collection.create(options);
const first = backend.create(options);
await env.firstWriteStarted.promise;
const second =
secondOperation === "create"
? collection.create(options)
: collection.fork(
? backend.create(options)
: backend.fork(
{
id: "source",
createdAt: "2025-01-01T00:00:00.000Z",
@ -353,8 +333,8 @@ describe("JsonlSessionCollection", () => {
it("encodes custom session IDs used in filenames", async () => {
const root = createTempDir();
const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const snapshot = await collection.create({ cwd: root, id: "../unsafe\\id" });
const backend = new JsonlSessionBackend({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const snapshot = await backend.create({ cwd: root, id: "../unsafe\\id" });
expect(snapshot.metadata.path).toContain("..%2Funsafe%5Cid.jsonl");
expect(existsSync(snapshot.metadata.path)).toBe(true);
@ -363,9 +343,9 @@ describe("JsonlSessionCollection", () => {
it("allows appends to different sessions to run concurrently", async () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root });
const first = await collection.create({ cwd: "/tmp/first", id: "first" });
const second = await collection.create({ cwd: "/tmp/second", id: "second" });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: root });
const first = await backend.create({ cwd: "/tmp/first", id: "first" });
const second = await backend.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.metadata.path);
const secondBlock = env.blockAppend(second.metadata.path);
@ -383,10 +363,10 @@ describe("JsonlSessionCollection", () => {
it("caps concurrent operations across JSONL sessions at four by default", async () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: root });
const snapshots = [];
for (let index = 0; index < 6; index++) {
snapshots.push(await collection.create({ cwd: `/tmp/session-${index}`, id: `session-${index}` }));
snapshots.push(await backend.create({ cwd: `/tmp/session-${index}`, id: `session-${index}` }));
}
const blocks = snapshots.map((snapshot) => env.blockAppend(snapshot.metadata.path));
const appends = snapshots.map((snapshot, index) => snapshot.appendEntry(createEntry(`entry-${index}`)));
@ -404,9 +384,9 @@ describe("JsonlSessionCollection", () => {
it("allows overriding the JSONL concurrency limit", async () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root, maxConcurrentOperations: 1 });
const first = await collection.create({ cwd: "/tmp/first", id: "first" });
const second = await collection.create({ cwd: "/tmp/second", id: "second" });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: root, maxConcurrentOperations: 1 });
const first = await backend.create({ cwd: "/tmp/first", id: "first" });
const second = await backend.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.metadata.path);
const secondBlock = env.blockAppend(second.metadata.path);
const firstAppend = first.appendEntry(createEntry("first-entry"));
@ -424,25 +404,26 @@ describe("JsonlSessionCollection", () => {
"rejects invalid JSONL concurrency limit %s",
(maxConcurrentOperations) => {
const root = createTempDir();
expect(() =>
createJsonlSessionCollection({
fs: new NodeExecutionEnv({ cwd: root }),
sessionsRoot: root,
maxConcurrentOperations,
}),
expect(
() =>
new JsonlSessionBackend({
fs: new NodeExecutionEnv({ cwd: root }),
sessionsRoot: root,
maxConcurrentOperations,
}),
).toThrow("maxConcurrentOperations must be a positive integer");
},
);
it("releases JSONL concurrency capacity after an operation fails", async () => {
const root = createTempDir();
const collection = createJsonlSessionCollection({
const backend = new JsonlSessionBackend({
fs: new NodeExecutionEnv({ cwd: root }),
sessionsRoot: root,
maxConcurrentOperations: 1,
});
const first = await collection.create({ cwd: "/tmp/first", id: "first" });
const second = await collection.create({ cwd: "/tmp/second", id: "second" });
const first = await backend.create({ cwd: "/tmp/first", id: "first" });
const second = await backend.create({ cwd: "/tmp/second", id: "second" });
const duplicate = createEntry("duplicate-entry");
await first.appendEntry(duplicate);
@ -456,8 +437,8 @@ describe("JsonlSessionCollection", () => {
it("serializes appends to the same session", async () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root });
const snapshot = await collection.create({ cwd: root, id: "session" });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: root });
const snapshot = await backend.create({ cwd: root, id: "session" });
const block = env.blockAppend(snapshot.metadata.path);
const firstAppend = snapshot.appendEntry(createEntry("first-entry"));
@ -469,7 +450,7 @@ describe("JsonlSessionCollection", () => {
block.release.resolve();
await Promise.all([firstAppend, secondAppend]);
expect(appendCountWhileFirstWasBlocked).toBe(1);
expect((await (await collection.open(snapshot.metadata)).readEntries()).map((entry) => entry.id)).toEqual([
expect((await (await backend.open(snapshot.metadata)).readEntries()).map((entry) => entry.id)).toEqual([
"first-entry",
"second-entry",
]);
@ -478,16 +459,16 @@ describe("JsonlSessionCollection", () => {
it("uses listing as a barrier between accepted session operations", async () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root });
const first = await collection.create({ cwd: "/tmp/first", id: "first" });
const second = await collection.create({ cwd: "/tmp/second", id: "second" });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: root });
const first = await backend.create({ cwd: "/tmp/first", id: "first" });
const second = await backend.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.metadata.path);
const secondBlock = env.blockAppend(second.metadata.path);
const listBlock = env.blockNextList();
const firstAppend = first.appendEntry(createEntry("first-entry"));
await firstBlock.started.promise;
const list = collection.list();
const list = backend.list();
const secondAppend = second.appendEntry(createEntry("second-entry"));
const listStartedBeforeFirstFinished = await settlesBeforeNextTurn(listBlock.started.promise);
@ -508,9 +489,9 @@ describe("JsonlSessionCollection", () => {
it("waits for every accepted session operation during disposal", async () => {
const root = createTempDir();
const env = new PerPathBlockingAppendEnv({ cwd: root });
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root });
const first = await collection.create({ cwd: "/tmp/first", id: "first" });
const second = await collection.create({ cwd: "/tmp/second", id: "second" });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: root });
const first = await backend.create({ cwd: "/tmp/first", id: "first" });
const second = await backend.create({ cwd: "/tmp/second", id: "second" });
const firstBlock = env.blockAppend(first.metadata.path);
const secondBlock = env.blockAppend(second.metadata.path);
const firstAppend = first.appendEntry(createEntry("first-entry"));
@ -518,7 +499,7 @@ describe("JsonlSessionCollection", () => {
await Promise.all([firstBlock.started.promise, secondBlock.started.promise]);
let disposed = false;
const dispose = collection[Symbol.asyncDispose]().then(() => {
const dispose = backend[Symbol.asyncDispose]().then(() => {
disposed = true;
});
firstBlock.release.resolve();
@ -535,13 +516,12 @@ describe("JsonlSessionCollection", () => {
it("waits for accepted appends before disposal and rejects later writes", async () => {
const root = createTempDir();
const env = new BlockingAppendEnv({ cwd: root });
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root });
const repo = createSessionRepository({ collection });
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const session = await repo.create({ cwd: root, id: "session-1" });
const append = session.appendMessage(createUserMessage("accepted"));
await env.appendStarted.promise;
let closed = false;
const dispose = collection[Symbol.asyncDispose]().then(() => {
const dispose = repo[Symbol.asyncDispose]().then(() => {
closed = true;
});
await Promise.resolve();
@ -550,16 +530,14 @@ describe("JsonlSessionCollection", () => {
await append;
await dispose;
await expect(session.appendMessage(createUserMessage("late"))).rejects.toThrow(
"JSONL session collection is disposed",
"JSONL session repository is disposed",
);
});
it("parses once when opened and retains state across appends", async () => {
const root = createTempDir();
const env = new CountingReadEnv({ cwd: root });
const repo = createSessionRepository({
collection: createJsonlSessionCollection({ fs: env, sessionsRoot: root }),
});
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const created = await repo.create({ cwd: root, id: "session-1" });
const metadata = await created.getMetadata();
await created.appendMessage(createUserMessage("initial"));
@ -569,14 +547,12 @@ describe("JsonlSessionCollection", () => {
expect(env.readCount).toBe(1);
});
it("collections sessions below encoded cwd directories and lists by cwd", async () => {
it("collects sessions below encoded cwd directories and lists by cwd", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const cwd = "/tmp/my-project";
const otherCwd = "/tmp/other-project";
const repo = createSessionRepository({
collection: createJsonlSessionCollection({ fs: env, sessionsRoot: root }),
});
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const session = await repo.create({ cwd, id: "019de8c2-de29-73e9-ae0c-e134db34c447" });
const otherSession = await repo.create({ cwd: otherCwd, id: "other-session" });
const metadata = await session.getMetadata();
@ -593,9 +569,7 @@ describe("JsonlSessionCollection", () => {
it("fails loudly when listing a malformed session file", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const repo = createSessionRepository({
collection: createJsonlSessionCollection({ fs: env, sessionsRoot: root }),
});
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const session = await repo.create({ cwd: root, id: "session-1" });
const metadata = await session.getMetadata();
writeFileSync(metadata.path, "not json\n");
@ -605,9 +579,10 @@ describe("JsonlSessionCollection", () => {
it("rejects a missing active leaf when opened", async () => {
const root = createTempDir();
const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const repo = createSessionRepository({ collection });
const storage = await collection.create({ cwd: root, id: "session-1" });
const env = new NodeExecutionEnv({ cwd: root });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: root });
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const storage = await backend.create({ cwd: root, id: "session-1" });
await storage.appendEntry({
type: "leaf",
id: "leaf",
@ -625,9 +600,7 @@ describe("JsonlSessionCollection", () => {
it("opens, deletes, and forks by metadata", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const repo = createSessionRepository({
collection: createJsonlSessionCollection({ fs: env, sessionsRoot: root }),
});
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const source = await repo.create({ cwd: "/tmp/source", id: "source-session" });
const sourceMetadata = await source.getMetadata();
const user1 = await source.appendMessage(createUserMessage("one"));
@ -649,9 +622,7 @@ describe("JsonlSessionCollection", () => {
it("persists header metadata through create, list, and fork", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const repo = createSessionRepository({
collection: createJsonlSessionCollection({ fs: env, sessionsRoot: root }),
});
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const source = await repo.create({
cwd: "/tmp/source",
id: "source-session",

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

@ -1,9 +1,12 @@
import { existsSync, readFileSync, writeFileSync } from "node:fs";
import { describe, expect, it } from "vitest";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { createJsonlSessionCollection, loadJsonlSessionMetadata } from "../../src/harness/session/jsonl-collection.ts";
import { createInMemorySessionCollection } from "../../src/harness/session/memory-collection.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import {
JsonlSessionBackend,
JsonlSessionRepository,
loadJsonlSessionMetadata,
} from "../../src/harness/session/jsonl-repo.ts";
import { InMemorySessionRepository } from "../../src/harness/session/memory-repo.ts";
import type { Session } from "../../src/harness/session/session.ts";
import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts";
@ -40,17 +43,21 @@ async function appendUsageEntries(session: Session) {
});
}
function createInMemoryRepository() {
return createSessionRepository({ collection: createInMemorySessionCollection() });
}
function createJsonlRepository(options: Parameters<typeof createJsonlSessionCollection>[0]) {
return createSessionRepository({ collection: createJsonlSessionCollection(options) });
}
describe("Session aggregate", () => {
it("repository disposal closes its owned storage", async () => {
let session: Session;
{
await using repository = new InMemorySessionRepository();
session = await repository.create({ id: "session-1" });
}
await expect(session!.appendMessage(createUserMessage("late"))).rejects.toThrow(
"In-memory session repository is disposed",
);
});
it("owns leaf navigation, labels, names, stats, and branch traversal", async () => {
const repo = createInMemoryRepository();
const repo = new InMemorySessionRepository();
const session = await repo.create({ id: "session-1" });
const root = await session.appendMessage(createUserMessage("root"));
const child = await session.appendMessage(createAssistantMessage("child"));
@ -69,7 +76,7 @@ describe("Session aggregate", () => {
});
it("serializes concurrent appends into one parent chain", async () => {
const session = await createInMemoryRepository().create({});
const session = await new InMemorySessionRepository().create({});
await Promise.all(
Array.from({ length: 20 }, (_, index) => session.appendMessage(createUserMessage(`message ${index}`))),
);
@ -81,7 +88,7 @@ describe("Session aggregate", () => {
});
it("includes assistant and summary usage in statistics", async () => {
const session = await createInMemoryRepository().create({});
const session = await new InMemorySessionRepository().create({});
await appendUsageEntries(session);
expect(await session.getSessionStats()).toEqual({
messageCount: 1,
@ -93,7 +100,7 @@ describe("Session aggregate", () => {
});
it("stops branch traversal at retained-tail compaction", async () => {
const session = await createInMemoryRepository().create({});
const session = await new InMemorySessionRepository().create({});
await session.appendMessage(createUserMessage("root"));
const child = await session.appendMessage(createAssistantMessage("child"));
const compaction = await session.appendCompaction("summary", child, 1234, undefined, undefined, undefined, [
@ -104,10 +111,10 @@ describe("Session aggregate", () => {
});
});
describe("JsonlSessionCollection", () => {
describe("JsonlSessionBackend", () => {
it("writes headers and entries and reopens the aggregate", async () => {
const root = createTempDir();
const repo = createJsonlRepository({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const repo = new JsonlSessionRepository({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const session = await repo.create({ cwd: root, id: "session-1", metadata: { profile: "reviewer" } });
const metadata = await session.getMetadata();
const entryId = await session.appendMessage(createUserMessage("one"));
@ -126,7 +133,7 @@ describe("JsonlSessionCollection", () => {
it("fails loudly for malformed headers and entries", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const repo = createJsonlRepository({ fs: env, sessionsRoot: root });
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const session = await repo.create({ cwd: root, id: "session-1" });
const metadata = await session.getMetadata();
writeFileSync(metadata.path, "not json\n");
@ -139,8 +146,8 @@ describe("JsonlSessionCollection", () => {
it("enforces entry uniqueness and does not recreate deleted files", async () => {
const root = createTempDir();
const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const snapshot = await collection.create({ cwd: root, id: "session-1" });
const backend = new JsonlSessionBackend({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const snapshot = await backend.create({ cwd: root, id: "session-1" });
const entry = {
type: "message" as const,
id: "entry-1",
@ -150,16 +157,16 @@ describe("JsonlSessionCollection", () => {
};
await snapshot.appendEntry(entry);
await expect(snapshot.appendEntry(entry)).rejects.toThrow("Entry entry-1 already exists");
await collection.delete(snapshot.metadata);
await backend.delete(snapshot.metadata);
await expect(snapshot.appendEntry({ ...entry, id: "entry-2" })).rejects.toThrow("Session not found");
expect(existsSync(snapshot.metadata.path)).toBe(false);
});
it("scopes entry uniqueness to the session path", async () => {
const root = createTempDir();
const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const first = await collection.create({ cwd: "/tmp/first", id: "shared-session-id" });
const second = await collection.create({ cwd: "/tmp/second", id: "shared-session-id" });
const backend = new JsonlSessionBackend({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root });
const first = await backend.create({ cwd: "/tmp/first", id: "shared-session-id" });
const second = await backend.create({ cwd: "/tmp/second", id: "shared-session-id" });
const entry = {
type: "message" as const,
id: "shared-entry-id",
@ -175,7 +182,7 @@ describe("JsonlSessionCollection", () => {
it("rejects non-object header metadata", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const repo = createJsonlRepository({ fs: env, sessionsRoot: root });
const repo = new JsonlSessionRepository({ fs: env, sessionsRoot: root });
const session = await repo.create({ cwd: root, id: "session-1" });
const metadata = await session.getMetadata();
const header = {

6
packages/agent/test/harness/session-test-utils.ts
View file

@ -3,13 +3,11 @@ import { tmpdir } from "node:os";
import { join } from "node:path";
import type { AgentMessage } from "@earendil-works/pi-agent-core";
import { afterEach } from "vitest";
import { createInMemorySessionCollection } from "../../src/harness/session/memory-collection.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import { InMemorySessionRepository } from "../../src/harness/session/memory-repo.ts";
import type { Session } from "../../src/harness/session/session.ts";
export async function createInMemorySession(id?: string): Promise<Session> {
const collection = createInMemorySessionCollection();
return createSessionRepository({ collection }).create({ id });
return new InMemorySessionRepository().create({ id });
}
export function createUserMessage(text: string): AgentMessage {

38
packages/agent/test/harness/session.test.ts
View file

@ -2,10 +2,14 @@ import { readFileSync } from "node:fs";
import { join } from "node:path";
import { describe, expect, it } from "vitest";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { createJsonlSessionCollection } from "../../src/harness/session/jsonl-collection.ts";
import { createInMemorySessionCollection } from "../../src/harness/session/memory-collection.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import type { ContextEntryTransform, Session, SessionContextBuildOptions } from "../../src/harness/session/session.ts";
import { JsonlSessionBackend } from "../../src/harness/session/jsonl-repo.ts";
import { InMemorySessionBackend } from "../../src/harness/session/memory-repo.ts";
import {
type ContextEntryTransform,
createSession,
type Session,
type SessionContextBuildOptions,
} from "../../src/harness/session/session.ts";
import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts";
function getTextData(data: unknown): string {
@ -227,7 +231,7 @@ async function runSessionSuite(name: string, createFixture: () => Promise<Sessio
await expect(session.appendLabel("missing", "checkpoint")).rejects.toThrow("Entry missing not found");
});
it("persists leaf changes and appended entries through the collection", async () => {
it("persists leaf changes and appended entries through the backend", async () => {
const fixture = await createFixture();
const session = await fixture.createSession();
const user1 = await session.appendMessage(createUserMessage("one"));
@ -246,33 +250,29 @@ async function runSessionSuite(name: string, createFixture: () => Promise<Sessio
});
}
runSessionSuite("Session with in-memory collection", async () => {
const collection = createInMemorySessionCollection();
const created = await createSessionRepository({ collection }).create({});
runSessionSuite("Session with in-memory backend", async () => {
const backend = new InMemorySessionBackend();
const created = await createSession(await backend.create({}));
const metadata = await created.getMetadata();
return {
createSession: (contextBuildOptions) =>
createSessionRepository({ collection, contextBuildOptions }).open(metadata),
reloadSession: (contextBuildOptions) =>
createSessionRepository({ collection, contextBuildOptions }).open(metadata),
createSession: async (contextBuildOptions) => createSession(await backend.open(metadata), contextBuildOptions),
reloadSession: async (contextBuildOptions) => createSession(await backend.open(metadata), contextBuildOptions),
};
});
let jsonlSessionPath = "";
runSessionSuite(
"Session with JSONL collection",
"Session with JSONL backend",
async () => {
const dir = createTempDir();
const env = new NodeExecutionEnv({ cwd: dir });
const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: join(dir, "sessions") });
const created = await createSessionRepository({ collection }).create({ cwd: dir, id: "session-1" });
const backend = new JsonlSessionBackend({ fs: env, sessionsRoot: join(dir, "sessions") });
const created = await createSession(await backend.create({ cwd: dir, id: "session-1" }));
const metadata = await created.getMetadata();
jsonlSessionPath = metadata.path;
return {
createSession: (contextBuildOptions) =>
createSessionRepository({ collection, contextBuildOptions }).open(metadata),
reloadSession: (contextBuildOptions) =>
createSessionRepository({ collection, contextBuildOptions }).open(metadata),
createSession: async (contextBuildOptions) => createSession(await backend.open(metadata), contextBuildOptions),
reloadSession: async (contextBuildOptions) => createSession(await backend.open(metadata), contextBuildOptions),
};
},
() => {

35
packages/agent/test/harness/sqlite-branch-cache.test.ts
View file

@ -2,9 +2,8 @@ import { mkdtempSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { describe, expect, it } from "vitest";
import { createNodeSqliteFactory, createSqliteSessionCollection } from "../../../storage/sqlite-node/src/index.ts";
import { createNodeSqliteFactory, SqliteSessionRepository } from "../../../storage/sqlite-node/src/index.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import { createAssistantMessage, createUserMessage } from "./session-test-utils.ts";
function createTempDir(): string {
@ -17,9 +16,7 @@ describe("SQLite branch cache", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite, databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const rootId = await session.appendMessage(createUserMessage("root"));
const keptId = await session.appendMessage(createUserMessage("kept"));
@ -48,9 +45,7 @@ describe("SQLite branch cache", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite, databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const oldId = await session.appendMessage(createUserMessage("old"));
const keptId = await session.appendMessage(createUserMessage("kept"));
@ -73,9 +68,7 @@ describe("SQLite branch cache", () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite: createNodeSqliteFactory(), databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const rootId = await session.appendMessage(createUserMessage("root"));
const firstCompactionId = await session.appendCompaction(
@ -104,9 +97,7 @@ describe("SQLite branch cache", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite, databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const rootId = await session.appendMessage(createUserMessage("root"));
const childId = await session.appendMessage(createAssistantMessage("child"));
@ -142,9 +133,7 @@ describe("SQLite branch cache", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite, databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
await session.appendMessage(createUserMessage("root"));
@ -185,9 +174,7 @@ describe("SQLite branch cache", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite, databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const source = await repo.create({ cwd: root, id: "source" });
const rootId = await source.appendMessage(createUserMessage("root"));
const childId = await source.appendMessage(createAssistantMessage("child"));
@ -214,9 +201,7 @@ describe("SQLite branch cache", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite, databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const rootId = await session.appendMessage(createUserMessage("root"));
const staleId = await session.appendMessage(createAssistantMessage("stale"));
@ -257,9 +242,7 @@ describe("SQLite branch cache", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite, databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
await session.appendMessage(createUserMessage("root"));
const metadata = await session.getMetadata();

60
packages/agent/test/harness/sqlite-migrations.test.ts
View file

@ -5,17 +5,16 @@ import { describe, expect, it } from "vitest";
import {
applyMigrations,
createNodeSqliteFactory,
createSqliteSessionCollection,
loadMigrations,
type SqliteDatabase,
type SqliteDatabaseFactory,
type SqliteRunResult,
type SqliteSessionMetadata,
SqliteSessionRepository,
type SqliteStatement,
} from "../../../storage/sqlite-node/src/index.ts";
import { SqliteSessionConnection } from "../../../storage/sqlite-node/src/sqlite/storage/index.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import { createAssistantMessage, createUserMessage } from "./session-test-utils.ts";
function createTempDir(): string {
@ -97,8 +96,7 @@ describe("SQLite migrations", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
await repo.create({ cwd: root, id: "session-1" });
const db = await sqlite.open(databasePath);
@ -178,9 +176,7 @@ describe("SQLite migrations", () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite: createNodeSqliteFactory(), databasePath });
const source = await repo.create({
cwd: root,
id: "session-1",
@ -205,8 +201,7 @@ describe("SQLite migrations", () => {
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const source = await repo.create({ cwd: root, id: "source" });
await source.appendMessage(createUserMessage("one"));
await source.appendMessage(createAssistantMessage("two"));
@ -245,8 +240,7 @@ END;
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const rootId = await session.appendMessage(createUserMessage("root"));
const childId = await session.appendMessage(createAssistantMessage("child"));
@ -287,8 +281,7 @@ END;
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const rootId = await session.appendMessage(createUserMessage("root"));
const firstChildId = await session.appendMessage(createAssistantMessage("first child"));
@ -321,9 +314,7 @@ END;
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite: createNodeSqliteFactory(), databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const rootId = await session.appendMessage(createUserMessage("root"));
await session.appendMessage(createAssistantMessage("first child"));
@ -343,9 +334,7 @@ END;
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const repo = createSessionRepository({
collection: createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath }),
});
const repo = new SqliteSessionRepository({ env, sqlite: createNodeSqliteFactory(), databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const ids = [
await session.appendMessage(createUserMessage("one")),
@ -371,12 +360,11 @@ END;
open: async () => db,
};
const env = new NodeExecutionEnv({ cwd: root });
const collection = createSqliteSessionCollection({ env, sqlite, databasePath: join(root, "sessions.sqlite") });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath: join(root, "sessions.sqlite") });
await expect(repo.create({ cwd: root, id: "session-1" })).rejects.toThrow("insert failed");
expect(db.closeCount).toBe(0);
await collection[Symbol.asyncDispose]();
await repo[Symbol.asyncDispose]();
expect(db.closeCount).toBe(1);
});
@ -392,8 +380,7 @@ END;
open: async () => db,
};
const env = new NodeExecutionEnv({ cwd: root });
const collection = createSqliteSessionCollection({ env, sqlite, databasePath: join(root, "sessions.sqlite") });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath: join(root, "sessions.sqlite") });
const metadata: SqliteSessionMetadata = {
id: "missing",
createdAt: new Date().toISOString(),
@ -404,7 +391,7 @@ END;
await expect(repo.open(metadata)).rejects.toThrow("Session not found: missing");
expect(db.closeCount).toBe(0);
await collection[Symbol.asyncDispose]();
await repo[Symbol.asyncDispose]();
expect(db.closeCount).toBe(1);
});
@ -413,32 +400,30 @@ END;
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const { sqlite, counts } = createCloseCountingSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
for (let i = 0; i < 10; i++) await session.appendMessage(createUserMessage(`message ${i}`));
await session.getEntries();
expect(counts).toEqual({ opens: 1, closes: 0 });
await collection[Symbol.asyncDispose]();
await repo[Symbol.asyncDispose]();
expect(counts).toEqual({ opens: 1, closes: 1 });
await collection[Symbol.asyncDispose]();
await repo[Symbol.asyncDispose]();
expect(counts).toEqual({ opens: 1, closes: 1 });
});
it("shares one connection across source and fork until the collection is disposed", async () => {
it("shares one connection across source and fork until the repository is disposed", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const { sqlite, counts } = createCloseCountingSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const source = await repo.create({ cwd: root, id: "session-1" });
const fork = await repo.fork(await source.getMetadata(), { cwd: root, id: "session-2" });
await fork.appendMessage(createUserMessage("fork"));
expect(counts).toEqual({ opens: 1, closes: 0 });
await collection[Symbol.asyncDispose]();
await repo[Symbol.asyncDispose]();
expect(counts).toEqual({ opens: 1, closes: 1 });
});
@ -447,8 +432,7 @@ END;
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const metadata = await session.getMetadata();
@ -470,8 +454,7 @@ END;
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const entryId = await session.appendMessage(createUserMessage("message"));
const metadata = await session.getMetadata();
@ -544,8 +527,7 @@ END;
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const collection = createSqliteSessionCollection({ env, sqlite, databasePath });
const repo = createSessionRepository({ collection });
const repo = new SqliteSessionRepository({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const userId = await session.appendMessage(createUserMessage("one"));
await session.appendThinkingLevelChange("high");

74
packages/agent/test/harness/sqlite-node.test.ts
View file

@ -2,71 +2,67 @@ import { join } from "node:path";
import { afterEach, describe, expect, it } from "vitest";
import {
createNodeSqliteFactory,
createSqliteSessionCollection,
createSqliteSessionSearch,
type SqliteSessionMetadata,
SqliteSessionRepository,
} from "../../../storage/sqlite-node/src/index.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import { createJsonlSessionCollection } from "../../src/harness/session/jsonl-collection.ts";
import { createSessionRepository } from "../../src/harness/session/repository.ts";
import { createScanningSessionSearch } from "../../src/harness/session/scanning-session-search.ts";
import { JsonlSessionRepository } from "../../src/harness/session/jsonl-repo.ts";
import { createScanningSessionSearch } from "../../src/harness/session/search.ts";
import type { SessionSearch, SessionSearchHit, SessionSearchOptions } from "../../src/harness/types.ts";
import { createTempDir, createUserMessage } from "./session-test-utils.ts";
const ownedCollections: AsyncDisposable[] = [];
const ownedRepositories: AsyncDisposable[] = [];
afterEach(async () => {
for (const collection of ownedCollections.splice(0)) await collection[Symbol.asyncDispose]();
for (const repository of ownedRepositories.splice(0)) await repository[Symbol.asyncDispose]();
});
function createSqliteRepository(options: Parameters<typeof createSqliteSessionCollection>[0]) {
const collection = createSqliteSessionCollection(options);
ownedCollections.push(collection);
return createSessionRepository({
collection,
search: createSqliteSessionSearch(options),
});
function createSqliteFixture(options: ConstructorParameters<typeof SqliteSessionRepository>[0]) {
const repository = new SqliteSessionRepository(options);
ownedRepositories.push(repository);
return { repository, search: createSqliteSessionSearch(options) };
}
function createJsonlRepository(options: Parameters<typeof createJsonlSessionCollection>[0]) {
const collection = createJsonlSessionCollection(options);
ownedCollections.push(collection);
return createSessionRepository({ collection, search: createScanningSessionSearch(collection) });
function createJsonlFixture(options: ConstructorParameters<typeof JsonlSessionRepository>[0]) {
const repository = new JsonlSessionRepository(options);
ownedRepositories.push(repository);
return { repository, search: createScanningSessionSearch(repository) };
}
describe("JsonlSessionCollection with scanning search", () => {
describe("JsonlSessionBackend with scanning search", () => {
it("searches canonical session entries by scanning", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const repo = createJsonlRepository({ fs: env, sessionsRoot: join(root, "sessions") });
const { repository: repo, search } = createJsonlFixture({ fs: env, sessionsRoot: join(root, "sessions") });
const included = await repo.create({ cwd: root, id: "included" });
const excluded = await repo.create({ cwd: `${root}/other`, id: "excluded" });
const entryId = await included.appendMessage(createUserMessage("Find the auth defect"));
await excluded.appendMessage(createUserMessage("Find the auth defect"));
await expect(repo.search({ text: "AUTH", cwd: root })).resolves.toEqual([
await expect(search.search({ text: "AUTH", cwd: root })).resolves.toEqual([
expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "included" }) }),
]);
});
});
describe("SqliteSessionCollection with explicit SQLite FTS5 search", () => {
it("uses SQLite FTS5 when composed with its search collection", async () => {
describe("SqliteSessionBackend with explicit SQLite FTS5 search", () => {
it("uses SQLite FTS5 when composed with its search implementation", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const databasePath = join(root, "sessions.sqlite");
const repo = createSqliteRepository({ env, sqlite, databasePath });
const { repository: repo, search } = createSqliteFixture({ env, sqlite, databasePath });
const included = await repo.create({ cwd: root, id: "included" });
const excluded = await repo.create({ cwd: `${root}/other`, id: "excluded" });
const metadata = await included.getMetadata();
const entryId = await included.appendMessage(createUserMessage("Find the auth defect"));
await excluded.appendMessage(createUserMessage("Find the auth defect"));
await expect(repo.search({ text: "auth", cwd: root })).resolves.toEqual([
await expect(search.search({ text: "auth", cwd: root })).resolves.toEqual([
expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "included" }) }),
]);
await expect(repo.search({ text: "uth", cwd: root })).resolves.toEqual([
await expect(search.search({ text: "uth", cwd: root })).resolves.toEqual([
expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "included" }) }),
]);
@ -82,7 +78,7 @@ describe("SqliteSessionCollection with explicit SQLite FTS5 search", () => {
}
await repo.delete(metadata);
await expect(repo.search({ text: "auth", cwd: root })).resolves.toEqual([]);
await expect(search.search({ text: "auth", cwd: root })).resolves.toEqual([]);
});
it("creates an empty canonical session without initializing FTS", async () => {
@ -90,7 +86,7 @@ describe("SqliteSessionCollection with explicit SQLite FTS5 search", () => {
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const databasePath = join(root, "sessions.sqlite");
const repo = createSqliteRepository({ env, sqlite, databasePath });
const { repository: repo } = createSqliteFixture({ env, sqlite, databasePath });
const session = await repo.create({ cwd: root, id: "session-1" });
const db = await sqlite.open(databasePath);
@ -110,8 +106,8 @@ describe("SqliteSessionCollection with explicit SQLite FTS5 search", () => {
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const databasePath = join(root, "sessions.sqlite");
const repo = createSqliteRepository({ env, sqlite, databasePath });
await repo.search({ text: "initialize" });
const { repository: repo, search } = createSqliteFixture({ env, sqlite, databasePath });
await search.search({ text: "initialize" });
const session = await repo.create({ cwd: root, id: "session-1" });
const db = await sqlite.open(databasePath);
@ -130,8 +126,8 @@ describe("SqliteSessionCollection with explicit SQLite FTS5 search", () => {
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const databasePath = join(root, "sessions.sqlite");
const repo = createSqliteRepository({ env, sqlite, databasePath });
await repo.search({ text: "initialize" });
const { repository: repo, search } = createSqliteFixture({ env, sqlite, databasePath });
await search.search({ text: "initialize" });
const session = await repo.create({ cwd: root, id: "session-1" });
await session.appendMessage(createUserMessage("must remain"));
const metadata = await session.getMetadata();
@ -151,25 +147,25 @@ describe("SqliteSessionCollection with explicit SQLite FTS5 search", () => {
it("initializes canonical storage when searched before the first session is created", async () => {
const root = createTempDir();
const env = new NodeExecutionEnv({ cwd: root });
const repo = createSqliteRepository({
const { repository: repo, search } = createSqliteFixture({
env,
sqlite: createNodeSqliteFactory(),
databasePath: join(root, "sessions.sqlite"),
});
await expect(repo.search({ text: "auth" })).resolves.toEqual([]);
await expect(search.search({ text: "auth" })).resolves.toEqual([]);
const session = await repo.create({ cwd: root, id: "session-1" });
const entryId = await session.appendMessage(createUserMessage("Find the auth defect"));
await expect(repo.search({ text: "auth" })).resolves.toEqual([
await expect(search.search({ text: "auth" })).resolves.toEqual([
expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "session-1" }) }),
]);
await expect(session.appendMessage(createUserMessage("Still writable"))).resolves.toBeTypeOf("string");
});
});
describe("SqliteSessionCollection with custom search", () => {
it("can swap out the default search implementation", async () => {
describe("SqliteSessionRepository with custom search", () => {
it("uses an independently supplied search implementation", async () => {
const root = createTempDir();
const searches: SessionSearchOptions[] = [];
const search: SessionSearch<SqliteSessionMetadata> = {
@ -179,16 +175,16 @@ describe("SqliteSessionCollection with custom search", () => {
},
};
const env = new NodeExecutionEnv({ cwd: root });
const collection = createSqliteSessionCollection({
const repo = new SqliteSessionRepository({
env,
sqlite: createNodeSqliteFactory(),
databasePath: join(root, "sessions.sqlite"),
});
const repo = createSessionRepository({ collection, search });
ownedRepositories.push(repo);
const session = await repo.create({ cwd: root, id: "session-1" });
await session.appendMessage(createUserMessage("stored canonically"));
await expect(repo.search({ text: "custom query" })).resolves.toEqual([]);
await expect(search.search({ text: "custom query" })).resolves.toEqual([]);
expect(searches).toEqual([{ text: "custom query" }]);
});
});

7
packages/agent/test/scratch/simple.ts
View file

@ -8,11 +8,10 @@ import {
AgentHarness,
createBashTool,
createEditTool,
createInMemorySessionCollection,
createReadTool,
createSessionRepository,
createWriteTool,
formatSkillsForSystemPrompt,
InMemorySessionRepository,
loadSourcedPromptTemplates,
loadSourcedSkills,
type PromptTemplate,
@ -51,8 +50,8 @@ if (!model) {
process.exit(-1);
}
await using collection = createInMemorySessionCollection();
const session = await createSessionRepository({ collection }).create({});
await using repository = new InMemorySessionRepository();
const session = await repository.create({});
const agent = new AgentHarness({
session,
models,

5
packages/storage/sqlite-node/CHANGELOG.md
View file

@ -2,8 +2,13 @@
## [Unreleased]
### Breaking Changes
- Replaced `SqliteSessionRepo` with the resource-owning `SqliteSessionRepository`.
### Added
- Added query-only SQLite FTS search over canonical session entries.
- Added bounded active-branch queries to the SQLite session reader.
## [0.83.0] - 2026-07-29

13
packages/storage/sqlite-node/README.md
View file

@ -1,12 +1,15 @@
# @earendil-works/pi-storage-sqlite-node
Node sqlite storage backend for `@earendil-works/pi-agent-core` sessions. Provides the
`node:sqlite` adapter (`SqliteDatabase` implementation) and the SQLite session
session collection (`createSqliteSessionCollection`, migrations, materialized views). The
collection lazily owns one shared database connection and implements `AsyncDisposable`.
`node:sqlite` adapter (`SqliteDatabase` implementation), SQLite session repository,
migrations, materialized views, and optional FTS search.
```ts
await using collection = createSqliteSessionCollection(options);
await using repository = new SqliteSessionRepository(options);
const search = createSqliteSessionSearch(options);
const repository = createSessionRepository({ collection, search });
const session = await repository.create({ cwd });
const hits = await search.search({ text: "needle" });
```
The repository lazily owns one shared database connection. Search is an independent,
query-only projection over the same canonical database.

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

@ -1,4 +1,7 @@
export * from "./migrations.ts";
export {
SqliteSessionRepository,
type SqliteSessionRepositoryOptions,
} from "./repo.ts";
export * from "./search-backend.ts";
export * from "./session-collection.ts";
export * from "./types.ts";

73
packages/storage/sqlite-node/src/sqlite/session-collection.ts → packages/storage/sqlite-node/src/sqlite/repo.ts
View file

@ -1,13 +1,18 @@
import type {
SessionCollection,
SessionForkOptions,
SessionForkSelection,
SessionRepository,
SessionStorage,
SessionTreeEntry,
} from "@earendil-works/pi-agent-core";
import {
createSession,
createSessionForkSelection,
createSessionId,
getFileSystemResultOrThrow,
readSessionEntriesForFork,
type Session,
type SessionContextBuildOptions,
SessionError,
} from "@earendil-works/pi-agent-core";
import { applyMigrations } from "./migrations.ts";
@ -16,10 +21,10 @@ import { rowToMetadata, type SessionRow } from "./storage/sessions.ts";
import type {
SqliteDatabase,
SqliteDatabaseFactory,
SqliteSessionCollectionEnv,
SqliteSessionCreateOptions,
SqliteSessionListOptions,
SqliteSessionMetadata,
SqliteSessionRepositoryEnv,
} from "./types.ts";
function getParentPath(path: string): string {
@ -36,8 +41,8 @@ async function configureSqliteDatabase(db: SqliteDatabase): Promise<void> {
await db.exec("PRAGMA busy_timeout=5000");
}
export type SqliteSessionCollectionOptions = {
env: SqliteSessionCollectionEnv;
export type SqliteSessionBackendOptions = {
env: SqliteSessionRepositoryEnv;
sqlite: SqliteDatabaseFactory;
databasePath: string;
};
@ -59,10 +64,8 @@ class SerialOperationQueue {
}
}
class SqliteSessionCollection
implements SessionCollection<SqliteSessionMetadata, SqliteSessionCreateOptions, SqliteSessionListOptions>
{
private readonly env: SqliteSessionCollectionEnv;
class SqliteSessionBackend {
private readonly env: SqliteSessionRepositoryEnv;
private readonly sqlite: SqliteDatabaseFactory;
private readonly databasePathInput: string;
private databasePath: string | undefined;
@ -73,7 +76,7 @@ class SqliteSessionCollection
private readonly operations = new SerialOperationQueue();
private readonly writers = new Map<string, SqliteSessionConnection>();
constructor(options: SqliteSessionCollectionOptions) {
constructor(options: SqliteSessionBackendOptions) {
this.env = options.env;
this.sqlite = options.sqlite;
this.databasePathInput = options.databasePath;
@ -209,7 +212,7 @@ class SqliteSessionCollection
}
private assertOpen(): void {
if (this.disposed) throw new SessionError("storage", "SQLite session collection is disposed");
if (this.disposed) throw new SessionError("storage", "SQLite session repository is disposed");
}
private storage(connection: SqliteSessionConnection): SessionStorage<SqliteSessionMetadata> {
@ -275,8 +278,50 @@ class SqliteSessionCollection
}
}
export function createSqliteSessionCollection(
options: SqliteSessionCollectionOptions,
): SessionCollection<SqliteSessionMetadata, SqliteSessionCreateOptions, SqliteSessionListOptions> {
return new SqliteSessionCollection(options);
export interface SqliteSessionRepositoryOptions extends SqliteSessionBackendOptions {
contextBuildOptions?: SessionContextBuildOptions;
}
export class SqliteSessionRepository
implements SessionRepository<SqliteSessionMetadata, SqliteSessionCreateOptions, SqliteSessionListOptions>
{
private readonly backend: SqliteSessionBackend;
private readonly contextBuildOptions: SessionContextBuildOptions;
constructor(options: SqliteSessionRepositoryOptions) {
const { contextBuildOptions, ...backendOptions } = options;
this.backend = new SqliteSessionBackend(backendOptions);
this.contextBuildOptions = contextBuildOptions ?? {};
}
async create(options: SqliteSessionCreateOptions): Promise<Session<SqliteSessionMetadata>> {
return createSession(await this.backend.create(options), this.contextBuildOptions);
}
async open(metadata: SqliteSessionMetadata): Promise<Session<SqliteSessionMetadata>> {
return createSession(await this.backend.open(metadata), this.contextBuildOptions);
}
async list(options?: SqliteSessionListOptions): Promise<SqliteSessionMetadata[]> {
return await this.backend.list(options);
}
async delete(metadata: SqliteSessionMetadata): Promise<void> {
await this.backend.delete(metadata);
}
async fork(
source: SqliteSessionMetadata,
options: SessionForkOptions & SqliteSessionCreateOptions,
): Promise<Session<SqliteSessionMetadata>> {
const { entryId: _entryId, position: _position, ...createOptions } = options;
return createSession(
await this.backend.fork(source, createOptions, createSessionForkSelection(options)),
this.contextBuildOptions,
);
}
async [Symbol.asyncDispose](): Promise<void> {
await this.backend[Symbol.asyncDispose]();
}
}

4
packages/storage/sqlite-node/src/sqlite/search-backend.ts
View file

@ -5,8 +5,8 @@ import { rowToMetadata, type SessionRow } from "./storage/sessions.ts";
import type {
SqliteDatabase,
SqliteDatabaseFactory,
SqliteSessionCollectionEnv,
SqliteSessionMetadata,
SqliteSessionRepositoryEnv,
} from "./types.ts";
function getParentPath(path: string): string {
@ -24,7 +24,7 @@ async function configureSqliteDatabase(db: SqliteDatabase): Promise<void> {
}
export interface SqliteSessionSearchOptions {
env: Pick<SqliteSessionCollectionEnv, "absolutePath" | "createDir">;
env: Pick<SqliteSessionRepositoryEnv, "absolutePath" | "createDir">;
sqlite: SqliteDatabaseFactory;
databasePath: string;
}

2
packages/storage/sqlite-node/src/sqlite/types.ts
View file

@ -44,4 +44,4 @@ export interface SqliteSessionListOptions {
cwd?: string;
}
export type SqliteSessionCollectionEnv = Pick<FileSystem, "absolutePath" | "createDir" | "exists">;
export type SqliteSessionRepositoryEnv = Pick<FileSystem, "absolutePath" | "createDir" | "exists">;

5
scripts/browser-smoke-entry.ts
View file

@ -6,8 +6,7 @@ import {
bashExecutionToText,
convertToLlm,
createCustomMessage,
createInMemorySessionCollection,
createSessionRepository,
InMemorySessionRepository,
FileError,
formatPromptTemplateInvocation,
formatSkillInvocation,
@ -29,7 +28,7 @@ const stream = createAssistantMessageEventStream();
const agent = new Agent({ initialState: { model }, streamFn: streamSimple });
agent.steer({ role: "user", content: [{ type: "text", text: "queued" }], timestamp: 0 });
const repo = createSessionRepository({ collection: createInMemorySessionCollection() });
const repo = new InMemorySessionRepository();
const result = getOrThrow(ok({ value: 1 }));
const customMessage = createCustomMessage("note", "hello", true, undefined, "2026-01-01T00:00:00.000Z");
const llmMessages = convertToLlm([customMessage]);