diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index e3ce3dde2..9f3aa8396 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -4,8 +4,9 @@ ### Breaking Changes -- Changed `Session` into the sole opened-session aggregate and replaced `SessionStorage`, `SessionRepo`, and concrete per-session persistence classes with a non-owning `SessionRepository` and caller-owned, async-disposable `SessionStore` instances. Create stores with `createInMemorySessionStore()` or `createJsonlSessionStore()`, compose them with `createSessionRepository({ store, search: createScanningSessionSearch(store) })`, and dispose the store after draining harness and session work. -- `Session` instances are now created by `SessionRepository`; direct construction from an independently supplied store and snapshot was removed. +- 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. ### Added diff --git a/packages/agent/docs/agent-harness.md b/packages/agent/docs/agent-harness.md index 0de721497..d22fe878e 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({ store, 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 `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 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 9ca502c6b..d2dcec917 100644 --- a/packages/agent/docs/harness.md +++ b/packages/agent/docs/harness.md @@ -1993,38 +1993,45 @@ 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 stores +## 14. Session collections -`Session` is the only opened-session object. `SessionRepository` borrows a -caller-owned `SessionStore`, opens readers, and constructs `Session` objects. -The in-memory store is the simplest built-in option: +`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: ```ts -await using store = createInMemorySessionStore(); -const search = createScanningSessionSearch(store); -const repository = createSessionRepository({ store, search }); +await using collection = createInMemorySessionCollection(); +const search = createScanningSessionSearch(collection); +const repository = createSessionRepository({ collection, search }); const session = await repository.create({}); ``` -Use `createJsonlSessionStore({ fs, sessionsRoot })` for filesystem persistence; +Use `createJsonlSessionCollection({ fs, sessionsRoot })` for filesystem persistence; the same scanning search composes with it. Omitting `search` is valid, but `repository.search()` then returns no hits. SQLite support is provided -separately by `createSqliteSessionStore()` and should use +separately by `createSqliteSessionCollection()` and should use `createSqliteSessionSearch(options)` against the same canonical database. Drain -all session or harness operations before the store leaves +all session or harness operations before the collection leaves scope; repositories and sessions do not dispose it. -A custom store implements `SessionStore`: +A custom backend implements `SessionCollection`; its lifecycle methods return one +coherent `SessionStorage` for each opened session: ```ts -interface SessionReader { +interface SessionStorage { readonly metadata: TMetadata; readHead(): Promise; readEntry(id: string): Promise; readEntries(options?: SessionEntryCursorOptions): Promise; + appendEntry(entry: SessionTreeEntry): Promise; + findEntriesOnBranch(query: SessionBranchQuery & { start: string | null }): Promise; readPathToRootOrCompaction( leafId: string | null, ): Promise; + getLabel(id: string): Promise; + getName(): Promise; + getStats(): Promise; } type SessionForkSelection = @@ -2032,48 +2039,50 @@ type SessionForkSelection = | { kind: "before_user_message"; entryId: string } | { kind: "through_entry"; entryId: string }; -interface SessionStore< +interface SessionCollection< TMetadata extends SessionMetadata = SessionMetadata, TCreateOptions extends SessionCreateOptions = SessionCreateOptions, TListOptions = void, > extends AsyncDisposable { - create(options: TCreateOptions): Promise>; - load(metadata: TMetadata): Promise>; + create(options: TCreateOptions): Promise>; + open(metadata: TMetadata): Promise>; list(options?: TListOptions): Promise; - appendEntry(metadata: TMetadata, entry: SessionTreeEntry): Promise; delete(metadata: TMetadata): Promise; fork( source: TMetadata, options: TCreateOptions, selection: SessionForkSelection, - ): Promise>; + ): Promise>; } ``` -`create()` and `load()` return a `SessionReader` with canonical metadata, head, -entry, range, and branch-path reads. Reader lifetimes belong to the store. -`appendEntry()` receives a complete entry and must preserve append order and -entry-id uniqueness. `SessionRepository` converts public fork options into a +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 store, allowing backend-native copies without repository-level +the owning collection, allowing backend-native copies without repository-level history materialization. -Disposal rejects new operations and reader calls, drains accepted writes, and -releases owned resources. Store implementations know nothing about harness +Disposal rejects new operations and storage calls, drains accepted writes, and +releases owned resources. Collection implementations know nothing about harness operations, queues, or recovery. -`Session.buildContext()` reads only the active path. Complete history is read on -demand by `getEntries()` and by derived queries that currently require full -hydration, such as session statistics, labels, and names. -Expose custom implementations through a factory returning `SessionStore` rather +`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 stores: +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. -- Reader methods return immutable entry arrays; callers cannot mutate stored state. +- 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 store: one SQLite database can host many sessions, all writable concurrently — each through its own single live harness. The same applies to a directory of JSONL files. +- 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. ### JSONL @@ -2124,7 +2133,7 @@ CREATE INDEX idx_harness_ref_kind_seq ON harness_entries(session_id, ref, type, ### Append failure -Any store append failure faults the harness (section 4): the instance stops, in-flight calls resolve `faulted`, and the log remains a valid prefix. For SQLite, a failed transaction rolls back cleanly; for JSONL, a partial line becomes the torn tail the next open repairs. +Any storage append failure faults the harness (section 4): the instance stops, in-flight calls resolve `faulted`, and the log remains a valid prefix. For SQLite, a failed transaction rolls back cleanly; for JSONL, a partial line becomes the torn tail the next open repairs. ## 15. Telemetry @@ -2361,10 +2370,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 store contract, event/hook types. +8. `packages/agent/src/harness/types.ts` — entry union, session collection contract, event/hook types. 9. `packages/agent/src/harness/session/session.ts` — Session, context build, entry creation. -10. `packages/agent/src/harness/session/jsonl-store.ts` — JSONL session store and v3 document codec. -11. `packages/agent/src/harness/session/memory-store.ts` — in-memory session store. +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. 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. @@ -2378,7 +2387,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-store.ts` — create/open/fork. +21. `packages/storage/sqlite-node/src/sqlite/session-collection.ts` — create/open/fork. Behavioral tests (compatibility requirements): @@ -2386,5 +2395,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-stores.test.ts` +26. `packages/agent/test/harness/session-collections.test.ts` 27. `packages/agent/test/harness/sqlite-migrations.test.ts` diff --git a/packages/agent/src/harness/session/array-session-index.ts b/packages/agent/src/harness/session/array-session-index.ts new file mode 100644 index 000000000..248982abf --- /dev/null +++ b/packages/agent/src/harness/session/array-session-index.ts @@ -0,0 +1,187 @@ +import type { + SessionBranchQuery, + SessionEntryCursorOptions, + SessionHead, + SessionStats, + SessionTreeEntry, +} from "../types.ts"; +import { SessionError } from "../types.ts"; + +interface SessionEntryProjection { + name: string | undefined; + labelsById: Map; + stats: SessionStats; +} + +function createProjection(): SessionEntryProjection { + return { + name: undefined, + labelsById: new Map(), + stats: { messageCount: 0, cachedTokens: 0, uncachedTokens: 0, totalTokens: 0, costTotal: 0 }, + }; +} + +function applyProjection(projection: SessionEntryProjection, entry: SessionTreeEntry): void { + if (entry.type === "session_info") { + projection.name = entry.name?.trim() || undefined; + } else if (entry.type === "label") { + const label = entry.label?.trim(); + if (label) projection.labelsById.set(entry.targetId, label); + else projection.labelsById.delete(entry.targetId); + } + if (entry.type === "message") projection.stats.messageCount += 1; + const usage = + entry.type === "message" + ? entry.message.role === "assistant" + ? entry.message.usage + : undefined + : entry.type === "compaction" || entry.type === "branch_summary" + ? entry.usage + : undefined; + if ( + !usage || + typeof usage.input !== "number" || + typeof usage.output !== "number" || + typeof usage.cacheRead !== "number" || + typeof usage.cacheWrite !== "number" || + typeof usage.cost?.total !== "number" + ) { + return; + } + projection.stats.cachedTokens += usage.cacheRead; + projection.stats.uncachedTokens += usage.input + usage.cacheWrite; + projection.stats.totalTokens += usage.input + usage.output + usage.cacheRead + usage.cacheWrite; + projection.stats.costTotal += usage.cost.total; +} + +/** Ordered entries and derived projections for array-backed session storage. */ +export class ArraySessionIndex { + private entries: SessionTreeEntry[] = []; + private byId = new Map(); + private leafId: string | null = null; + private projection = createProjection(); + + constructor(entries: readonly SessionTreeEntry[] = []) { + this.replace(entries); + } + + has(id: string): boolean { + return this.byId.has(id); + } + + append(entry: SessionTreeEntry): void { + if (this.byId.has(entry.id)) { + throw new SessionError("invalid_entry", `Entry ${entry.id} already exists`); + } + this.entries.push(entry); + this.byId.set(entry.id, entry); + this.leafId = entry.type === "leaf" ? entry.targetId : entry.id; + applyProjection(this.projection, entry); + } + + replace(entries: readonly SessionTreeEntry[]): void { + const nextEntries = [...entries]; + const nextById = new Map(); + const nextProjection = createProjection(); + let nextLeafId: string | null = null; + for (const entry of nextEntries) { + if (nextById.has(entry.id)) { + throw new SessionError("invalid_entry", `Entry ${entry.id} already exists`); + } + nextById.set(entry.id, entry); + nextLeafId = entry.type === "leaf" ? entry.targetId : entry.id; + applyProjection(nextProjection, entry); + } + this.entries = nextEntries; + this.byId = nextById; + this.leafId = nextLeafId; + this.projection = nextProjection; + } + + readHead(): SessionHead { + if (this.leafId !== null && !this.byId.has(this.leafId)) { + throw new SessionError("invalid_session", `Entry ${this.leafId} not found`); + } + return { leafId: this.leafId }; + } + + readEntry(id: string): SessionTreeEntry | undefined { + return this.byId.get(id); + } + + readEntries(options?: SessionEntryCursorOptions): readonly SessionTreeEntry[] { + const start = options?.afterEntrySeq ?? 0; + const end = options?.limit === undefined ? undefined : start + options.limit; + return this.entries.slice(start, end); + } + + findEntriesOnBranch(query: SessionBranchQuery & { start: string | null }): readonly SessionTreeEntry[] { + if (query.limit !== undefined && (!Number.isInteger(query.limit) || query.limit <= 0)) { + throw new RangeError("Session branch query limit must be a positive integer"); + } + if (query.start === null) return []; + const pathFromStart: SessionTreeEntry[] = []; + const visited = new Set(); + let current = this.byId.get(query.start); + if (!current) throw new SessionError("not_found", `Entry ${query.start} not found`); + while (current) { + if (visited.has(current.id)) { + throw new SessionError("invalid_session", `Session branch contains a cycle at ${current.id}`); + } + visited.add(current.id); + pathFromStart.push(current); + if (query.order !== "oldestFirst" && (current.id === query.stopAtId || current.type === query.stopAtType)) { + break; + } + if (!current.parentId) break; + const parent = this.byId.get(current.parentId); + if (!parent) throw new SessionError("invalid_session", `Entry ${current.parentId} not found`); + current = parent; + } + const traversal = query.order === "oldestFirst" ? pathFromStart.reverse() : pathFromStart; + const stopIndex = + query.order === "oldestFirst" + ? traversal.findIndex((entry) => entry.id === query.stopAtId || entry.type === query.stopAtType) + : -1; + const bounded = stopIndex === -1 ? traversal : traversal.slice(0, stopIndex + 1); + const entries = bounded.filter( + (entry) => + (query.type === undefined || entry.type === query.type) && + (query.customType === undefined || (entry.type === "custom" && entry.customType === query.customType)), + ); + return query.limit === undefined ? entries : entries.slice(0, query.limit); + } + + getLabel(id: string): string | undefined { + return this.projection.labelsById.get(id); + } + + getName(): string | undefined { + return this.projection.name; + } + + getStats(): SessionStats { + return { ...this.projection.stats }; + } + + readPathToRootOrCompaction(requestedLeafId: string | null): readonly SessionTreeEntry[] { + if (requestedLeafId === null) return []; + const path: SessionTreeEntry[] = []; + let stopAtEntryId: string | null = null; + let current = this.byId.get(requestedLeafId); + if (!current) throw new SessionError("not_found", `Entry ${requestedLeafId} not found`); + while (current) { + path.push(current); + if (stopAtEntryId !== null && current.id === stopAtEntryId) break; + if (current.type === "compaction") { + if (current.retainedTail) break; + stopAtEntryId = current.firstKeptEntryId ?? null; + } + if (!current.parentId) break; + const parent = this.byId.get(current.parentId); + if (!parent) throw new SessionError("invalid_session", `Entry ${current.parentId} not found`); + current = parent; + } + return path.reverse(); + } +} diff --git a/packages/agent/src/harness/session/array-session-reader.ts b/packages/agent/src/harness/session/array-session-reader.ts deleted file mode 100644 index 696a7b14f..000000000 --- a/packages/agent/src/harness/session/array-session-reader.ts +++ /dev/null @@ -1,107 +0,0 @@ -import type { SessionEntryCursorOptions, SessionMetadata, SessionReader, SessionTreeEntry } from "../types.ts"; -import { SessionError } from "../types.ts"; - -export function createArraySessionReader( - metadata: TMetadata, - readCurrentEntries: () => readonly SessionTreeEntry[], -): SessionReader { - let indexedCount = 0; - let indexedEntries: readonly SessionTreeEntry[] | undefined; - let leafId: string | null = null; - const byId = new Map(); - - const updateIndex = (): readonly SessionTreeEntry[] => { - const entries = readCurrentEntries(); - if (entries !== indexedEntries || entries.length < indexedCount) { - indexedCount = 0; - leafId = null; - byId.clear(); - indexedEntries = entries; - } - for (; indexedCount < entries.length; indexedCount++) { - const entry = entries[indexedCount]!; - byId.set(entry.id, entry); - leafId = entry.type === "leaf" ? entry.targetId : entry.id; - } - return entries; - }; - - return { - metadata, - async readHead() { - updateIndex(); - if (leafId !== null && !byId.has(leafId)) { - throw new SessionError("invalid_session", `Entry ${leafId} not found`); - } - return { leafId }; - }, - async readEntry(id) { - updateIndex(); - return byId.get(id); - }, - async readEntries(options?: SessionEntryCursorOptions) { - const entries = updateIndex(); - const start = options?.afterEntrySeq ?? 0; - const end = options?.limit === undefined ? undefined : start + options.limit; - return entries.slice(start, end); - }, - async findEntriesOnBranch(query) { - updateIndex(); - if (query.limit !== undefined && (!Number.isInteger(query.limit) || query.limit <= 0)) { - throw new RangeError("Session branch query limit must be a positive integer"); - } - if (query.start === null) return []; - const pathFromStart: SessionTreeEntry[] = []; - const visited = new Set(); - let current = byId.get(query.start); - if (!current) throw new SessionError("not_found", `Entry ${query.start} not found`); - while (current) { - if (visited.has(current.id)) { - throw new SessionError("invalid_session", `Session branch contains a cycle at ${current.id}`); - } - visited.add(current.id); - pathFromStart.push(current); - if (query.order !== "oldestFirst" && (current.id === query.stopAtId || current.type === query.stopAtType)) { - break; - } - if (!current.parentId) break; - const parent = byId.get(current.parentId); - if (!parent) throw new SessionError("invalid_session", `Entry ${current.parentId} not found`); - current = parent; - } - const traversal = query.order === "oldestFirst" ? pathFromStart.reverse() : pathFromStart; - const stopIndex = - query.order === "oldestFirst" - ? traversal.findIndex((entry) => entry.id === query.stopAtId || entry.type === query.stopAtType) - : -1; - const bounded = stopIndex === -1 ? traversal : traversal.slice(0, stopIndex + 1); - const entries = bounded.filter( - (entry) => - (query.type === undefined || entry.type === query.type) && - (query.customType === undefined || (entry.type === "custom" && entry.customType === query.customType)), - ); - return query.limit === undefined ? entries : entries.slice(0, query.limit); - }, - async readPathToRootOrCompaction(requestedLeafId) { - updateIndex(); - if (requestedLeafId === null) return []; - const path: SessionTreeEntry[] = []; - let stopAtEntryId: string | null = null; - let current = byId.get(requestedLeafId); - if (!current) throw new SessionError("not_found", `Entry ${requestedLeafId} not found`); - while (current) { - path.push(current); - if (stopAtEntryId !== null && current.id === stopAtEntryId) break; - if (current.type === "compaction") { - if (current.retainedTail) break; - stopAtEntryId = current.firstKeptEntryId ?? null; - } - if (!current.parentId) break; - const parent = byId.get(current.parentId); - if (!parent) throw new SessionError("invalid_session", `Entry ${current.parentId} not found`); - current = parent; - } - return path.reverse(); - }, - }; -} diff --git a/packages/agent/src/harness/session/fork.ts b/packages/agent/src/harness/session/fork.ts index 51d6fd69b..8dc79e5f7 100644 --- a/packages/agent/src/harness/session/fork.ts +++ b/packages/agent/src/harness/session/fork.ts @@ -1,6 +1,14 @@ -import type { SessionForkOptions, SessionForkSelection, SessionReader, SessionTreeEntry } from "../types.ts"; +import type { SessionForkOptions, SessionForkSelection, SessionTreeEntry } from "../types.ts"; import { SessionError } from "../types.ts"; +type MaybePromise = T | 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" @@ -8,16 +16,17 @@ export function createSessionForkSelection(options: SessionForkOptions): Session : { kind: "before_user_message", entryId: options.entryId }; } +/** @internal Shared fork selection validation for built-in collections. */ export async function readSessionEntriesForFork( - reader: SessionReader, + source: SessionForkEntrySource, selection: SessionForkSelection, ): Promise { - if (selection.kind === "all") return reader.readEntries(); - const target = await reader.readEntry(selection.entryId); + 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 reader.readPathToRootOrCompaction(target.id); + 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 reader.readPathToRootOrCompaction(target.parentId); + return source.readPathToRootOrCompaction(target.parentId); } diff --git a/packages/agent/src/harness/session/jsonl-store.ts b/packages/agent/src/harness/session/jsonl-collection.ts similarity index 75% rename from packages/agent/src/harness/session/jsonl-store.ts rename to packages/agent/src/harness/session/jsonl-collection.ts index 455548cdc..5e68b47c3 100644 --- a/packages/agent/src/harness/session/jsonl-store.ts +++ b/packages/agent/src/harness/session/jsonl-collection.ts @@ -3,24 +3,24 @@ import type { JsonlSessionCreateOptions, JsonlSessionListOptions, JsonlSessionMetadata, + SessionCollection, SessionForkSelection, - SessionReader, - SessionStore, + SessionStorage, SessionTreeEntry, } from "../types.ts"; import { SessionError, toError } from "../types.ts"; -import { createArraySessionReader } from "./array-session-reader.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"; -export interface JsonlSessionStoreOptions { - fs: JsonlSessionStoreFileSystem; +export interface JsonlSessionCollectionOptions { + fs: JsonlSessionCollectionFileSystem; sessionsRoot: string; /** Maximum active operations across session keys. Defaults to 4. */ maxConcurrentOperations?: number; } -export type JsonlSessionStoreFileSystem = Pick< +export type JsonlSessionCollectionFileSystem = Pick< FileSystem, | "absolutePath" | "joinPath" @@ -117,7 +117,13 @@ function parseEntry(line: string, path: string, lineNumber: number): SessionTree } if (typeof value !== "object" || value === null) throw invalidEntry(path, lineNumber, "is not a valid session entry"); - const entry = value as { type?: unknown; id?: unknown; parentId?: unknown; timestamp?: unknown; targetId?: unknown }; + const entry = value as { + type?: unknown; + id?: unknown; + parentId?: unknown; + timestamp?: unknown; + targetId?: unknown; + }; if (typeof entry.type !== "string") throw invalidEntry(path, lineNumber, "is missing entry type"); if (typeof entry.id !== "string" || !entry.id) throw invalidEntry(path, lineNumber, "is missing entry id"); if (entry.parentId !== null && typeof entry.parentId !== "string") @@ -193,20 +199,19 @@ function createDocumentDescriptor(options: JsonlSessionCreateOptions): SessionDo }; } -class JsonlSessionStore - implements SessionStore +class JsonlSessionCollection + implements SessionCollection { - private readonly fs: JsonlSessionStoreFileSystem; + private readonly fs: JsonlSessionCollectionFileSystem; private readonly sessionsRootInput: string; private sessionsRoot: string | undefined; - private readonly entryIdsByPath = new Map>(); - private readonly entriesByPath = new Map(); + private readonly entryIndexesByPath = new Map(); private readonly operationKeysByPath = new Map(); private readonly operations: KeyedOperationQueue; private disposed = false; private disposePromise: Promise | undefined; - constructor(options: JsonlSessionStoreOptions) { + constructor(options: JsonlSessionCollectionOptions) { this.fs = options.fs; this.sessionsRootInput = options.sessionsRoot; this.operations = new KeyedOperationQueue({ @@ -214,29 +219,32 @@ class JsonlSessionStore }); } - create(options: JsonlSessionCreateOptions): Promise> { + create(options: JsonlSessionCreateOptions): Promise> { this.assertOpen(); const descriptor = createDocumentDescriptor(options); - return this.operations.enqueue(descriptor.operationKey, () => - this.createDocument(descriptor, options, options.parentSessionPath, options.metadata, []), + return this.operations.enqueue(descriptor.operationKey, async () => + this.storage(await this.createDocument(descriptor, options, options.parentSessionPath, options.metadata, [])), ); } - load(metadata: JsonlSessionMetadata): Promise> { + open(metadata: JsonlSessionMetadata): Promise> { this.assertOpen(); - return this.operations.enqueue(this.operationKey(metadata), () => this.loadDocument(metadata)); + return this.operations.enqueue(this.operationKey(metadata), async () => + this.storage(await this.loadDocument(metadata)), + ); } - private async loadDocument(metadata: JsonlSessionMetadata): Promise> { + private async loadDocument(metadata: JsonlSessionMetadata): Promise { if ( !getFileSystemResultOrThrow(await this.fs.exists(metadata.path), `Failed to check session ${metadata.path}`) ) { throw new SessionError("not_found", `Session not found: ${metadata.path}`); } const document = await loadJsonlSession(this.fs, metadata.path); - this.entriesByPath.set(metadata.path, document.entries); - this.entryIdsByPath.set(metadata.path, new Set(document.entries.map((entry) => entry.id))); - return this.reader(document.metadata); + const entries = this.entryIndexesByPath.get(metadata.path); + if (entries) entries.replace(document.entries); + else this.entryIndexesByPath.set(metadata.path, new ArraySessionIndex(document.entries)); + return document.metadata; } list(options: JsonlSessionListOptions = {}): Promise { @@ -259,7 +267,7 @@ class JsonlSessionStore return sessions.sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()); } - appendEntry(metadata: JsonlSessionMetadata, entry: SessionTreeEntry): Promise { + private appendEntry(metadata: JsonlSessionMetadata, entry: SessionTreeEntry): Promise { this.assertOpen(); return this.operations.enqueue(this.operationKey(metadata), async () => { if ( @@ -267,19 +275,17 @@ class JsonlSessionStore ) { throw new SessionError("not_found", `Session not found: ${metadata.path}`); } - let entryIds = this.entryIdsByPath.get(metadata.path); - if (!entryIds) { + let entries = this.entryIndexesByPath.get(metadata.path); + if (!entries) { await this.loadDocument(metadata); - entryIds = this.entryIdsByPath.get(metadata.path)!; + entries = this.entryIndexesByPath.get(metadata.path)!; } - if (entryIds.has(entry.id)) throw new SessionError("invalid_entry", `Entry ${entry.id} already exists`); + if (entries.has(entry.id)) throw new SessionError("invalid_entry", `Entry ${entry.id} already exists`); getFileSystemResultOrThrow( await this.fs.appendFile(metadata.path, `${JSON.stringify(entry)}\n`), `Failed to append session entry ${entry.id}`, ); - entryIds.add(entry.id); - this.entryIdsByPath.set(metadata.path, entryIds); - this.entriesByPath.get(metadata.path)!.push(entry); + entries.append(entry); }); } @@ -290,8 +296,7 @@ class JsonlSessionStore await this.fs.remove(metadata.path, { force: true }), `Failed to delete session ${metadata.path}`, ); - this.entryIdsByPath.delete(metadata.path); - this.entriesByPath.delete(metadata.path); + this.entryIndexesByPath.delete(metadata.path); this.operationKeysByPath.delete(metadata.path); }); } @@ -300,7 +305,7 @@ class JsonlSessionStore source: JsonlSessionMetadata, options: JsonlSessionCreateOptions, selection: SessionForkSelection, - ): Promise> { + ): Promise> { this.assertOpen(); const descriptor = createDocumentDescriptor(options); const sourceEntries = this.operations.enqueue(this.operationKey(source), async () => { @@ -308,20 +313,20 @@ class JsonlSessionStore throw new SessionError("not_found", `Session not found: ${source.path}`); } const document = await loadJsonlSession(this.fs, source.path); - this.entriesByPath.set(source.path, document.entries); - this.entryIdsByPath.set(source.path, new Set(document.entries.map((entry) => entry.id))); - return readSessionEntriesForFork( - createArraySessionReader(document.metadata, () => document.entries), - selection, - ); + const entries = this.entryIndexesByPath.get(source.path); + if (entries) entries.replace(document.entries); + else this.entryIndexesByPath.set(source.path, new ArraySessionIndex(document.entries)); + return readSessionEntriesForFork(this.entryIndexesByPath.get(source.path)!, selection); }); return this.operations.enqueue(descriptor.operationKey, async () => - this.createDocument( - descriptor, - options, - options.parentSessionPath ?? source.path, - options.metadata ?? source.metadata, - await sourceEntries, + this.storage( + await this.createDocument( + descriptor, + options, + options.parentSessionPath ?? source.path, + options.metadata ?? source.metadata, + await sourceEntries, + ), ), ); } @@ -335,7 +340,7 @@ class JsonlSessionStore } private assertOpen(): void { - if (this.disposed) throw new SessionError("storage", "JSONL session store is disposed"); + if (this.disposed) throw new SessionError("storage", "JSONL session collection is disposed"); } private operationKey(metadata: JsonlSessionMetadata): string { @@ -348,7 +353,7 @@ class JsonlSessionStore parentSessionPath: string | undefined, metadata: Record | undefined, entries: readonly SessionTreeEntry[], - ): Promise> { + ): Promise { const dir = await this.getSessionDir(options.cwd); getFileSystemResultOrThrow( await this.fs.createDir(dir, { recursive: true }), @@ -372,45 +377,38 @@ class JsonlSessionStore }; const content = [JSON.stringify(header), ...entries.map((entry) => JSON.stringify(entry)), ""].join("\n"); getFileSystemResultOrThrow(await this.fs.writeFile(path, content), `Failed to create session ${path}`); - const storedEntries = [...entries]; - this.entriesByPath.set(path, storedEntries); - this.entryIdsByPath.set(path, new Set(storedEntries.map((entry) => entry.id))); + this.entryIndexesByPath.set(path, new ArraySessionIndex(entries)); this.operationKeysByPath.set(path, descriptor.operationKey); - return this.reader(metadataFromHeader(header, path)); + return metadataFromHeader(header, path); } - private reader(metadata: JsonlSessionMetadata): SessionReader { - const reader = createArraySessionReader(metadata, () => { - const entries = this.entriesByPath.get(metadata.path); - if (!entries) throw new SessionError("not_found", `Session not found: ${metadata.path}`); - return entries; - }); - const operationKey = this.operationKey(metadata); + private readIndex(metadata: JsonlSessionMetadata, read: (entries: ArraySessionIndex) => T): Promise { + this.assertOpen(); + return this.operations.enqueue(this.operationKey(metadata), () => read(this.entryIndex(metadata.path))); + } + + private storage(metadata: JsonlSessionMetadata): SessionStorage { return { - metadata: reader.metadata, - readHead: () => { - this.assertOpen(); - return this.operations.enqueue(operationKey, () => reader.readHead()); - }, - readEntry: (id) => { - this.assertOpen(); - return this.operations.enqueue(operationKey, () => reader.readEntry(id)); - }, - readEntries: (options) => { - this.assertOpen(); - return this.operations.enqueue(operationKey, () => reader.readEntries(options)); - }, - findEntriesOnBranch: (query) => { - this.assertOpen(); - return this.operations.enqueue(operationKey, () => reader.findEntriesOnBranch(query)); - }, - readPathToRootOrCompaction: (leafId) => { - this.assertOpen(); - return this.operations.enqueue(operationKey, () => reader.readPathToRootOrCompaction(leafId)); - }, + metadata, + readHead: () => this.readIndex(metadata, (entries) => entries.readHead()), + readEntry: (id) => this.readIndex(metadata, (entries) => entries.readEntry(id)), + readEntries: (options) => this.readIndex(metadata, (entries) => entries.readEntries(options)), + appendEntry: (entry) => this.appendEntry(metadata, entry), + findEntriesOnBranch: (query) => this.readIndex(metadata, (entries) => entries.findEntriesOnBranch(query)), + readPathToRootOrCompaction: (leafId) => + this.readIndex(metadata, (entries) => entries.readPathToRootOrCompaction(leafId)), + getLabel: (id) => this.readIndex(metadata, (entries) => entries.getLabel(id)), + getName: () => this.readIndex(metadata, (entries) => entries.getName()), + getStats: () => this.readIndex(metadata, (entries) => entries.getStats()), }; } + private entryIndex(path: string): ArraySessionIndex { + const entries = this.entryIndexesByPath.get(path); + if (!entries) throw new SessionError("not_found", `Session not found: ${path}`); + return entries; + } + private async getSessionsRoot(): Promise { this.sessionsRoot ??= getFileSystemResultOrThrow( await this.fs.absolutePath(this.sessionsRootInput), @@ -435,8 +433,8 @@ class JsonlSessionStore } } -export function createJsonlSessionStore( - options: JsonlSessionStoreOptions, -): SessionStore { - return new JsonlSessionStore(options); +export function createJsonlSessionCollection( + options: JsonlSessionCollectionOptions, +): SessionCollection { + return new JsonlSessionCollection(options); } diff --git a/packages/agent/src/harness/session/memory-collection.ts b/packages/agent/src/harness/session/memory-collection.ts new file mode 100644 index 000000000..639c4357e --- /dev/null +++ b/packages/agent/src/harness/session/memory-collection.ts @@ -0,0 +1,128 @@ +import type { + SessionCollection, + SessionForkSelection, + SessionMetadata, + SessionStorage, + SessionTreeEntry, +} 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"; + +export type InMemorySessionCreateOptions = { id?: string }; + +interface InMemorySessionState { + metadata: SessionMetadata; + entries: ArraySessionIndex; +} + +class InMemorySessionCollection implements SessionCollection { + private readonly sessions = new Map(); + private readonly operations = new KeyedOperationQueue(); + private disposed = false; + private disposePromise: Promise | undefined; + + create(options: InMemorySessionCreateOptions = {}): Promise> { + this.assertOpen(); + const id = options.id ?? createSessionId(); + return this.operations.enqueue(id, () => { + const state: InMemorySessionState = { + metadata: { id, createdAt: createTimestamp() }, + entries: new ArraySessionIndex(), + }; + this.sessions.set(id, state); + return this.storage(state); + }); + } + + open(metadata: SessionMetadata): Promise> { + this.assertOpen(); + return this.operations.enqueue(metadata.id, () => this.storage(this.getState(metadata))); + } + + list(): Promise { + this.assertOpen(); + return this.operations.enqueueBarrier(() => [...this.sessions.values()].map((state) => state.metadata)); + } + + delete(metadata: SessionMetadata): Promise { + this.assertOpen(); + return this.operations.enqueue(metadata.id, () => { + this.sessions.delete(metadata.id); + }); + } + + fork( + source: SessionMetadata, + options: InMemorySessionCreateOptions, + selection: SessionForkSelection, + ): Promise> { + this.assertOpen(); + const id = options.id ?? createSessionId(); + const sourceEntries = this.operations.enqueue(source.id, () => + readSessionEntriesForFork(this.getState(source).entries, selection), + ); + return this.operations.enqueue(id, async () => { + const state: InMemorySessionState = { + metadata: { id, createdAt: createTimestamp() }, + entries: new ArraySessionIndex(await sourceEntries), + }; + this.sessions.set(id, state); + return this.storage(state); + }); + } + + async [Symbol.asyncDispose](): Promise { + if (!this.disposePromise) { + this.disposed = true; + this.disposePromise = this.operations.drain(); + } + await this.disposePromise; + } + + private storage(state: InMemorySessionState): SessionStorage { + const read = (operation: (entries: ArraySessionIndex) => T): Promise => { + this.assertOpen(); + return this.operations.enqueue(state.metadata.id, () => operation(this.getState(state.metadata).entries)); + }; + return { + metadata: state.metadata, + readHead: () => read((entries) => entries.readHead()), + readEntry: (id) => read((entries) => entries.readEntry(id)), + readEntries: (options) => read((entries) => entries.readEntries(options)), + appendEntry: (entry) => this.appendEntry(state.metadata, entry), + findEntriesOnBranch: (query) => read((entries) => entries.findEntriesOnBranch(query)), + readPathToRootOrCompaction: (leafId) => read((entries) => entries.readPathToRootOrCompaction(leafId)), + getLabel: (id) => read((entries) => entries.getLabel(id)), + getName: () => read((entries) => entries.getName()), + getStats: () => read((entries) => entries.getStats()), + }; + } + + private appendEntry(metadata: SessionMetadata, entry: SessionTreeEntry): Promise { + this.assertOpen(); + return this.operations.enqueue(metadata.id, () => { + this.getState(metadata).entries.append(entry); + }); + } + + private assertOpen(): void { + if (this.disposed) throw new SessionError("storage", "In-memory session collection is disposed"); + } + + private getState(metadata: SessionMetadata): InMemorySessionState { + const state = this.sessions.get(metadata.id); + if (!state) throw new SessionError("not_found", `Session not found: ${metadata.id}`); + return state; + } +} + +export function createInMemorySessionCollection(): SessionCollection< + SessionMetadata, + InMemorySessionCreateOptions, + void +> { + return new InMemorySessionCollection(); +} diff --git a/packages/agent/src/harness/session/memory-store.ts b/packages/agent/src/harness/session/memory-store.ts deleted file mode 100644 index f69a9bb6f..000000000 --- a/packages/agent/src/harness/session/memory-store.ts +++ /dev/null @@ -1,134 +0,0 @@ -import type { SessionForkSelection, SessionMetadata, SessionReader, SessionStore, SessionTreeEntry } from "../types.ts"; -import { SessionError } from "../types.ts"; -import { createArraySessionReader } from "./array-session-reader.ts"; -import { readSessionEntriesForFork } from "./fork.ts"; -import { KeyedOperationQueue } from "./keyed-operation-queue.ts"; -import { createSessionId, createTimestamp } from "./repository.ts"; - -export type InMemorySessionCreateOptions = { id?: string }; - -interface InMemorySessionState { - metadata: SessionMetadata; - entries: SessionTreeEntry[]; -} - -class InMemorySessionStore implements SessionStore { - private readonly sessions = new Map(); - private readonly operations = new KeyedOperationQueue(); - private disposed = false; - private disposePromise: Promise | undefined; - - create(options: InMemorySessionCreateOptions = {}): Promise> { - this.assertOpen(); - const id = options.id ?? createSessionId(); - return this.operations.enqueue(id, () => { - const state: InMemorySessionState = { - metadata: { id, createdAt: createTimestamp() }, - entries: [], - }; - this.sessions.set(state.metadata.id, state); - return this.reader(state); - }); - } - - load(metadata: SessionMetadata): Promise> { - this.assertOpen(); - return this.operations.enqueue(metadata.id, () => this.reader(this.getState(metadata))); - } - - list(): Promise { - this.assertOpen(); - return this.operations.enqueueBarrier(() => [...this.sessions.values()].map((state) => state.metadata)); - } - - appendEntry(metadata: SessionMetadata, entry: SessionTreeEntry): Promise { - this.assertOpen(); - return this.operations.enqueue(metadata.id, () => { - const state = this.getState(metadata); - if (state.entries.some((existing) => existing.id === entry.id)) { - throw new SessionError("invalid_entry", `Entry ${entry.id} already exists`); - } - state.entries.push(entry); - }); - } - - delete(metadata: SessionMetadata): Promise { - this.assertOpen(); - return this.operations.enqueue(metadata.id, () => { - this.sessions.delete(metadata.id); - }); - } - - fork( - source: SessionMetadata, - options: InMemorySessionCreateOptions, - selection: SessionForkSelection, - ): Promise> { - this.assertOpen(); - const id = options.id ?? createSessionId(); - const sourceEntries = this.operations.enqueue(source.id, () => { - const sourceState = this.getState(source); - return readSessionEntriesForFork( - createArraySessionReader(sourceState.metadata, () => sourceState.entries), - selection, - ); - }); - return this.operations.enqueue(id, async () => { - const state: InMemorySessionState = { - metadata: { id, createdAt: createTimestamp() }, - entries: [...(await sourceEntries)], - }; - this.sessions.set(state.metadata.id, state); - return this.reader(state); - }); - } - - async [Symbol.asyncDispose](): Promise { - if (!this.disposePromise) { - this.disposed = true; - this.disposePromise = this.operations.drain(); - } - await this.disposePromise; - } - - private assertOpen(): void { - if (this.disposed) throw new SessionError("storage", "In-memory session store is disposed"); - } - - private getState(metadata: SessionMetadata): InMemorySessionState { - const state = this.sessions.get(metadata.id); - if (!state) throw new SessionError("not_found", `Session not found: ${metadata.id}`); - return state; - } - - private reader(state: InMemorySessionState): SessionReader { - const reader = createArraySessionReader(state.metadata, () => state.entries); - return { - metadata: reader.metadata, - readHead: () => { - this.assertOpen(); - return this.operations.enqueue(state.metadata.id, () => reader.readHead()); - }, - readEntry: (id) => { - this.assertOpen(); - return this.operations.enqueue(state.metadata.id, () => reader.readEntry(id)); - }, - readEntries: (options) => { - this.assertOpen(); - return this.operations.enqueue(state.metadata.id, () => reader.readEntries(options)); - }, - findEntriesOnBranch: (query) => { - this.assertOpen(); - return this.operations.enqueue(state.metadata.id, () => reader.findEntriesOnBranch(query)); - }, - readPathToRootOrCompaction: (leafId) => { - this.assertOpen(); - return this.operations.enqueue(state.metadata.id, () => reader.readPathToRootOrCompaction(leafId)); - }, - }; - } -} - -export function createInMemorySessionStore(): SessionStore { - return new InMemorySessionStore(); -} diff --git a/packages/agent/src/harness/session/repository.ts b/packages/agent/src/harness/session/repository.ts index 36f6bd81b..54c9dbbe0 100644 --- a/packages/agent/src/harness/session/repository.ts +++ b/packages/agent/src/harness/session/repository.ts @@ -2,17 +2,17 @@ import { uuidv7 } from "@earendil-works/pi-ai"; import { type FileError, type Result, + type SessionCollection, type SessionCreateOptions, SessionError, type SessionForkOptions, type SessionMetadata, type SessionSearch, type SessionSearchHit, - type SessionStore, type SessionTreeEntry, } from "../types.ts"; import { createSessionForkSelection } from "./fork.ts"; -import { createSessionFromReader, type Session, type SessionContextBuildOptions } from "./session.ts"; +import { createSession, type Session, type SessionContextBuildOptions } from "./session.ts"; export function createSessionId(): string { return uuidv7(); @@ -27,34 +27,34 @@ export class SessionRepository< TCreateOptions extends SessionCreateOptions = SessionCreateOptions, TListOptions = void, > { - private readonly store: SessionStore; - private readonly searchStore: SessionSearch | null; + private readonly collection: SessionCollection; + private readonly sessionSearch: SessionSearch | null; private readonly contextBuildOptions: SessionContextBuildOptions; constructor(options: { - store: SessionStore; + collection: SessionCollection; search?: SessionSearch | null; contextBuildOptions?: SessionContextBuildOptions; }) { - this.store = options.store; - this.searchStore = options.search ?? null; + this.collection = options.collection; + this.sessionSearch = options.search ?? null; this.contextBuildOptions = options.contextBuildOptions ?? {}; } async create(options: TCreateOptions): Promise> { - return createSessionFromReader(this.store, await this.store.create(options), this.contextBuildOptions); + return createSession(await this.collection.create(options), this.contextBuildOptions); } async open(metadata: TMetadata): Promise> { - return createSessionFromReader(this.store, await this.store.load(metadata), this.contextBuildOptions); + return createSession(await this.collection.open(metadata), this.contextBuildOptions); } async list(options?: TListOptions): Promise { - return await this.store.list(options); + return await this.collection.list(options); } async delete(metadata: TMetadata): Promise { - await this.store.delete(metadata); + await this.collection.delete(metadata); } async fork(source: TMetadata, options: SessionForkOptions & TCreateOptions): Promise> { @@ -62,15 +62,11 @@ export class SessionRepository< const createOptions = { ...options }; delete createOptions.entryId; delete createOptions.position; - return createSessionFromReader( - this.store, - await this.store.fork(source, createOptions, selection), - this.contextBuildOptions, - ); + return createSession(await this.collection.fork(source, createOptions, selection), this.contextBuildOptions); } async search(options: Parameters["search"]>[0]): Promise[]> { - return this.searchStore ? await this.searchStore.search(options) : []; + return this.sessionSearch ? await this.sessionSearch.search(options) : []; } } @@ -79,7 +75,7 @@ export function createSessionRepository< TCreateOptions extends SessionCreateOptions = SessionCreateOptions, TListOptions = void, >(options: { - store: SessionStore; + collection: SessionCollection; search?: SessionSearch | null; contextBuildOptions?: SessionContextBuildOptions; }): SessionRepository { diff --git a/packages/agent/src/harness/session/search-backend.ts b/packages/agent/src/harness/session/search-backend.ts index 1551757c6..99b00b0d7 100644 --- a/packages/agent/src/harness/session/search-backend.ts +++ b/packages/agent/src/harness/session/search-backend.ts @@ -1,14 +1,14 @@ import type { SessionMetadata, - SessionReader, SessionSearch, SessionSearchHit, SessionSearchOptions, + SessionStorage, } from "../types.ts"; import { findSessionEntryMatches } from "./repository.ts"; type SessionSearchSource = { - load(metadata: TMetadata): Promise>; + open(metadata: TMetadata): Promise>; list(): Promise; }; @@ -25,8 +25,8 @@ 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 reader = await this.source.load(metadata); - hits.push(...findSessionEntryMatches(reader.metadata, await reader.readEntries(), options.text)); + const storage = await this.source.open(metadata); + hits.push(...findSessionEntryMatches(storage.metadata, await storage.readEntries(), options.text)); } return hits; } diff --git a/packages/agent/src/harness/session/session.ts b/packages/agent/src/harness/session/session.ts index f7dfcdf0f..1a1bd7b4a 100644 --- a/packages/agent/src/harness/session/session.ts +++ b/packages/agent/src/harness/session/session.ts @@ -16,9 +16,8 @@ import type { SessionEntryCursorOptions, SessionInfoEntry, SessionMetadata, - SessionReader, SessionStats, - SessionStore, + SessionStorage, SessionTreeEntry, ThinkingLevelChangeEntry, } from "../types.ts"; @@ -150,127 +149,21 @@ export function buildSessionContext( return { ...state, messages }; } -class HydratedSessionState { - private readonly entries: SessionTreeEntry[]; - private readonly labelsById = new Map(); - - constructor(entries: readonly SessionTreeEntry[]) { - this.entries = [...entries]; - for (const entry of this.entries) this.applyProjection(entry); - } - - getLabel(id: string): string | undefined { - return this.labelsById.get(id); - } - - getSessionName(): string | undefined { - for (let i = this.entries.length - 1; i >= 0; i--) { - const entry = this.entries[i]!; - if (entry.type === "session_info") return entry.name?.trim() || undefined; - } - return undefined; - } - - getSessionStats(): SessionStats { - let messageCount = 0; - let cachedTokens = 0; - let uncachedTokens = 0; - let totalTokens = 0; - let costTotal = 0; - for (const entry of this.entries) { - if (entry.type === "message") messageCount += 1; - const usage = - entry.type === "message" - ? entry.message.role === "assistant" - ? entry.message.usage - : undefined - : entry.type === "compaction" || entry.type === "branch_summary" - ? entry.usage - : undefined; - if ( - !usage || - typeof usage.input !== "number" || - typeof usage.output !== "number" || - typeof usage.cacheRead !== "number" || - typeof usage.cacheWrite !== "number" || - typeof usage.cost?.total !== "number" - ) - continue; - cachedTokens += usage.cacheRead; - uncachedTokens += usage.input + usage.cacheWrite; - totalTokens += usage.input + usage.output + usage.cacheRead + usage.cacheWrite; - costTotal += usage.cost.total; - } - return { messageCount, cachedTokens, uncachedTokens, totalTokens, costTotal }; - } - - private applyProjection(entry: SessionTreeEntry): void { - if (entry.type !== "label") return; - const label = entry.label?.trim(); - if (label) this.labelsById.set(entry.targetId, label); - else this.labelsById.delete(entry.targetId); - } -} - -export interface Session { - getMetadata(): Promise; - getLeafId(): Promise; - getEntry(id: string): Promise; - getEntries(options?: SessionEntryCursorOptions): Promise; - getBranch(fromId?: string | null): Promise; - findEntriesOnBranch(query?: SessionBranchQuery): Promise; - findEntryOnBranch(query?: SessionBranchQuery): Promise; - buildContextEntries(options?: SessionContextBuildOptions): Promise; - buildContext(options?: SessionContextBuildOptions): Promise; - getLabel(id: string): Promise; - getSessionStats(): Promise; - getSessionName(): Promise; - appendMessage(message: AgentMessage): Promise; - appendThinkingLevelChange(thinkingLevel: string): Promise; - appendModelChange(provider: string, modelId: string): Promise; - appendActiveToolsChange(activeToolNames: string[]): Promise; - appendCompaction( - summary: string, - firstKeptEntryId: string | undefined, - tokensBefore: number, - details?: T, - fromHook?: boolean, - usage?: Usage, - retainedTail?: AgentMessage[], - ): Promise; - appendCustomEntry(customType: string, data?: unknown): Promise; - appendCustomMessageEntry( - customType: string, - content: string | (TextContent | ImageContent)[], - display: boolean, - details?: T, - ): Promise; - appendLabel(targetId: string, label: string | undefined): Promise; - appendSessionName(name: string): Promise; - moveTo( - entryId: string | null, - summary?: { summary: string; details?: unknown; usage?: Usage; fromHook?: boolean }, - ): Promise; -} - -class StoreSession implements Session { - private readonly store: Pick, "appendEntry">; - private readonly reader: SessionReader; +export class Session { + private readonly storage: SessionStorage; private readonly metadata: TMetadata; - private hydratedStatePromise: Promise | undefined; private leafId: string | null; private readonly contextBuildOptions: SessionContextBuildOptions; private appendTail: Promise = Promise.resolve(); + /** @internal Construct sessions through SessionRepository. */ constructor( - store: Pick, "appendEntry">, - reader: SessionReader, + storage: SessionStorage, leafId: string | null, contextBuildOptions: SessionContextBuildOptions = {}, ) { - this.store = store; - this.reader = reader; - this.metadata = reader.metadata; + this.storage = storage; + this.metadata = storage.metadata; this.leafId = leafId; this.contextBuildOptions = contextBuildOptions; } @@ -282,19 +175,19 @@ class StoreSession implemen return this.leafId; } async getEntry(id: string): Promise { - return this.reader.readEntry(id); + return this.storage.readEntry(id); } async getEntries(options?: SessionEntryCursorOptions): Promise { - return [...(await this.reader.readEntries(options))]; + return [...(await this.storage.readEntries(options))]; } async getBranch(fromId?: string | null): Promise { - return [...(await this.reader.readPathToRootOrCompaction(fromId === undefined ? this.leafId : fromId))]; + return [...(await this.storage.readPathToRootOrCompaction(fromId === undefined ? this.leafId : fromId))]; } async findEntriesOnBranch(query: SessionBranchQuery = {}): Promise { return [ - ...(await this.reader.findEntriesOnBranch({ + ...(await this.storage.findEntriesOnBranch({ ...query, start: query.start === undefined ? this.leafId : query.start, })), @@ -324,18 +217,13 @@ class StoreSession implemen } async getLabel(id: string): Promise { - return (await this.getHydratedState()).getLabel(id); + return this.storage.getLabel(id); } async getSessionStats(): Promise { - return (await this.getHydratedState()).getSessionStats(); + return this.storage.getStats(); } async getSessionName(): Promise { - return (await this.getHydratedState()).getSessionName(); - } - - private async getHydratedState(): Promise { - this.hydratedStatePromise ??= this.reader.readEntries().then((entries) => new HydratedSessionState(entries)); - return this.hydratedStatePromise; + return this.storage.getName(); } private async createEntryId(): Promise { @@ -349,22 +237,21 @@ class StoreSession implemen private enqueueAppend( createEntry: (base: Pick) => TEntry, ): Promise { - const operation = this.appendTail.then(async () => { + const commit = this.appendTail.then(async () => { const entry = createEntry({ id: await this.createEntryId(), parentId: this.leafId, timestamp: new Date().toISOString(), }); - await this.store.appendEntry(this.metadata, entry); + await this.storage.appendEntry(entry); this.leafId = entry.type === "leaf" ? entry.targetId : entry.id; - this.hydratedStatePromise = undefined; return entry; }); - this.appendTail = operation.then( + this.appendTail = commit.then( () => undefined, () => undefined, ); - return operation; + return commit; } private async setLeafId(leafId: string | null): Promise { @@ -535,10 +422,9 @@ class StoreSession implemen } /** @internal Construct sessions only through SessionRepository. */ -export async function createSessionFromReader( - store: Pick, "appendEntry">, - reader: SessionReader, +export async function createSession( + storage: SessionStorage, contextBuildOptions: SessionContextBuildOptions = {}, ): Promise> { - return new StoreSession(store, reader, (await reader.readHead()).leafId, contextBuildOptions); + return new Session(storage, (await storage.readHead()).leafId, contextBuildOptions); } diff --git a/packages/agent/src/harness/types.ts b/packages/agent/src/harness/types.ts index 90ece1d33..c5f3c0886 100644 --- a/packages/agent/src/harness/types.ts +++ b/packages/agent/src/harness/types.ts @@ -555,29 +555,36 @@ export interface SessionHead { leafId: string | null; } -/** Canonical reads for one stored session. Its lifetime is owned by the containing {@link SessionStore}. */ -export interface SessionReader { +/** Complete storage contract for one opened session. Its lifetime is owned by its {@link SessionCollection}. */ +export interface SessionStorage { readonly metadata: TMetadata; /** Rejects with `invalid_session` when a non-null active leaf does not reference a stored entry. */ readHead(): Promise; readEntry(id: string): Promise; readEntries(options?: SessionEntryCursorOptions): Promise; + appendEntry(entry: SessionTreeEntry): Promise; findEntriesOnBranch(query: SessionBranchQuery & { start: string | null }): Promise; readPathToRootOrCompaction(leafId: string | null): Promise; + getLabel(id: string): Promise; + getName(): Promise; + getStats(): Promise; } -/** Owns persistence and resources shared by all sessions in a repository. */ -export interface SessionStore< +/** 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>; - load(metadata: TMetadata): Promise>; + create(options: TCreateOptions): Promise>; + open(metadata: TMetadata): Promise>; list(options?: TListOptions): Promise; - appendEntry(metadata: TMetadata, entry: SessionTreeEntry): Promise; delete(metadata: TMetadata): Promise; - fork(source: TMetadata, options: TCreateOptions, selection: SessionForkSelection): Promise>; + fork( + source: TMetadata, + options: TCreateOptions, + selection: SessionForkSelection, + ): Promise>; } export interface JsonlSessionCreateOptions extends SessionCreateOptions { diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 18edd1112..ddf2c570b 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -31,14 +31,14 @@ export * from "./harness/messages.ts"; export * from "./harness/prompt-templates.ts"; export { readSessionEntriesForFork } from "./harness/session/fork.ts"; export { - createJsonlSessionStore, - type JsonlSessionStoreFileSystem, - type JsonlSessionStoreOptions, -} from "./harness/session/jsonl-store.ts"; + createJsonlSessionCollection, + type JsonlSessionCollectionFileSystem, + type JsonlSessionCollectionOptions, +} from "./harness/session/jsonl-collection.ts"; export { - createInMemorySessionStore, + createInMemorySessionCollection, type InMemorySessionCreateOptions, -} from "./harness/session/memory-store.ts"; +} from "./harness/session/memory-collection.ts"; export * from "./harness/session/repository.ts"; export * from "./harness/session/search-backend.ts"; export { diff --git a/packages/agent/test/harness/agent-harness.test.ts b/packages/agent/test/harness/agent-harness.test.ts index fe945ae8f..59f87005a 100644 --- a/packages/agent/test/harness/agent-harness.test.ts +++ b/packages/agent/test/harness/agent-harness.test.ts @@ -11,7 +11,7 @@ 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 { createInMemorySessionStore } from "../../src/harness/session/memory-store.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 type { AgentHarnessTool, PromptTemplate, Skill } from "../../src/harness/types.ts"; @@ -97,25 +97,28 @@ async function createBlockingSession(expectedWrites: number): Promise<{ releaseWrites: ReturnType; getEntries(): ReturnType; }> { - const source = createInMemorySessionStore(); + const source = createInMemorySessionCollection(); const allWritesStarted = deferred(); const releaseWrites = deferred(); let writesStarted = 0; - const blockingStore: ReturnType = { - create: (options) => source.create(options), - load: (metadata) => source.load(metadata), - list: (options) => source.list(options), - async appendEntry(metadata, entry) { + const blockWrites = async (storage: Awaited>) => ({ + ...storage, + async appendEntry(entry: Parameters[0]) { writesStarted++; if (writesStarted === expectedWrites) allWritesStarted.resolve(); await releaseWrites.promise; - await source.appendEntry(metadata, entry); + await storage.appendEntry(entry); }, + }); + const blockingCollection: ReturnType = { + create: async (options) => blockWrites(await source.create(options)), + open: async (metadata) => blockWrites(await source.open(metadata)), + list: (options) => source.list(options), delete: (metadata) => source.delete(metadata), - fork: (metadata, options, selection) => source.fork(metadata, options, selection), + fork: async (metadata, options, selection) => blockWrites(await source.fork(metadata, options, selection)), [Symbol.asyncDispose]: () => source[Symbol.asyncDispose](), }; - const session = await createSessionRepository({ store: blockingStore }).create({}); + const session = await createSessionRepository({ collection: blockingCollection }).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 8810a27fa..152c08e02 100644 --- a/packages/agent/test/harness/branch-query.test.ts +++ b/packages/agent/test/harness/branch-query.test.ts @@ -1,17 +1,17 @@ import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; -import { createNodeSqliteFactory, createSqliteSessionStore } from "../../../storage/sqlite-node/src/index.ts"; +import { createNodeSqliteFactory, createSqliteSessionCollection } from "../../../storage/sqlite-node/src/index.ts"; import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts"; -import { createJsonlSessionStore } from "../../src/harness/session/jsonl-store.ts"; -import { createInMemorySessionStore } from "../../src/harness/session/memory-store.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 { Session } from "../../src/harness/session/session.ts"; import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts"; -const ownedStores: AsyncDisposable[] = []; +const ownedCollections: AsyncDisposable[] = []; afterEach(async () => { - for (const store of ownedStores.splice(0)) await store[Symbol.asyncDispose](); + for (const collection of ownedCollections.splice(0)) await collection[Symbol.asyncDispose](); }); async function verifyBranchQueries(session: Session): Promise<{ tail: string; fullPath: string[] }> { @@ -90,9 +90,9 @@ async function verifyBranchQueries(session: Session): Promise<{ tail: string; fu describe("bounded session branch queries", () => { it("provides identical in-memory query semantics", async () => { - const store = createInMemorySessionStore(); - ownedStores.push(store); - const repo = createSessionRepository({ store }); + const collection = createInMemorySessionCollection(); + ownedCollections.push(collection); + const repo = createSessionRepository({ collection }); const session = await repo.create({ id: "memory" }); const expected = await verifyBranchQueries(session); const reopened = await repo.open(await session.getMetadata()); @@ -102,10 +102,10 @@ describe("bounded session branch queries", () => { }); it("rejects corrupt parent chains in array-backed readers", async () => { - const store = createInMemorySessionStore(); - ownedStores.push(store); - const reader = await store.create({ id: "corrupt-memory" }); - await store.appendEntry(reader.metadata, { + const collection = createInMemorySessionCollection(); + ownedCollections.push(collection); + const storage = await collection.create({ id: "corrupt-memory" }); + await storage.appendEntry({ type: "message", id: "orphan", parentId: "missing-parent", @@ -114,30 +114,30 @@ describe("bounded session branch queries", () => { }); expect( - (await reader.findEntriesOnBranch({ start: "orphan", stopAtId: "orphan" })).map((entry) => entry.id), + (await storage.findEntriesOnBranch({ start: "orphan", stopAtId: "orphan" })).map((entry) => entry.id), ).toEqual(["orphan"]); expect( - (await reader.findEntriesOnBranch({ start: "orphan", stopAtType: "message" })).map((entry) => entry.id), + (await storage.findEntriesOnBranch({ start: "orphan", stopAtType: "message" })).map((entry) => entry.id), ).toEqual(["orphan"]); - await expect(reader.findEntriesOnBranch({ start: "orphan" })).rejects.toMatchObject({ + await expect(storage.findEntriesOnBranch({ start: "orphan" })).rejects.toMatchObject({ code: "invalid_session", message: "Entry missing-parent not found", }); - await store.appendEntry(reader.metadata, { + await storage.appendEntry({ type: "message", id: "cycle-a", parentId: "cycle-b", timestamp: "2026-01-01T00:00:01.000Z", message: createUserMessage("a"), }); - await store.appendEntry(reader.metadata, { + await storage.appendEntry({ type: "message", id: "cycle-b", parentId: "cycle-a", timestamp: "2026-01-01T00:00:02.000Z", message: createUserMessage("b"), }); - await expect(reader.findEntriesOnBranch({ start: "cycle-b" })).rejects.toMatchObject({ + await expect(storage.findEntriesOnBranch({ start: "cycle-b" })).rejects.toMatchObject({ code: "invalid_session", message: "Session branch contains a cycle at cycle-b", }); @@ -145,9 +145,9 @@ describe("bounded session branch queries", () => { it("provides identical JSONL query semantics", async () => { const root = createTempDir(); - const store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); - ownedStores.push(store); - const repo = createSessionRepository({ store }); + const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); + ownedCollections.push(collection); + const repo = createSessionRepository({ collection }); const session = await repo.create({ id: "jsonl", cwd: root }); const expected = await verifyBranchQueries(session); const reopened = await repo.open(await session.getMetadata()); @@ -160,13 +160,13 @@ describe("bounded session branch queries", () => { const root = createTempDir(); const databasePath = join(root, "sessions.sqlite"); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ + const collection = createSqliteSessionCollection({ env: new NodeExecutionEnv({ cwd: root }), sqlite, databasePath, }); - ownedStores.push(store); - const repo = createSessionRepository({ store }); + ownedCollections.push(collection); + const repo = createSessionRepository({ collection }); 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 +226,13 @@ describe("bounded session branch queries", () => { const root = createTempDir(); const databasePath = join(root, "sessions.sqlite"); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ + const collection = createSqliteSessionCollection({ env: new NodeExecutionEnv({ cwd: root }), sqlite, databasePath, }); - ownedStores.push(store); - const repo = createSessionRepository({ store }); + ownedCollections.push(collection); + const repo = createSessionRepository({ collection }); 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 +269,13 @@ describe("bounded session branch queries", () => { const root = createTempDir(); const databasePath = join(root, "sessions.sqlite"); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ + const collection = createSqliteSessionCollection({ env: new NodeExecutionEnv({ cwd: root }), sqlite, databasePath, }); - ownedStores.push(store); - const repo = createSessionRepository({ store }); + ownedCollections.push(collection); + const repo = createSessionRepository({ collection }); 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 +324,13 @@ describe("bounded session branch queries", () => { it("provides identical SQLite query semantics", async () => { const root = createTempDir(); - const store = createSqliteSessionStore({ + const collection = createSqliteSessionCollection({ env: new NodeExecutionEnv({ cwd: root }), sqlite: createNodeSqliteFactory(), databasePath: join(root, "sessions.sqlite"), }); - ownedStores.push(store); - const repo = createSessionRepository({ store }); + ownedCollections.push(collection); + const repo = createSessionRepository({ collection }); 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 1adee7740..60709da50 100644 --- a/packages/agent/test/harness/repo.test.ts +++ b/packages/agent/test/harness/repo.test.ts @@ -1,13 +1,18 @@ import { existsSync, writeFileSync } from "node:fs"; import { afterEach, describe, expect, it, vi } from "vitest"; import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts"; -import { createJsonlSessionStore } from "../../src/harness/session/jsonl-store.ts"; +import { createJsonlSessionCollection } from "../../src/harness/session/jsonl-collection.ts"; import { - createInMemorySessionStore, + createInMemorySessionCollection, type InMemorySessionCreateOptions, -} from "../../src/harness/session/memory-store.ts"; +} from "../../src/harness/session/memory-collection.ts"; import { createSessionRepository } from "../../src/harness/session/repository.ts"; -import type { SessionForkSelection, SessionMetadata, SessionReader, SessionStore } from "../../src/harness/types.ts"; +import type { + SessionCollection, + SessionForkSelection, + SessionMetadata, + SessionStorage, +} from "../../src/harness/types.ts"; import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts"; afterEach(() => { @@ -134,53 +139,56 @@ class BlockingAppendEnv extends NodeExecutionEnv { } } -interface SessionStoreReadCounter { - loadCount: number; +interface SessionCollectionReadCounter { + openCount: number; readHeadCount: number; readEntriesCount: number; readPathCount: number; forkSelections: SessionForkSelection[]; } -function createCountingInMemorySessionStore(): { - store: SessionStore; - counter: SessionStoreReadCounter; +function createCountingInMemorySessionCollection(): { + collection: SessionCollection; + counter: SessionCollectionReadCounter; } { - const source = createInMemorySessionStore(); - const counter: SessionStoreReadCounter = { - loadCount: 0, + const source = createInMemorySessionCollection(); + const counter: SessionCollectionReadCounter = { + openCount: 0, readHeadCount: 0, readEntriesCount: 0, readPathCount: 0, forkSelections: [], }; - const countReads = (reader: SessionReader): SessionReader => ({ - metadata: reader.metadata, + const countReads = (storage: SessionStorage): SessionStorage => ({ + metadata: storage.metadata, readHead: () => { counter.readHeadCount += 1; - return reader.readHead(); + return storage.readHead(); }, - readEntry: (id) => reader.readEntry(id), + readEntry: (id) => storage.readEntry(id), readEntries: (options) => { counter.readEntriesCount += 1; - return reader.readEntries(options); + return storage.readEntries(options); }, - findEntriesOnBranch: (query) => reader.findEntriesOnBranch(query), + appendEntry: (entry) => storage.appendEntry(entry), + findEntriesOnBranch: (query) => storage.findEntriesOnBranch(query), readPathToRootOrCompaction: (leafId) => { counter.readPathCount += 1; - return reader.readPathToRootOrCompaction(leafId); + return storage.readPathToRootOrCompaction(leafId); }, + getLabel: (id) => storage.getLabel(id), + getName: () => storage.getName(), + getStats: () => storage.getStats(), }); return { counter, - store: { + collection: { create: (options) => source.create(options), - async load(metadata) { - counter.loadCount += 1; - return countReads(await source.load(metadata)); + async open(metadata) { + counter.openCount += 1; + return countReads(await source.open(metadata)); }, list: (options) => source.list(options), - appendEntry: (metadata, entry) => source.appendEntry(metadata, entry), delete: (metadata) => source.delete(metadata), fork: (metadata, options, selection) => { counter.forkSelections.push(selection); @@ -191,9 +199,9 @@ function createCountingInMemorySessionStore(): { }; } -describe("InMemorySessionStore", () => { +describe("InMemorySessionCollection", () => { it("opens, deletes, and forks by metadata", async () => { - const repo = createSessionRepository({ store: createInMemorySessionStore() }); + const repo = createSessionRepository({ collection: createInMemorySessionCollection() }); const session = await repo.create({ id: "session-1" }); const metadata = await session.getMetadata(); const user1 = await session.appendMessage(createUserMessage("one")); @@ -222,8 +230,8 @@ describe("InMemorySessionStore", () => { }); it("delegates full-session fork selection without opening the source", async () => { - const { store, counter } = createCountingInMemorySessionStore(); - const repo = createSessionRepository({ store }); + const { collection, counter } = createCountingInMemorySessionCollection(); + const repo = createSessionRepository({ collection }); const source = await repo.create({ id: "session-1" }); await source.appendMessage(createUserMessage("one")); @@ -231,7 +239,7 @@ describe("InMemorySessionStore", () => { expect((await fork.getEntries()).map((entry) => entry.id)).toHaveLength(1); expect(counter).toMatchObject({ - loadCount: 0, + openCount: 0, readHeadCount: 0, readEntriesCount: 0, readPathCount: 0, @@ -240,55 +248,55 @@ describe("InMemorySessionStore", () => { }); it("retains the opened aggregate instead of reloading for scoped reads", async () => { - const { store, counter } = createCountingInMemorySessionStore(); - const repo = createSessionRepository({ store }); + const { collection, counter } = createCountingInMemorySessionCollection(); + const repo = createSessionRepository({ collection }); const session = await repo.create({ id: "session-1" }); const entryId = await session.appendMessage(createUserMessage("one")); await session.getMetadata(); await session.getLeafId(); await session.getEntry(entryId); - expect(counter.loadCount).toBe(0); + expect(counter.openCount).toBe(0); }); - it("builds context from the branch reader without loading complete history", async () => { - const { store, counter } = createCountingInMemorySessionStore(); - const repo = createSessionRepository({ store }); + 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" }); await created.appendMessage(createUserMessage("one")); const opened = await repo.open(await created.getMetadata()); await opened.buildContext(); - expect(counter).toMatchObject({ loadCount: 1, readHeadCount: 1, readEntriesCount: 0, readPathCount: 1 }); + expect(counter).toMatchObject({ openCount: 1, readHeadCount: 1, readEntriesCount: 0, readPathCount: 1 }); }); - it("rejects repository operations and session writes after store disposal", async () => { - const store = createInMemorySessionStore(); - const repo = createSessionRepository({ store }); + it("rejects repository operations and session writes after collection disposal", async () => { + const collection = createInMemorySessionCollection(); + const repo = createSessionRepository({ collection }); const session = await repo.create({ id: "session-1" }); - await store[Symbol.asyncDispose](); + await collection[Symbol.asyncDispose](); - await expect(repo.list()).rejects.toThrow("In-memory session store is disposed"); + await expect(repo.list()).rejects.toThrow("In-memory session collection is disposed"); await expect(session.appendMessage(createUserMessage("late"))).rejects.toThrow( - "In-memory session store is disposed", + "In-memory session collection is disposed", ); }); it("supports lexical ownership with await using", async () => { - let listDisposedStore: (() => Promise) | undefined; + let listDisposedCollection: (() => Promise) | undefined; { - await using store = createInMemorySessionStore(); - const repository = createSessionRepository({ store }); - listDisposedStore = () => repository.list(); + await using collection = createInMemorySessionCollection(); + const repository = createSessionRepository({ collection }); + listDisposedCollection = () => repository.list(); await repository.create({ id: "session-1" }); } - await expect(listDisposedStore!()).rejects.toThrow("In-memory session store is disposed"); + await expect(listDisposedCollection!()).rejects.toThrow("In-memory session collection is disposed"); }); }); -describe("JsonlSessionStore", () => { +describe("JsonlSessionCollection", () => { it.each(["create", "fork"] as const)( "serializes conflicting create and %s destinations", async (secondOperation) => { @@ -302,14 +310,14 @@ describe("JsonlSessionStore", () => { `${JSON.stringify({ type: "session", version: 3, id: "source", timestamp: "2025-01-01T00:00:00.000Z", cwd: "/tmp/source" })}\n`, ); env.writtenPaths.add(sourcePath); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: root }); + const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root }); const options = { cwd: "/tmp/target", id: "shared-id" }; - const first = store.create(options); + const first = collection.create(options); await env.firstWriteStarted.promise; const second = secondOperation === "create" - ? store.create(options) - : store.fork( + ? collection.create(options) + : collection.fork( { id: "source", createdAt: "2025-01-01T00:00:00.000Z", @@ -336,8 +344,8 @@ describe("JsonlSessionStore", () => { it("encodes custom session IDs used in filenames", async () => { const root = createTempDir(); - const store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); - const snapshot = await store.create({ cwd: root, id: "../unsafe\\id" }); + const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); + const snapshot = await collection.create({ cwd: root, id: "../unsafe\\id" }); expect(snapshot.metadata.path).toContain("..%2Funsafe%5Cid.jsonl"); expect(existsSync(snapshot.metadata.path)).toBe(true); @@ -346,15 +354,15 @@ describe("JsonlSessionStore", () => { it("allows appends to different sessions to run concurrently", async () => { const root = createTempDir(); const env = new PerPathBlockingAppendEnv({ cwd: root }); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: root }); - const first = await store.create({ cwd: "/tmp/first", id: "first" }); - const second = await store.create({ cwd: "/tmp/second", id: "second" }); + const 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 firstBlock = env.blockAppend(first.metadata.path); const secondBlock = env.blockAppend(second.metadata.path); - const firstAppend = store.appendEntry(first.metadata, createEntry("first-entry")); + const firstAppend = first.appendEntry(createEntry("first-entry")); await firstBlock.started.promise; - const secondAppend = store.appendEntry(second.metadata, createEntry("second-entry")); + const secondAppend = second.appendEntry(createEntry("second-entry")); const secondStartedBeforeFirstFinished = await settlesBeforeNextTurn(secondBlock.started.promise); firstBlock.release.resolve(); @@ -366,15 +374,13 @@ describe("JsonlSessionStore", () => { it("caps concurrent operations across JSONL sessions at four by default", async () => { const root = createTempDir(); const env = new PerPathBlockingAppendEnv({ cwd: root }); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: root }); + const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root }); const snapshots = []; for (let index = 0; index < 6; index++) { - snapshots.push(await store.create({ cwd: `/tmp/session-${index}`, id: `session-${index}` })); + snapshots.push(await collection.create({ cwd: `/tmp/session-${index}`, id: `session-${index}` })); } const blocks = snapshots.map((snapshot) => env.blockAppend(snapshot.metadata.path)); - const appends = snapshots.map((snapshot, index) => - store.appendEntry(snapshot.metadata, createEntry(`entry-${index}`)), - ); + const appends = snapshots.map((snapshot, index) => snapshot.appendEntry(createEntry(`entry-${index}`))); await Promise.all(blocks.slice(0, 4).map((block) => block.started.promise)); const fifthStartedWithoutCapacity = await settlesBeforeNextTurn(blocks[4]!.started.promise); const sixthStartedWithoutCapacity = await settlesBeforeNextTurn(blocks[5]!.started.promise); @@ -389,14 +395,14 @@ describe("JsonlSessionStore", () => { it("allows overriding the JSONL concurrency limit", async () => { const root = createTempDir(); const env = new PerPathBlockingAppendEnv({ cwd: root }); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: root, maxConcurrentOperations: 1 }); - const first = await store.create({ cwd: "/tmp/first", id: "first" }); - const second = await store.create({ cwd: "/tmp/second", id: "second" }); + const 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 firstBlock = env.blockAppend(first.metadata.path); const secondBlock = env.blockAppend(second.metadata.path); - const firstAppend = store.appendEntry(first.metadata, createEntry("first-entry")); + const firstAppend = first.appendEntry(createEntry("first-entry")); await firstBlock.started.promise; - const secondAppend = store.appendEntry(second.metadata, createEntry("second-entry")); + const secondAppend = second.appendEntry(createEntry("second-entry")); const secondStartedBeforeFirstFinished = await settlesBeforeNextTurn(secondBlock.started.promise); firstBlock.release.resolve(); @@ -410,7 +416,7 @@ describe("JsonlSessionStore", () => { (maxConcurrentOperations) => { const root = createTempDir(); expect(() => - createJsonlSessionStore({ + createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root, maxConcurrentOperations, @@ -421,18 +427,18 @@ describe("JsonlSessionStore", () => { it("releases JSONL concurrency capacity after an operation fails", async () => { const root = createTempDir(); - const store = createJsonlSessionStore({ + const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root, maxConcurrentOperations: 1, }); - const first = await store.create({ cwd: "/tmp/first", id: "first" }); - const second = await store.create({ cwd: "/tmp/second", id: "second" }); + const first = await collection.create({ cwd: "/tmp/first", id: "first" }); + const second = await collection.create({ cwd: "/tmp/second", id: "second" }); const duplicate = createEntry("duplicate-entry"); - await store.appendEntry(first.metadata, duplicate); + await first.appendEntry(duplicate); - const failure = store.appendEntry(first.metadata, duplicate); - const nextSession = store.appendEntry(second.metadata, createEntry("next-entry")); + const failure = first.appendEntry(duplicate); + const nextSession = second.appendEntry(createEntry("next-entry")); await expect(failure).rejects.toThrow("Entry duplicate-entry already exists"); await expect(nextSession).resolves.toBeUndefined(); @@ -441,20 +447,20 @@ describe("JsonlSessionStore", () => { it("serializes appends to the same session", async () => { const root = createTempDir(); const env = new PerPathBlockingAppendEnv({ cwd: root }); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: root }); - const snapshot = await store.create({ cwd: root, id: "session" }); + const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root }); + const snapshot = await collection.create({ cwd: root, id: "session" }); const block = env.blockAppend(snapshot.metadata.path); - const firstAppend = store.appendEntry(snapshot.metadata, createEntry("first-entry")); + const firstAppend = snapshot.appendEntry(createEntry("first-entry")); await block.started.promise; - const secondAppend = store.appendEntry(snapshot.metadata, createEntry("second-entry")); + const secondAppend = snapshot.appendEntry(createEntry("second-entry")); await new Promise((resolve) => setImmediate(resolve)); const appendCountWhileFirstWasBlocked = env.appendCounts.get(snapshot.metadata.path); block.release.resolve(); await Promise.all([firstAppend, secondAppend]); expect(appendCountWhileFirstWasBlocked).toBe(1); - expect((await (await store.load(snapshot.metadata)).readEntries()).map((entry) => entry.id)).toEqual([ + expect((await (await collection.open(snapshot.metadata)).readEntries()).map((entry) => entry.id)).toEqual([ "first-entry", "second-entry", ]); @@ -463,17 +469,17 @@ describe("JsonlSessionStore", () => { it("uses listing as a barrier between accepted session operations", async () => { const root = createTempDir(); const env = new PerPathBlockingAppendEnv({ cwd: root }); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: root }); - const first = await store.create({ cwd: "/tmp/first", id: "first" }); - const second = await store.create({ cwd: "/tmp/second", id: "second" }); + const 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 firstBlock = env.blockAppend(first.metadata.path); const secondBlock = env.blockAppend(second.metadata.path); const listBlock = env.blockNextList(); - const firstAppend = store.appendEntry(first.metadata, createEntry("first-entry")); + const firstAppend = first.appendEntry(createEntry("first-entry")); await firstBlock.started.promise; - const list = store.list(); - const secondAppend = store.appendEntry(second.metadata, createEntry("second-entry")); + const list = collection.list(); + const secondAppend = second.appendEntry(createEntry("second-entry")); const listStartedBeforeFirstFinished = await settlesBeforeNextTurn(listBlock.started.promise); firstBlock.release.resolve(); @@ -493,17 +499,17 @@ describe("JsonlSessionStore", () => { it("waits for every accepted session operation during disposal", async () => { const root = createTempDir(); const env = new PerPathBlockingAppendEnv({ cwd: root }); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: root }); - const first = await store.create({ cwd: "/tmp/first", id: "first" }); - const second = await store.create({ cwd: "/tmp/second", id: "second" }); + const 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 firstBlock = env.blockAppend(first.metadata.path); const secondBlock = env.blockAppend(second.metadata.path); - const firstAppend = store.appendEntry(first.metadata, createEntry("first-entry")); - const secondAppend = store.appendEntry(second.metadata, createEntry("second-entry")); + const firstAppend = first.appendEntry(createEntry("first-entry")); + const secondAppend = second.appendEntry(createEntry("second-entry")); await Promise.all([firstBlock.started.promise, secondBlock.started.promise]); let disposed = false; - const dispose = store[Symbol.asyncDispose]().then(() => { + const dispose = collection[Symbol.asyncDispose]().then(() => { disposed = true; }); firstBlock.release.resolve(); @@ -520,13 +526,13 @@ describe("JsonlSessionStore", () => { it("waits for accepted appends before disposal and rejects later writes", async () => { const root = createTempDir(); const env = new BlockingAppendEnv({ cwd: root }); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: root }); - const repo = createSessionRepository({ store }); + const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: root }); + const repo = createSessionRepository({ collection }); 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 = store[Symbol.asyncDispose]().then(() => { + const dispose = collection[Symbol.asyncDispose]().then(() => { closed = true; }); await Promise.resolve(); @@ -534,13 +540,17 @@ describe("JsonlSessionStore", () => { env.releaseAppend.resolve(); await append; await dispose; - await expect(session.appendMessage(createUserMessage("late"))).rejects.toThrow("JSONL session store is disposed"); + await expect(session.appendMessage(createUserMessage("late"))).rejects.toThrow( + "JSONL session collection 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({ store: createJsonlSessionStore({ fs: env, sessionsRoot: root }) }); + const repo = createSessionRepository({ + collection: createJsonlSessionCollection({ fs: env, sessionsRoot: root }), + }); const created = await repo.create({ cwd: root, id: "session-1" }); const metadata = await created.getMetadata(); await created.appendMessage(createUserMessage("initial")); @@ -550,12 +560,14 @@ describe("JsonlSessionStore", () => { expect(env.readCount).toBe(1); }); - it("stores sessions below encoded cwd directories and lists by cwd", async () => { + it("collections 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({ store: createJsonlSessionStore({ fs: env, sessionsRoot: root }) }); + const repo = createSessionRepository({ + collection: createJsonlSessionCollection({ 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(); @@ -572,7 +584,9 @@ describe("JsonlSessionStore", () => { it("fails loudly when listing a malformed session file", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); - const repo = createSessionRepository({ store: createJsonlSessionStore({ fs: env, sessionsRoot: root }) }); + const repo = createSessionRepository({ + collection: createJsonlSessionCollection({ 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"); @@ -582,10 +596,10 @@ describe("JsonlSessionStore", () => { it("rejects a missing active leaf when opened", async () => { const root = createTempDir(); - const store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); - const repo = createSessionRepository({ store }); - const reader = await store.create({ cwd: root, id: "session-1" }); - await store.appendEntry(reader.metadata, { + const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); + const repo = createSessionRepository({ collection }); + const storage = await collection.create({ cwd: root, id: "session-1" }); + await storage.appendEntry({ type: "leaf", id: "leaf", parentId: null, @@ -593,7 +607,7 @@ describe("JsonlSessionStore", () => { timestamp: "2026-01-01T00:00:00.000Z", }); - await expect(repo.open(reader.metadata)).rejects.toMatchObject({ + await expect(repo.open(storage.metadata)).rejects.toMatchObject({ code: "invalid_session", message: "Entry missing not found", }); @@ -602,7 +616,9 @@ describe("JsonlSessionStore", () => { it("opens, deletes, and forks by metadata", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); - const repo = createSessionRepository({ store: createJsonlSessionStore({ fs: env, sessionsRoot: root }) }); + const repo = createSessionRepository({ + collection: createJsonlSessionCollection({ 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")); @@ -624,7 +640,9 @@ describe("JsonlSessionStore", () => { it("persists header metadata through create, list, and fork", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); - const repo = createSessionRepository({ store: createJsonlSessionStore({ fs: env, sessionsRoot: root }) }); + const repo = createSessionRepository({ + collection: createJsonlSessionCollection({ 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 9939f8338..50f521930 100644 --- a/packages/agent/test/harness/session-backends.test.ts +++ b/packages/agent/test/harness/session-backends.test.ts @@ -1,8 +1,8 @@ import { existsSync, readFileSync, writeFileSync } from "node:fs"; import { describe, expect, it } from "vitest"; import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts"; -import { createJsonlSessionStore, loadJsonlSessionMetadata } from "../../src/harness/session/jsonl-store.ts"; -import { createInMemorySessionStore } from "../../src/harness/session/memory-store.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 type { Session } from "../../src/harness/session/session.ts"; import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts"; @@ -41,11 +41,11 @@ async function appendUsageEntries(session: Session) { } function createInMemoryRepository() { - return createSessionRepository({ store: createInMemorySessionStore() }); + return createSessionRepository({ collection: createInMemorySessionCollection() }); } -function createJsonlRepository(options: Parameters[0]) { - return createSessionRepository({ store: createJsonlSessionStore(options) }); +function createJsonlRepository(options: Parameters[0]) { + return createSessionRepository({ collection: createJsonlSessionCollection(options) }); } describe("Session aggregate", () => { @@ -104,7 +104,7 @@ describe("Session aggregate", () => { }); }); -describe("JsonlSessionStore", () => { +describe("JsonlSessionCollection", () => { it("writes headers and entries and reopens the aggregate", async () => { const root = createTempDir(); const repo = createJsonlRepository({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); @@ -139,8 +139,8 @@ describe("JsonlSessionStore", () => { it("enforces entry uniqueness and does not recreate deleted files", async () => { const root = createTempDir(); - const store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); - const snapshot = await store.create({ cwd: root, id: "session-1" }); + const collection = createJsonlSessionCollection({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); + const snapshot = await collection.create({ cwd: root, id: "session-1" }); const entry = { type: "message" as const, id: "entry-1", @@ -148,20 +148,18 @@ describe("JsonlSessionStore", () => { timestamp: "2026-01-01T00:00:00.000Z", message: createUserMessage("one"), }; - await store.appendEntry(snapshot.metadata, entry); - await expect(store.appendEntry(snapshot.metadata, entry)).rejects.toThrow("Entry entry-1 already exists"); - await store.delete(snapshot.metadata); - await expect(store.appendEntry(snapshot.metadata, { ...entry, id: "entry-2" })).rejects.toThrow( - "Session not found", - ); + await snapshot.appendEntry(entry); + await expect(snapshot.appendEntry(entry)).rejects.toThrow("Entry entry-1 already exists"); + await collection.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 store = createJsonlSessionStore({ fs: new NodeExecutionEnv({ cwd: root }), sessionsRoot: root }); - const first = await store.create({ cwd: "/tmp/first", id: "shared-session-id" }); - const second = await store.create({ cwd: "/tmp/second", id: "shared-session-id" }); + const 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 entry = { type: "message" as const, id: "shared-entry-id", @@ -170,8 +168,8 @@ describe("JsonlSessionStore", () => { message: createUserMessage("one"), }; - await store.appendEntry(first.metadata, entry); - await expect(store.appendEntry(second.metadata, entry)).resolves.toBeUndefined(); + await first.appendEntry(entry); + await expect(second.appendEntry(entry)).resolves.toBeUndefined(); }); it("rejects non-object header metadata", async () => { diff --git a/packages/agent/test/harness/session-test-utils.ts b/packages/agent/test/harness/session-test-utils.ts index b2f6e6210..1312adc39 100644 --- a/packages/agent/test/harness/session-test-utils.ts +++ b/packages/agent/test/harness/session-test-utils.ts @@ -3,13 +3,13 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import type { AgentMessage } from "@earendil-works/pi-agent-core"; import { afterEach } from "vitest"; -import { createInMemorySessionStore } from "../../src/harness/session/memory-store.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"; export async function createInMemorySession(id?: string): Promise { - const store = createInMemorySessionStore(); - return createSessionRepository({ store }).create({ id }); + const collection = createInMemorySessionCollection(); + return createSessionRepository({ collection }).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 7b39de1a6..330bce824 100644 --- a/packages/agent/test/harness/session.test.ts +++ b/packages/agent/test/harness/session.test.ts @@ -2,8 +2,8 @@ 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 { createJsonlSessionStore } from "../../src/harness/session/jsonl-store.ts"; -import { createInMemorySessionStore } from "../../src/harness/session/memory-store.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 { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.ts"; @@ -227,7 +227,7 @@ async function runSessionSuite(name: string, createFixture: () => Promise { + it("persists leaf changes and appended entries through the collection", async () => { const fixture = await createFixture(); const session = await fixture.createSession(); const user1 = await session.appendMessage(createUserMessage("one")); @@ -246,29 +246,33 @@ async function runSessionSuite(name: string, createFixture: () => Promise { - const store = createInMemorySessionStore(); - const created = await createSessionRepository({ store }).create({}); +runSessionSuite("Session with in-memory collection", async () => { + const collection = createInMemorySessionCollection(); + const created = await createSessionRepository({ collection }).create({}); const metadata = await created.getMetadata(); return { - createSession: (contextBuildOptions) => createSessionRepository({ store, contextBuildOptions }).open(metadata), - reloadSession: (contextBuildOptions) => createSessionRepository({ store, contextBuildOptions }).open(metadata), + createSession: (contextBuildOptions) => + createSessionRepository({ collection, contextBuildOptions }).open(metadata), + reloadSession: (contextBuildOptions) => + createSessionRepository({ collection, contextBuildOptions }).open(metadata), }; }); let jsonlSessionPath = ""; runSessionSuite( - "Session with JSONL store", + "Session with JSONL collection", async () => { const dir = createTempDir(); const env = new NodeExecutionEnv({ cwd: dir }); - const store = createJsonlSessionStore({ fs: env, sessionsRoot: join(dir, "sessions") }); - const created = await createSessionRepository({ store }).create({ cwd: dir, id: "session-1" }); + const collection = createJsonlSessionCollection({ fs: env, sessionsRoot: join(dir, "sessions") }); + const created = await createSessionRepository({ collection }).create({ cwd: dir, id: "session-1" }); const metadata = await created.getMetadata(); jsonlSessionPath = metadata.path; return { - createSession: (contextBuildOptions) => createSessionRepository({ store, contextBuildOptions }).open(metadata), - reloadSession: (contextBuildOptions) => createSessionRepository({ store, contextBuildOptions }).open(metadata), + createSession: (contextBuildOptions) => + createSessionRepository({ collection, contextBuildOptions }).open(metadata), + reloadSession: (contextBuildOptions) => + createSessionRepository({ collection, contextBuildOptions }).open(metadata), }; }, () => { diff --git a/packages/agent/test/harness/sqlite-branch-cache.test.ts b/packages/agent/test/harness/sqlite-branch-cache.test.ts index 10b8989e2..b3a1d67f2 100644 --- a/packages/agent/test/harness/sqlite-branch-cache.test.ts +++ b/packages/agent/test/harness/sqlite-branch-cache.test.ts @@ -2,7 +2,7 @@ import { mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import { createNodeSqliteFactory, createSqliteSessionStore } from "../../../storage/sqlite-node/src/index.ts"; +import { createNodeSqliteFactory, createSqliteSessionCollection } 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"; @@ -12,12 +12,14 @@ function createTempDir(): string { } describe("SQLite branch cache", () => { - it("stores complete root paths for branches created after compaction", async () => { + it("collections complete root paths for branches created after compaction", async () => { const root = createTempDir(); const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const repo = createSessionRepository({ store: createSqliteSessionStore({ env, sqlite, databasePath }) }); + const repo = createSessionRepository({ + collection: createSqliteSessionCollection({ 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")); @@ -46,7 +48,9 @@ describe("SQLite branch cache", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const repo = createSessionRepository({ store: createSqliteSessionStore({ env, sqlite, databasePath }) }); + const repo = createSessionRepository({ + collection: createSqliteSessionCollection({ 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")); @@ -70,7 +74,7 @@ describe("SQLite branch cache", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const repo = createSessionRepository({ - store: createSqliteSessionStore({ env, sqlite: createNodeSqliteFactory(), databasePath }), + collection: createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath }), }); const session = await repo.create({ cwd: root, id: "session-1" }); const rootId = await session.appendMessage(createUserMessage("root")); @@ -100,7 +104,9 @@ describe("SQLite branch cache", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const repo = createSessionRepository({ store: createSqliteSessionStore({ env, sqlite, databasePath }) }); + const repo = createSessionRepository({ + collection: createSqliteSessionCollection({ 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")); @@ -136,7 +142,9 @@ describe("SQLite branch cache", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const repo = createSessionRepository({ store: createSqliteSessionStore({ env, sqlite, databasePath }) }); + const repo = createSessionRepository({ + collection: createSqliteSessionCollection({ env, sqlite, databasePath }), + }); const session = await repo.create({ cwd: root, id: "session-1" }); await session.appendMessage(createUserMessage("root")); @@ -177,7 +185,9 @@ describe("SQLite branch cache", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const repo = createSessionRepository({ store: createSqliteSessionStore({ env, sqlite, databasePath }) }); + const repo = createSessionRepository({ + collection: createSqliteSessionCollection({ 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")); @@ -204,7 +214,9 @@ describe("SQLite branch cache", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const repo = createSessionRepository({ store: createSqliteSessionStore({ env, sqlite, databasePath }) }); + const repo = createSessionRepository({ + collection: createSqliteSessionCollection({ 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")); @@ -245,7 +257,9 @@ describe("SQLite branch cache", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const repo = createSessionRepository({ store: createSqliteSessionStore({ env, sqlite, databasePath }) }); + const repo = createSessionRepository({ + collection: createSqliteSessionCollection({ 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 3bd4431bb..0ceb213e8 100644 --- a/packages/agent/test/harness/sqlite-migrations.test.ts +++ b/packages/agent/test/harness/sqlite-migrations.test.ts @@ -5,7 +5,7 @@ import { describe, expect, it } from "vitest"; import { applyMigrations, createNodeSqliteFactory, - createSqliteSessionStore, + createSqliteSessionCollection, loadMigrations, type SqliteDatabase, type SqliteDatabaseFactory, @@ -97,8 +97,8 @@ describe("SQLite migrations", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); await repo.create({ cwd: root, id: "session-1" }); const db = await sqlite.open(databasePath); @@ -179,7 +179,7 @@ describe("SQLite migrations", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const repo = createSessionRepository({ - store: createSqliteSessionStore({ env, sqlite: createNodeSqliteFactory(), databasePath }), + collection: createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath }), }); const source = await repo.create({ cwd: root, @@ -205,8 +205,8 @@ describe("SQLite migrations", () => { const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); const source = await repo.create({ cwd: root, id: "source" }); await source.appendMessage(createUserMessage("one")); await source.appendMessage(createAssistantMessage("two")); @@ -245,8 +245,8 @@ END; const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); 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 +287,8 @@ END; const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); 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")); @@ -322,7 +322,7 @@ END; const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const repo = createSessionRepository({ - store: createSqliteSessionStore({ env, sqlite: createNodeSqliteFactory(), databasePath }), + collection: createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath }), }); const session = await repo.create({ cwd: root, id: "session-1" }); const rootId = await session.appendMessage(createUserMessage("root")); @@ -344,7 +344,7 @@ END; const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const repo = createSessionRepository({ - store: createSqliteSessionStore({ env, sqlite: createNodeSqliteFactory(), databasePath }), + collection: createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath }), }); const session = await repo.create({ cwd: root, id: "session-1" }); const ids = [ @@ -371,12 +371,12 @@ END; open: async () => db, }; const env = new NodeExecutionEnv({ cwd: root }); - const store = createSqliteSessionStore({ env, sqlite, databasePath: join(root, "sessions.sqlite") }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath: join(root, "sessions.sqlite") }); + const repo = createSessionRepository({ collection }); await expect(repo.create({ cwd: root, id: "session-1" })).rejects.toThrow("insert failed"); expect(db.closeCount).toBe(0); - await store[Symbol.asyncDispose](); + await collection[Symbol.asyncDispose](); expect(db.closeCount).toBe(1); }); @@ -392,8 +392,8 @@ END; open: async () => db, }; const env = new NodeExecutionEnv({ cwd: root }); - const store = createSqliteSessionStore({ env, sqlite, databasePath: join(root, "sessions.sqlite") }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath: join(root, "sessions.sqlite") }); + const repo = createSessionRepository({ collection }); const metadata: SqliteSessionMetadata = { id: "missing", createdAt: new Date().toISOString(), @@ -404,7 +404,7 @@ END; await expect(repo.open(metadata)).rejects.toThrow("Session not found: missing"); expect(db.closeCount).toBe(0); - await store[Symbol.asyncDispose](); + await collection[Symbol.asyncDispose](); expect(db.closeCount).toBe(1); }); @@ -413,32 +413,32 @@ END; const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const { sqlite, counts } = createCloseCountingSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); 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 store[Symbol.asyncDispose](); + await collection[Symbol.asyncDispose](); expect(counts).toEqual({ opens: 1, closes: 1 }); - await store[Symbol.asyncDispose](); + await collection[Symbol.asyncDispose](); expect(counts).toEqual({ opens: 1, closes: 1 }); }); - it("shares one connection across source and fork until the store is disposed", async () => { + it("shares one connection across source and fork until the collection is disposed", async () => { const root = createTempDir(); const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const { sqlite, counts } = createCloseCountingSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); 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 store[Symbol.asyncDispose](); + await collection[Symbol.asyncDispose](); expect(counts).toEqual({ opens: 1, closes: 1 }); }); @@ -447,8 +447,8 @@ END; const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); const session = await repo.create({ cwd: root, id: "session-1" }); const metadata = await session.getMetadata(); @@ -470,8 +470,8 @@ END; const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); const session = await repo.create({ cwd: root, id: "session-1" }); const entryId = await session.appendMessage(createUserMessage("message")); const metadata = await session.getMetadata(); @@ -544,8 +544,8 @@ END; const databasePath = join(root, "sessions.sqlite"); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); - const store = createSqliteSessionStore({ env, sqlite, databasePath }); - const repo = createSessionRepository({ store }); + const collection = createSqliteSessionCollection({ env, sqlite, databasePath }); + const repo = createSessionRepository({ collection }); 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 21fecb584..bdff9636d 100644 --- a/packages/agent/test/harness/sqlite-node.test.ts +++ b/packages/agent/test/harness/sqlite-node.test.ts @@ -2,39 +2,39 @@ import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; import { createNodeSqliteFactory, + createSqliteSessionCollection, createSqliteSessionSearch, - createSqliteSessionStore, type SqliteSessionMetadata, } from "../../../storage/sqlite-node/src/index.ts"; import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts"; -import { createJsonlSessionStore } from "../../src/harness/session/jsonl-store.ts"; +import { createJsonlSessionCollection } from "../../src/harness/session/jsonl-collection.ts"; import { createSessionRepository } from "../../src/harness/session/repository.ts"; import { createScanningSessionSearch } from "../../src/harness/session/search-backend.ts"; import type { SessionSearch, SessionSearchHit, SessionSearchOptions } from "../../src/harness/types.ts"; import { createTempDir, createUserMessage } from "./session-test-utils.ts"; -const ownedStores: AsyncDisposable[] = []; +const ownedCollections: AsyncDisposable[] = []; afterEach(async () => { - for (const store of ownedStores.splice(0)) await store[Symbol.asyncDispose](); + for (const collection of ownedCollections.splice(0)) await collection[Symbol.asyncDispose](); }); -function createSqliteRepository(options: Parameters[0]) { - const store = createSqliteSessionStore(options); - ownedStores.push(store); +function createSqliteRepository(options: Parameters[0]) { + const collection = createSqliteSessionCollection(options); + ownedCollections.push(collection); return createSessionRepository({ - store, + collection, search: createSqliteSessionSearch(options), }); } -function createJsonlRepository(options: Parameters[0]) { - const store = createJsonlSessionStore(options); - ownedStores.push(store); - return createSessionRepository({ store, search: createScanningSessionSearch(store) }); +function createJsonlRepository(options: Parameters[0]) { + const collection = createJsonlSessionCollection(options); + ownedCollections.push(collection); + return createSessionRepository({ collection, search: createScanningSessionSearch(collection) }); } -describe("JsonlSessionStore with scanning search", () => { +describe("JsonlSessionCollection with scanning search", () => { it("searches canonical session entries by scanning", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); @@ -50,8 +50,8 @@ describe("JsonlSessionStore with scanning search", () => { }); }); -describe("SqliteSessionStore with explicit SQLite FTS5 search", () => { - it("uses SQLite FTS5 when composed with its search store", async () => { +describe("SqliteSessionCollection with explicit SQLite FTS5 search", () => { + it("uses SQLite FTS5 when composed with its search collection", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); const sqlite = createNodeSqliteFactory(); @@ -168,7 +168,7 @@ describe("SqliteSessionStore with explicit SQLite FTS5 search", () => { }); }); -describe("SqliteSessionStore with custom search", () => { +describe("SqliteSessionCollection with custom search", () => { it("can swap out the default search implementation", async () => { const root = createTempDir(); const searches: SessionSearchOptions[] = []; @@ -179,12 +179,12 @@ describe("SqliteSessionStore with custom search", () => { }, }; const env = new NodeExecutionEnv({ cwd: root }); - const store = createSqliteSessionStore({ + const collection = createSqliteSessionCollection({ env, sqlite: createNodeSqliteFactory(), databasePath: join(root, "sessions.sqlite"), }); - const repo = createSessionRepository({ store, search }); + const repo = createSessionRepository({ collection, search }); const session = await repo.create({ cwd: root, id: "session-1" }); await session.appendMessage(createUserMessage("stored canonically")); diff --git a/packages/agent/test/scratch/simple.ts b/packages/agent/test/scratch/simple.ts index 3727c6f28..6433c7261 100644 --- a/packages/agent/test/scratch/simple.ts +++ b/packages/agent/test/scratch/simple.ts @@ -8,7 +8,7 @@ import { AgentHarness, createBashTool, createEditTool, - createInMemorySessionStore, + createInMemorySessionCollection, createReadTool, createSessionRepository, createWriteTool, @@ -51,8 +51,8 @@ if (!model) { process.exit(-1); } -await using store = createInMemorySessionStore(); -const session = await createSessionRepository({ store }).create({}); +await using collection = createInMemorySessionCollection(); +const session = await createSessionRepository({ collection }).create({}); const agent = new AgentHarness({ session, models, diff --git a/packages/storage/sqlite-node/README.md b/packages/storage/sqlite-node/README.md index 3fa61db84..c9a9d19f9 100644 --- a/packages/storage/sqlite-node/README.md +++ b/packages/storage/sqlite-node/README.md @@ -2,11 +2,11 @@ Node sqlite storage backend for `@earendil-works/pi-agent-core` sessions. Provides the `node:sqlite` adapter (`SqliteDatabase` implementation) and the SQLite session -store implementation (`createSqliteSessionStore`, migrations, materialized views). The store -lazily owns one shared database connection and implements `AsyncDisposable`. +session collection (`createSqliteSessionCollection`, migrations, materialized views). The +collection lazily owns one shared database connection and implements `AsyncDisposable`. ```ts -await using store = createSqliteSessionStore(options); +await using collection = createSqliteSessionCollection(options); const search = createSqliteSessionSearch(options); -const repository = createSessionRepository({ store, search }); +const repository = createSessionRepository({ collection, search }); ``` diff --git a/packages/storage/sqlite-node/src/sqlite/index.ts b/packages/storage/sqlite-node/src/sqlite/index.ts index 1a252ba46..b6e8f4c47 100644 --- a/packages/storage/sqlite-node/src/sqlite/index.ts +++ b/packages/storage/sqlite-node/src/sqlite/index.ts @@ -1,4 +1,4 @@ export * from "./migrations.ts"; export * from "./search-backend.ts"; -export * from "./session-store.ts"; +export * from "./session-collection.ts"; export * from "./types.ts"; diff --git a/packages/storage/sqlite-node/src/sqlite/search-backend.ts b/packages/storage/sqlite-node/src/sqlite/search-backend.ts index aa3dedad9..63722b5f8 100644 --- a/packages/storage/sqlite-node/src/sqlite/search-backend.ts +++ b/packages/storage/sqlite-node/src/sqlite/search-backend.ts @@ -2,7 +2,12 @@ import type { SessionSearch, SessionSearchHit, SessionSearchOptions } from "@ear import { getFileSystemResultOrThrow } from "@earendil-works/pi-agent-core"; import { applyMigrations } from "./migrations.ts"; import { rowToMetadata, type SessionRow } from "./storage/sessions.ts"; -import type { SqliteDatabase, SqliteDatabaseFactory, SqliteSessionMetadata, SqliteSessionStoreEnv } from "./types.ts"; +import type { + SqliteDatabase, + SqliteDatabaseFactory, + SqliteSessionCollectionEnv, + SqliteSessionMetadata, +} from "./types.ts"; function getParentPath(path: string): string { const normalized = path.replace(/[\\/]+$/, ""); @@ -19,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/session-store.ts b/packages/storage/sqlite-node/src/sqlite/session-collection.ts similarity index 75% rename from packages/storage/sqlite-node/src/sqlite/session-store.ts rename to packages/storage/sqlite-node/src/sqlite/session-collection.ts index 730e86934..60a010883 100644 --- a/packages/storage/sqlite-node/src/sqlite/session-store.ts +++ b/packages/storage/sqlite-node/src/sqlite/session-collection.ts @@ -1,7 +1,7 @@ import type { + SessionCollection, SessionForkSelection, - SessionReader, - SessionStore, + SessionStorage, SessionTreeEntry, } from "@earendil-works/pi-agent-core"; import { @@ -16,10 +16,10 @@ import { rowToMetadata, type SessionRow } from "./storage/sessions.ts"; import type { SqliteDatabase, SqliteDatabaseFactory, + SqliteSessionCollectionEnv, SqliteSessionCreateOptions, SqliteSessionListOptions, SqliteSessionMetadata, - SqliteSessionStoreEnv, } from "./types.ts"; function getParentPath(path: string): string { @@ -36,8 +36,8 @@ async function configureSqliteDatabase(db: SqliteDatabase): Promise { await db.exec("PRAGMA busy_timeout=5000"); } -export type SqliteSessionStoreOptions = { - env: SqliteSessionStoreEnv; +export type SqliteSessionCollectionOptions = { + env: SqliteSessionCollectionEnv; sqlite: SqliteDatabaseFactory; databasePath: string; }; @@ -59,10 +59,10 @@ class SerialOperationQueue { } } -class SqliteSessionStore - implements SessionStore +class SqliteSessionCollection + implements SessionCollection { - private readonly env: SqliteSessionStoreEnv; + private readonly env: SqliteSessionCollectionEnv; private readonly sqlite: SqliteDatabaseFactory; private readonly databasePathInput: string; private databasePath: string | undefined; @@ -73,13 +73,13 @@ class SqliteSessionStore private readonly operations = new SerialOperationQueue(); private readonly writers = new Map(); - constructor(options: SqliteSessionStoreOptions) { + constructor(options: SqliteSessionCollectionOptions) { this.env = options.env; this.sqlite = options.sqlite; this.databasePathInput = options.databasePath; } - create(options: SqliteSessionCreateOptions): Promise> { + create(options: SqliteSessionCreateOptions): Promise> { this.assertOpen(); return this.operations.enqueue(async () => { const db = await this.getDatabase(); @@ -92,18 +92,17 @@ class SqliteSessionStore metadata: options.metadata, }), ); - const metadata = connection.metadata; - this.writers.set(metadata.id, connection); - return this.reader(connection); + this.writers.set(connection.metadata.id, connection); + return this.storage(connection); }); } - load(metadata: SqliteSessionMetadata): Promise> { + open(metadata: SqliteSessionMetadata): Promise> { this.assertOpen(); return this.operations.enqueue(() => this.loadSession(metadata)); } - private async loadSession(metadata: SqliteSessionMetadata): Promise> { + private async loadSession(metadata: SqliteSessionMetadata): Promise> { if ( !getFileSystemResultOrThrow(await this.env.exists(metadata.path), `Failed to check database ${metadata.path}`) ) { @@ -112,7 +111,7 @@ class SqliteSessionStore const connection = this.writers.get(metadata.id) ?? (await SqliteSessionConnection.open(await this.getDatabase(), metadata)); this.writers.set(metadata.id, connection); - return this.reader(connection); + return this.storage(connection); } list(options: SqliteSessionListOptions = {}): Promise { @@ -138,7 +137,7 @@ class SqliteSessionStore return rows.map((row) => rowToMetadata(row, path)); } - appendEntry(metadata: SqliteSessionMetadata, entry: SessionTreeEntry): Promise { + private appendEntry(metadata: SqliteSessionMetadata, entry: SessionTreeEntry): Promise { this.assertOpen(); return this.operations.enqueue(async () => { const connection = @@ -170,7 +169,7 @@ class SqliteSessionStore source: SqliteSessionMetadata, options: SqliteSessionCreateOptions, selection: SessionForkSelection, - ): Promise> { + ): Promise> { this.assertOpen(); return this.operations.enqueue(async () => { const db = await this.getDatabase(); @@ -187,9 +186,8 @@ class SqliteSessionStore for (const entry of entries) await connection.appendEntry(entry, { transaction: false }); return connection; }); - const metadata = connection.metadata; - this.writers.set(metadata.id, connection); - return this.reader(connection); + this.writers.set(connection.metadata.id, connection); + return this.storage(connection); }); } @@ -211,35 +209,39 @@ class SqliteSessionStore } private assertOpen(): void { - if (this.disposed) throw new SessionError("storage", "SQLite session store is disposed"); + if (this.disposed) throw new SessionError("storage", "SQLite session collection is disposed"); } - private reader(connection: SqliteSessionConnection): SessionReader { + private storage(connection: SqliteSessionConnection): SessionStorage { + const metadata = connection.metadata; return { - metadata: connection.metadata, - readHead: () => { - this.assertOpen(); - return this.operations.enqueue(() => connection.readHead()); - }, - readEntry: (id) => { - this.assertOpen(); - return this.operations.enqueue(() => connection.readEntry(id)); - }, - readEntries: (options) => { - this.assertOpen(); - return this.operations.enqueue(() => connection.readEntries(options)); - }, - findEntriesOnBranch: (query) => { - this.assertOpen(); - return this.operations.enqueue(() => connection.findEntriesOnBranch(query)); - }, - readPathToRootOrCompaction: (leafId) => { - this.assertOpen(); - return this.operations.enqueue(() => connection.readPathToRootOrCompaction(leafId)); - }, + metadata, + readHead: () => this.read(metadata, (current) => current.readHead()), + readEntry: (id) => this.read(metadata, (current) => current.readEntry(id)), + readEntries: (options) => this.read(metadata, (current) => current.readEntries(options)), + appendEntry: (entry) => this.appendEntry(metadata, entry), + findEntriesOnBranch: (query) => this.read(metadata, (current) => current.findEntriesOnBranch(query)), + readPathToRootOrCompaction: (leafId) => + this.read(metadata, (current) => current.readPathToRootOrCompaction(leafId)), + getLabel: (id) => this.read(metadata, (current) => current.getLabel(id)), + getName: () => this.read(metadata, (current) => current.getName()), + getStats: () => this.read(metadata, (current) => current.getStats()), }; } + private read( + metadata: SqliteSessionMetadata, + read: (connection: SqliteSessionConnection) => Promise, + ): Promise { + this.assertOpen(); + return this.operations.enqueue(async () => { + const connection = + this.writers.get(metadata.id) ?? (await SqliteSessionConnection.open(await this.getDatabase(), metadata)); + this.writers.set(metadata.id, connection); + return read(connection); + }); + } + private async getDatabasePath(): Promise { this.databasePath ??= getFileSystemResultOrThrow( await this.env.absolutePath(this.databasePathInput), @@ -273,8 +275,8 @@ class SqliteSessionStore } } -export function createSqliteSessionStore( - options: SqliteSessionStoreOptions, -): SessionStore { - return new SqliteSessionStore(options); +export function createSqliteSessionCollection( + options: SqliteSessionCollectionOptions, +): SessionCollection { + return new SqliteSessionCollection(options); } diff --git a/packages/storage/sqlite-node/src/sqlite/storage/index.ts b/packages/storage/sqlite-node/src/sqlite/storage/index.ts index 72fbf162f..c60a73d78 100644 --- a/packages/storage/sqlite-node/src/sqlite/storage/index.ts +++ b/packages/storage/sqlite-node/src/sqlite/storage/index.ts @@ -1,7 +1,7 @@ import type { SessionBranchQuery, SessionEntryCursorOptions, - SessionReader, + SessionStats, SessionTreeEntry, } from "@earendil-works/pi-agent-core"; import { SessionError, toError } from "@earendil-works/pi-agent-core"; @@ -74,7 +74,7 @@ async function loadSqliteSession( }; } -export class SqliteSessionConnection implements SessionReader { +export class SqliteSessionConnection { private readonly db: SqliteDatabase; readonly metadata: SqliteSessionMetadata; private byId: Map; @@ -332,6 +332,19 @@ export class SqliteSessionConnection implements SessionReader { + return this.materializedState.labelsById.get(id); + } + + async getName(): Promise { + return this.materializedState.name; + } + + async getStats(): Promise { + const { messageCount, cachedTokens, uncachedTokens, totalTokens, costTotal } = this.materializedState; + return { messageCount, cachedTokens, uncachedTokens, totalTokens, costTotal }; + } + async readHead(): Promise<{ leafId: string | null }> { const row = await this.db .prepare( diff --git a/packages/storage/sqlite-node/src/sqlite/types.ts b/packages/storage/sqlite-node/src/sqlite/types.ts index ebffb2d02..72e3951b8 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 SqliteSessionStoreEnv = Pick; +export type SqliteSessionCollectionEnv = Pick; diff --git a/scripts/browser-smoke-entry.ts b/scripts/browser-smoke-entry.ts index fb3ce9524..371211481 100644 --- a/scripts/browser-smoke-entry.ts +++ b/scripts/browser-smoke-entry.ts @@ -6,7 +6,7 @@ import { bashExecutionToText, convertToLlm, createCustomMessage, - createInMemorySessionStore, + createInMemorySessionCollection, createSessionRepository, FileError, formatPromptTemplateInvocation, @@ -29,7 +29,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({ store: createInMemorySessionStore() }); +const repo = createSessionRepository({ collection: createInMemorySessionCollection() }); const result = getOrThrow(ok({ value: 1 })); const customMessage = createCustomMessage("note", "hello", true, undefined, "2026-01-01T00:00:00.000Z"); const llmMessages = convertToLlm([customMessage]);