diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index bf3744569..2273b8818 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -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 diff --git a/packages/agent/docs/agent-harness.md b/packages/agent/docs/agent-harness.md index d22fe878e..6a97a2778 100644 --- a/packages/agent/docs/agent-harness.md +++ b/packages/agent/docs/agent-harness.md @@ -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. diff --git a/packages/agent/docs/harness.md b/packages/agent/docs/harness.md index 35b2e6ffc..73cb5352a 100644 --- a/packages/agent/docs/harness.md +++ b/packages/agent/docs/harness.md @@ -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>; + open(metadata: TMetadata): Promise>; + list(options?: TListOptions): Promise; + delete(metadata: TMetadata): Promise; + fork( + source: TMetadata, + options: SessionForkOptions & TCreateOptions, + ): Promise>; +} +``` + +Repository implementations use `createSession(storage, contextBuildOptions)` to +wrap each created or opened storage connection. The complete per-session storage +contract is: ```ts interface SessionStorage { @@ -2034,56 +2050,27 @@ interface SessionStorage { getName(): Promise; getStats(): Promise; } - -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>; - open(metadata: TMetadata): Promise>; - list(options?: TListOptions): Promise; - delete(metadata: TMetadata): Promise; - fork( - source: TMetadata, - options: TCreateOptions, - selection: SessionForkSelection, - ): Promise>; -} ``` -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` diff --git a/packages/agent/src/harness/session/fork.ts b/packages/agent/src/harness/session/fork.ts deleted file mode 100644 index 8dc79e5f7..000000000 --- a/packages/agent/src/harness/session/fork.ts +++ /dev/null @@ -1,32 +0,0 @@ -import type { SessionForkOptions, SessionForkSelection, SessionTreeEntry } from "../types.ts"; -import { SessionError } from "../types.ts"; - -type MaybePromise = T | Promise; - -interface SessionForkEntrySource { - readEntry(id: string): MaybePromise; - readEntries(): MaybePromise; - readPathToRootOrCompaction(leafId: string | null): MaybePromise; -} - -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 { - 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); -} diff --git a/packages/agent/src/harness/session/jsonl-collection.ts b/packages/agent/src/harness/session/jsonl-repo.ts similarity index 87% rename from packages/agent/src/harness/session/jsonl-collection.ts rename to packages/agent/src/harness/session/jsonl-repo.ts index 5e68b47c3..fa074f24f 100644 --- a/packages/agent/src/harness/session/jsonl-collection.ts +++ b/packages/agent/src/harness/session/jsonl-repo.ts @@ -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 -{ - private readonly fs: JsonlSessionCollectionFileSystem; +export class JsonlSessionBackend { + private readonly fs: JsonlSessionRepositoryFileSystem; private readonly sessionsRootInput: string; private sessionsRoot: string | undefined; private readonly entryIndexesByPath = new Map(); @@ -211,7 +216,7 @@ class JsonlSessionCollection private disposed = false; private disposePromise: Promise | 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 { - return new JsonlSessionCollection(options); +export interface JsonlSessionRepositoryOptions extends JsonlSessionBackendOptions { + contextBuildOptions?: SessionContextBuildOptions; +} + +export class JsonlSessionRepository + implements SessionRepository +{ + 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> { + return createSession(await this.backend.create(options), this.contextBuildOptions); + } + + async open(metadata: JsonlSessionMetadata): Promise> { + return createSession(await this.backend.open(metadata), this.contextBuildOptions); + } + + async list(options?: JsonlSessionListOptions): Promise { + return await this.backend.list(options); + } + + async delete(metadata: JsonlSessionMetadata): Promise { + await this.backend.delete(metadata); + } + + async fork( + source: JsonlSessionMetadata, + options: SessionForkOptions & JsonlSessionCreateOptions, + ): Promise> { + const { entryId: _entryId, position: _position, ...createOptions } = options; + return createSession( + await this.backend.fork(source, createOptions, createSessionForkSelection(options)), + this.contextBuildOptions, + ); + } + + async [Symbol.asyncDispose](): Promise { + await this.backend[Symbol.asyncDispose](); + } } diff --git a/packages/agent/src/harness/session/memory-collection.ts b/packages/agent/src/harness/session/memory-repo.ts similarity index 68% rename from packages/agent/src/harness/session/memory-collection.ts rename to packages/agent/src/harness/session/memory-repo.ts index 639c4357e..b28e0369e 100644 --- a/packages/agent/src/harness/session/memory-collection.ts +++ b/packages/agent/src/harness/session/memory-repo.ts @@ -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 { +export class InMemorySessionBackend { private readonly sessions = new Map(); private readonly operations = new KeyedOperationQueue(); private disposed = false; @@ -109,7 +115,7 @@ class InMemorySessionCollection implements SessionCollection { - return new InMemorySessionCollection(); +export interface InMemorySessionRepositoryOptions { + contextBuildOptions?: SessionContextBuildOptions; +} + +export class InMemorySessionRepository + implements SessionRepository +{ + private readonly backend = new InMemorySessionBackend(); + private readonly contextBuildOptions: SessionContextBuildOptions; + + constructor(options: InMemorySessionRepositoryOptions = {}) { + this.contextBuildOptions = options.contextBuildOptions ?? {}; + } + + async create(options: InMemorySessionCreateOptions = {}): Promise> { + return createSession(await this.backend.create(options), this.contextBuildOptions); + } + + async open(metadata: SessionMetadata): Promise> { + return createSession(await this.backend.open(metadata), this.contextBuildOptions); + } + + async list(): Promise { + return await this.backend.list(); + } + + async delete(metadata: SessionMetadata): Promise { + await this.backend.delete(metadata); + } + + async fork( + source: SessionMetadata, + options: SessionForkOptions & InMemorySessionCreateOptions, + ): Promise> { + const { entryId: _entryId, position: _position, ...createOptions } = options; + return createSession( + await this.backend.fork(source, createOptions, createSessionForkSelection(options)), + this.contextBuildOptions, + ); + } + + async [Symbol.asyncDispose](): Promise { + await this.backend[Symbol.asyncDispose](); + } } diff --git a/packages/agent/src/harness/session/repository.ts b/packages/agent/src/harness/session/repository.ts index 3c5b414aa..76893efd5 100644 --- a/packages/agent/src/harness/session/repository.ts +++ b/packages/agent/src/harness/session/repository.ts @@ -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; - private readonly sessionSearch: SessionSearch | undefined; - private readonly contextBuildOptions: SessionContextBuildOptions; - - constructor(options: { - collection: SessionCollection; - search?: SessionSearch; - contextBuildOptions?: SessionContextBuildOptions; - }) { - this.collection = options.collection; - this.sessionSearch = options.search; - this.contextBuildOptions = options.contextBuildOptions ?? {}; - } - - async create(options: TCreateOptions): Promise> { - return createSession(await this.collection.create(options), this.contextBuildOptions); - } - - async open(metadata: TMetadata): Promise> { - return createSession(await this.collection.open(metadata), this.contextBuildOptions); - } - - async list(options?: TListOptions): Promise { - return await this.collection.list(options); - } - - async delete(metadata: TMetadata): Promise { - await this.collection.delete(metadata); - } - - async fork(source: TMetadata, options: SessionForkOptions & TCreateOptions): Promise> { - 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["search"]>[0]): Promise[]> { - 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; - search?: SessionSearch; - contextBuildOptions?: SessionContextBuildOptions; -}): SessionRepository { - return new SessionRepository(options); +> extends AsyncDisposable { + create(options: TCreateOptions): Promise>; + open(metadata: TMetadata): Promise>; + list(options?: TListOptions): Promise; + delete(metadata: TMetadata): Promise; + fork(source: TMetadata, options: SessionForkOptions & TCreateOptions): Promise>; } export function getFileSystemResultOrThrow(result: Result, message: string): TValue { @@ -89,3 +38,34 @@ export function getFileSystemResultOrThrow(result: Result = T | Promise; + +interface SessionForkEntrySource { + readEntry(id: string): MaybePromise; + readEntries(): MaybePromise; + readPathToRootOrCompaction(leafId: string | null): MaybePromise; +} + +/** @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 { + 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); +} diff --git a/packages/agent/src/harness/session/scanning-session-search.ts b/packages/agent/src/harness/session/search.ts similarity index 53% rename from packages/agent/src/harness/session/scanning-session-search.ts rename to packages/agent/src/harness/session/search.ts index 72548cde5..e4eaebf18 100644 --- a/packages/agent/src/harness/session/scanning-session-search.ts +++ b/packages/agent/src/harness/session/search.ts @@ -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 = { - open(metadata: TMetadata): Promise>; +type ScanningSessionSearchSource = { list(): Promise; + open(metadata: TMetadata): Promise>; }; /** Searches canonical sessions directly and therefore has no index to maintain. */ class ScanningSessionSearch implements SessionSearch { - private readonly source: SessionSearchSource; + private readonly source: ScanningSessionSearchSource; - constructor(source: SessionSearchSource) { + constructor(source: ScanningSessionSearchSource) { this.source = source; } @@ -26,24 +28,21 @@ class ScanningSessionSearch 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( - source: SessionSearchSource, -): SessionSearch { +export function createScanningSessionSearch< + TMetadata extends SessionMetadata, + TCreateOptions extends SessionCreateOptions, + TListOptions, +>(source: Pick, "list" | "open">): SessionSearch { return new ScanningSessionSearch(source); } diff --git a/packages/agent/src/harness/session/session.ts b/packages/agent/src/harness/session/session.ts index 1a1bd7b4a..dd8d1faa1 100644 --- a/packages/agent/src/harness/session/session.ts +++ b/packages/agent/src/harness/session/session.ts @@ -421,7 +421,7 @@ export class Session { } } -/** @internal Construct sessions only through SessionRepository. */ +/** Wraps an opened storage connection for a SessionRepository implementation. */ export async function createSession( storage: SessionStorage, contextBuildOptions: SessionContextBuildOptions = {}, diff --git a/packages/agent/src/harness/types.ts b/packages/agent/src/harness/types.ts index 2d8e9aa1e..868f9b55b 100644 --- a/packages/agent/src/harness/types.ts +++ b/packages/agent/src/harness/types.ts @@ -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 { 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; } -/** 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>; - open(metadata: TMetadata): Promise>; - list(options?: TListOptions): Promise; - delete(metadata: TMetadata): Promise; - fork( - source: TMetadata, - options: TCreateOptions, - selection: SessionForkSelection, - ): Promise>; -} - export interface JsonlSessionCreateOptions extends SessionCreateOptions { cwd: string; parentSessionPath?: string; diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index e228d0002..0e47b1dbe 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -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, diff --git a/packages/agent/test/harness/agent-harness.test.ts b/packages/agent/test/harness/agent-harness.test.ts index 59f87005a..ed7b55206 100644 --- a/packages/agent/test/harness/agent-harness.test.ts +++ b/packages/agent/test/harness/agent-harness.test.ts @@ -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; getEntries(): ReturnType; }> { - 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 = { + 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, diff --git a/packages/agent/test/harness/branch-query.test.ts b/packages/agent/test/harness/branch-query.test.ts index 152c08e02..fb2f4b062 100644 --- a/packages/agent/test/harness/branch-query.test.ts +++ b/packages/agent/test/harness/branch-query.test.ts @@ -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()); diff --git a/packages/agent/test/harness/repo.test.ts b/packages/agent/test/harness/repo.test.ts index 88f5267d6..5379a77ef 100644 --- a/packages/agent/test/harness/repo.test.ts +++ b/packages/agent/test/harness/repo.test.ts @@ -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; - counter: SessionCollectionReadCounter; +function createCountingInMemorySessionBackend(): { + backend: Pick; + 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) | undefined; + let listDisposedRepository: (() => Promise) | 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", diff --git a/packages/agent/test/harness/session-backends.test.ts b/packages/agent/test/harness/session-backends.test.ts index 50f521930..7ccc27ffe 100644 --- a/packages/agent/test/harness/session-backends.test.ts +++ b/packages/agent/test/harness/session-backends.test.ts @@ -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[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 = { diff --git a/packages/agent/test/harness/session-test-utils.ts b/packages/agent/test/harness/session-test-utils.ts index 1312adc39..621ce84c5 100644 --- a/packages/agent/test/harness/session-test-utils.ts +++ b/packages/agent/test/harness/session-test-utils.ts @@ -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 { - const collection = createInMemorySessionCollection(); - return createSessionRepository({ collection }).create({ id }); + return new InMemorySessionRepository().create({ id }); } export function createUserMessage(text: string): AgentMessage { diff --git a/packages/agent/test/harness/session.test.ts b/packages/agent/test/harness/session.test.ts index 330bce824..f7ee674e9 100644 --- a/packages/agent/test/harness/session.test.ts +++ b/packages/agent/test/harness/session.test.ts @@ -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 { + 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 { - 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), }; }, () => { diff --git a/packages/agent/test/harness/sqlite-branch-cache.test.ts b/packages/agent/test/harness/sqlite-branch-cache.test.ts index b3a1d67f2..28bdfaa8c 100644 --- a/packages/agent/test/harness/sqlite-branch-cache.test.ts +++ b/packages/agent/test/harness/sqlite-branch-cache.test.ts @@ -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(); diff --git a/packages/agent/test/harness/sqlite-migrations.test.ts b/packages/agent/test/harness/sqlite-migrations.test.ts index 0ceb213e8..9056754bd 100644 --- a/packages/agent/test/harness/sqlite-migrations.test.ts +++ b/packages/agent/test/harness/sqlite-migrations.test.ts @@ -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"); diff --git a/packages/agent/test/harness/sqlite-node.test.ts b/packages/agent/test/harness/sqlite-node.test.ts index 797210659..2fda00539 100644 --- a/packages/agent/test/harness/sqlite-node.test.ts +++ b/packages/agent/test/harness/sqlite-node.test.ts @@ -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[0]) { - const collection = createSqliteSessionCollection(options); - ownedCollections.push(collection); - return createSessionRepository({ - collection, - search: createSqliteSessionSearch(options), - }); +function createSqliteFixture(options: ConstructorParameters[0]) { + const repository = new SqliteSessionRepository(options); + ownedRepositories.push(repository); + return { repository, search: createSqliteSessionSearch(options) }; } -function createJsonlRepository(options: Parameters[0]) { - const collection = createJsonlSessionCollection(options); - ownedCollections.push(collection); - return createSessionRepository({ collection, search: createScanningSessionSearch(collection) }); +function createJsonlFixture(options: ConstructorParameters[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 = { @@ -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" }]); }); }); diff --git a/packages/agent/test/scratch/simple.ts b/packages/agent/test/scratch/simple.ts index 6433c7261..227b3d6e3 100644 --- a/packages/agent/test/scratch/simple.ts +++ b/packages/agent/test/scratch/simple.ts @@ -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, diff --git a/packages/storage/sqlite-node/CHANGELOG.md b/packages/storage/sqlite-node/CHANGELOG.md index 17c49d944..a6ef87e7d 100644 --- a/packages/storage/sqlite-node/CHANGELOG.md +++ b/packages/storage/sqlite-node/CHANGELOG.md @@ -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 diff --git a/packages/storage/sqlite-node/README.md b/packages/storage/sqlite-node/README.md index c9a9d19f9..8e46aa03e 100644 --- a/packages/storage/sqlite-node/README.md +++ b/packages/storage/sqlite-node/README.md @@ -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. diff --git a/packages/storage/sqlite-node/src/sqlite/index.ts b/packages/storage/sqlite-node/src/sqlite/index.ts index b6e8f4c47..1c2b85199 100644 --- a/packages/storage/sqlite-node/src/sqlite/index.ts +++ b/packages/storage/sqlite-node/src/sqlite/index.ts @@ -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"; diff --git a/packages/storage/sqlite-node/src/sqlite/session-collection.ts b/packages/storage/sqlite-node/src/sqlite/repo.ts similarity index 82% rename from packages/storage/sqlite-node/src/sqlite/session-collection.ts rename to packages/storage/sqlite-node/src/sqlite/repo.ts index 60a010883..f38e48dc5 100644 --- a/packages/storage/sqlite-node/src/sqlite/session-collection.ts +++ b/packages/storage/sqlite-node/src/sqlite/repo.ts @@ -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 { 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 -{ - 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(); - 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 { @@ -275,8 +278,50 @@ class SqliteSessionCollection } } -export function createSqliteSessionCollection( - options: SqliteSessionCollectionOptions, -): SessionCollection { - return new SqliteSessionCollection(options); +export interface SqliteSessionRepositoryOptions extends SqliteSessionBackendOptions { + contextBuildOptions?: SessionContextBuildOptions; +} + +export class SqliteSessionRepository + implements SessionRepository +{ + 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> { + return createSession(await this.backend.create(options), this.contextBuildOptions); + } + + async open(metadata: SqliteSessionMetadata): Promise> { + return createSession(await this.backend.open(metadata), this.contextBuildOptions); + } + + async list(options?: SqliteSessionListOptions): Promise { + return await this.backend.list(options); + } + + async delete(metadata: SqliteSessionMetadata): Promise { + await this.backend.delete(metadata); + } + + async fork( + source: SqliteSessionMetadata, + options: SessionForkOptions & SqliteSessionCreateOptions, + ): Promise> { + const { entryId: _entryId, position: _position, ...createOptions } = options; + return createSession( + await this.backend.fork(source, createOptions, createSessionForkSelection(options)), + this.contextBuildOptions, + ); + } + + async [Symbol.asyncDispose](): Promise { + await this.backend[Symbol.asyncDispose](); + } } diff --git a/packages/storage/sqlite-node/src/sqlite/search-backend.ts b/packages/storage/sqlite-node/src/sqlite/search-backend.ts index 63722b5f8..6ba2e865f 100644 --- a/packages/storage/sqlite-node/src/sqlite/search-backend.ts +++ b/packages/storage/sqlite-node/src/sqlite/search-backend.ts @@ -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 { } export interface SqliteSessionSearchOptions { - env: Pick; + env: Pick; sqlite: SqliteDatabaseFactory; databasePath: string; } diff --git a/packages/storage/sqlite-node/src/sqlite/types.ts b/packages/storage/sqlite-node/src/sqlite/types.ts index 72e3951b8..5624ccc43 100644 --- a/packages/storage/sqlite-node/src/sqlite/types.ts +++ b/packages/storage/sqlite-node/src/sqlite/types.ts @@ -44,4 +44,4 @@ export interface SqliteSessionListOptions { cwd?: string; } -export type SqliteSessionCollectionEnv = Pick; +export type SqliteSessionRepositoryEnv = Pick; diff --git a/scripts/browser-smoke-entry.ts b/scripts/browser-smoke-entry.ts index 371211481..dd8f064c7 100644 --- a/scripts/browser-smoke-entry.ts +++ b/scripts/browser-smoke-entry.ts @@ -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]);