diff --git a/packages/session-backends/sqlite-node/src/sqlite/session.ts b/packages/session-backends/sqlite-node/src/sqlite/session.ts new file mode 100644 index 000000000..d515a976c --- /dev/null +++ b/packages/session-backends/sqlite-node/src/sqlite/session.ts @@ -0,0 +1,183 @@ +import type { + AgentMessage, + BranchScan, + Entry, + EntryQuery, + JsonValue, + LaneConfiguration, + Register, + RegisterNamespace, + Session, + SessionMutator, + SessionStats, + SessionTree, +} from "@earendil-works/pi-agent-core"; +import type { SqliteSessionMetadata } from "./session/session-row.ts"; + +/** + * SQLite-specific open-session lifecycle wrapper. + * + * TODO: Wire SqliteSessionRepo.create/open to construct: + * SqliteStorage -> StorageBackedSession -> SqliteOpenSession. + * TODO: Keep the database connection open until close() drains this wrapper and + * the wrapped StorageBackedSession closes its storage. + * TODO: Keep the writer lease claimed for the lifetime of this open session, + * renew it while open, and release only the matching fenced lease on close. + * TODO: Implement SqliteStorage.commit(); until then this wrapper can expose the + * Session API shape but cannot persist Session writes. + */ +export class SqliteOpenSession implements Session { + readonly metadata: SqliteSessionMetadata; + readonly idGenerator: Session["idGenerator"]; + private readonly session: Session; + private readonly onClose: () => void; + private readonly admitted = new Set>(); + private readonly closedError = new Error("Session is closed"); + private state: "open" | "closing" | "closed" = "open"; + private closePromise: Promise | undefined; + + constructor(session: Session, onClose: () => void) { + this.session = session; + this.metadata = session.metadata; + this.idGenerator = session.idGenerator; + this.onClose = onClose; + } + + mutate(lane: string, mutation: (mutator: SessionMutator) => T | Promise): Promise { + return this.admit(() => this.session.mutate(lane, mutation)); + } + + getEntries(ids: string[]): Promise> { + return this.admit(() => this.session.getEntries(ids)); + } + + getRegister( + namespace: TNamespace, + key: string, + ): Promise | undefined> { + return this.admit(() => this.session.getRegister(namespace, key)); + } + + listRegisters( + namespace: TNamespace, + keyPrefix?: string, + ): Promise[]> { + return this.admit(() => this.session.listRegisters(namespace, keyPrefix)); + } + + view(lane: string): SessionTree { + const view = this.session.view(lane); + return { + getLeafId: () => this.admit(() => view.getLeafId()), + getEntry: (id) => this.admit(() => view.getEntry(id)), + getStats: () => this.admit(() => view.getStats()), + getName: () => this.admit(() => view.getName()), + setName: (name) => this.admit(() => view.setName(name)), + getLabel: (targetId) => this.admit(() => view.getLabel(targetId)), + setLabel: (targetId, label) => this.admit(() => view.setLabel(targetId, label)), + getCustomFact: (key) => this.admit(() => view.getCustomFact(key)), + setCustomFact: (key, value) => this.admit(() => view.setCustomFact(key, value)), + findEntries: (query) => this.admit(() => view.findEntries(query)), + findEntry: (query) => this.admit(() => view.findEntry(query)), + findEntriesOnBranch: (query) => this.admit(() => view.findEntriesOnBranch(query)), + findEntryOnBranch: (query) => this.admit(() => view.findEntryOnBranch(query)), + appendMessage: (message) => this.admit(() => view.appendMessage(message)), + appendCustomEntry: (customType, data) => this.admit(() => view.appendCustomEntry(customType, data)), + }; + } + + createLane(name: string, at: string | null, configuration: LaneConfiguration): Promise { + return this.admit(async () => { + await this.session.createLane(name, at, configuration); + return this.view(name); + }); + } + + getLeafId(): Promise { + return this.admit(() => this.session.getLeafId()); + } + + getEntry(id: string): Promise { + return this.admit(() => this.session.getEntry(id)); + } + + getStats(): Promise { + return this.admit(() => this.session.getStats()); + } + + getName(): Promise { + return this.admit(() => this.session.getName()); + } + + setName(name: string | undefined): Promise { + return this.admit(() => this.session.setName(name)); + } + + getLabel(targetId: string): Promise { + return this.admit(() => this.session.getLabel(targetId)); + } + + setLabel(targetId: string, label: string | undefined): Promise { + return this.admit(() => this.session.setLabel(targetId, label)); + } + + getCustomFact(key: string): Promise { + return this.admit(() => this.session.getCustomFact(key)); + } + + setCustomFact(key: string, value: JsonValue | undefined): Promise { + return this.admit(() => this.session.setCustomFact(key, value)); + } + + findEntries(query?: EntryQuery): Promise { + return this.admit(() => this.session.findEntries(query)); + } + + findEntry(query?: EntryQuery): Promise { + return this.admit(() => this.session.findEntry(query)); + } + + findEntriesOnBranch(query?: BranchScan): Promise { + return this.admit(() => this.session.findEntriesOnBranch(query)); + } + + findEntryOnBranch(query?: BranchScan): Promise { + return this.admit(() => this.session.findEntryOnBranch(query)); + } + + appendMessage(message: AgentMessage): Promise { + return this.admit(() => this.session.appendMessage(message)); + } + + appendCustomEntry(customType: string, data?: JsonValue): Promise { + return this.admit(() => this.session.appendCustomEntry(customType, data)); + } + + close(): Promise { + if (this.closePromise !== undefined) return this.closePromise; + this.state = "closing"; + this.closePromise = Promise.allSettled([...this.admitted]) + .then(() => this.session.close()) + .finally(() => { + this.state = "closed"; + this.onClose(); + }); + return this.closePromise; + } + + private admit(operation: () => Promise): Promise { + if (this.state !== "open") return Promise.reject(this.closedError); + let result: Promise; + try { + result = operation(); + } catch (error) { + result = Promise.reject(error); + } + this.admitted.add(result); + void result.then( + () => this.admitted.delete(result), + () => this.admitted.delete(result), + ); + return result; + } +}