diff --git a/packages/agent/package.json b/packages/agent/package.json index 0216d723d..47f81afb7 100644 --- a/packages/agent/package.json +++ b/packages/agent/package.json @@ -14,6 +14,14 @@ "types": "./dist/node.d.ts", "import": "./dist/node.js" }, + "./experimental": { + "types": "./dist/experimental.d.ts", + "import": "./dist/experimental.js" + }, + "./experimental/session/testing": { + "types": "./dist/harness/experimental/session/testing/index.d.ts", + "import": "./dist/harness/experimental/session/testing/index.js" + }, "./package.json": "./package.json" }, "files": [ diff --git a/packages/agent/src/experimental.ts b/packages/agent/src/experimental.ts new file mode 100644 index 000000000..8639f4073 --- /dev/null +++ b/packages/agent/src/experimental.ts @@ -0,0 +1 @@ +export * from "./harness/experimental/session/index.ts"; diff --git a/packages/agent/src/harness/experimental/session/index.ts b/packages/agent/src/harness/experimental/session/index.ts new file mode 100644 index 000000000..846a90907 --- /dev/null +++ b/packages/agent/src/harness/experimental/session/index.ts @@ -0,0 +1,3 @@ +export * from "./memory.ts"; +export * from "./session.ts"; +export * from "./types.ts"; diff --git a/packages/agent/src/harness/experimental/session/memory.ts b/packages/agent/src/harness/experimental/session/memory.ts new file mode 100644 index 000000000..a79085145 --- /dev/null +++ b/packages/agent/src/harness/experimental/session/memory.ts @@ -0,0 +1,374 @@ +import { uuidv7 } from "@earendil-works/pi-ai"; +import { Session } from "./session.ts"; +import { + type BranchBounds, + type Entry, + type EntryOrder, + type EntryQuery, + type ForkOptions, + type LanePointer, + type LaneRecord, + type LogItem, + type LogOptions, + type NewRecord, + type ProvisionedEntry, + type RecordQuery, + type SessionCreateOptions, + SessionError, + type SessionMetadata, + type SessionRepo, + type SessionStats, + type SessionStorage, +} from "./types.ts"; + +function* ordered(items: readonly T[], order: EntryOrder | undefined): IterableIterator { + if (order === "oldestFirst") { + yield* items; + return; + } + for (let index = items.length - 1; index >= 0; index--) yield items[index]!; +} + +function provisionEntry( + newEntry: ProvisionedEntry, + parentId: string | null, + seq: number, +): TEntry { + // Object spread does not preserve the correlation between a discriminant and the rest of a union member. + return { ...newEntry, parentId, seq, timestamp: Date.now() } as unknown as TEntry; +} + +function provisionRecord(newRecord: NewRecord, seq: number): TRecord { + // Object spread does not preserve the correlation between a discriminant and the rest of a union member. + return { ...newRecord, seq, timestamp: Date.now() } as unknown as TRecord; +} + +export class InMemorySessionStorage implements SessionStorage { + private readonly metadata: SessionMetadata; + private sequence = 0; + private readonly usedIds = new Set(); + private readonly entries: Entry[] = []; + private readonly entriesById = new Map(); + private readonly laneByEntryId = new Map(); + private readonly records: LaneRecord[] = []; + private readonly lanes = new Map([["main", null]]); + private readonly log: LogItem[] = []; + private readonly stats: SessionStats = { + messageCount: 0, + cachedTokens: 0, + uncachedTokens: 0, + totalTokens: 0, + costTotal: 0, + }; + private name: string | undefined; + private readonly labels = new Map(); + + constructor(metadata: SessionMetadata) { + this.metadata = structuredClone(metadata); + } + + fork(metadata: SessionMetadata, options: ForkOptions & SessionCreateOptions): InMemorySessionStorage { + const storage = new InMemorySessionStorage(metadata); + let copiedEntries: Entry[]; + if (options.scope === "tree") { + copiedEntries = this.entries; + storage.lanes.clear(); + for (const [lane, leafId] of this.lanes) storage.lanes.set(lane, leafId); + } else { + const selectedEntryId = options.entryId ?? this.lanes.get("main") ?? null; + let targetId: string | null = null; + if (selectedEntryId !== null) { + const target = this.entriesById.get(selectedEntryId); + if (!target || target.type !== "message") { + throw new SessionError("invalid_fork_target", `Fork target is not a message entry: ${selectedEntryId}`); + } + const position = options.position ?? (options.entryId === undefined ? "at" : "before"); + targetId = position === "at" ? target.id : target.parentId; + } + copiedEntries = [...this.walkToRoot(targetId)].reverse(); + storage.lanes.set("main", targetId); + } + + const copiedIds = new Set(copiedEntries.map((entry) => entry.id)); + for (const sourceEntry of copiedEntries) { + const entry = structuredClone(sourceEntry); + entry.seq = storage.nextSequence(); + const lane = options.scope === "tree" ? this.requireEntryLane(sourceEntry.id) : "main"; + storage.entries.push(entry); + storage.entriesById.set(entry.id, entry); + storage.laneByEntryId.set(entry.id, lane); + storage.usedIds.add(entry.id); + storage.log.push({ kind: "entry", seq: entry.seq, lane, entry }); + } + for (const [lane, leafId] of options.scope === "tree" ? this.lanes : []) { + storage.log.push({ + kind: "lane", + seq: storage.nextSequence(), + lane, + action: lane === "main" ? "move" : "create", + leafId, + }); + } + if (this.name !== undefined) { + storage.name = this.name; + storage.log.push({ kind: "fact", seq: storage.nextSequence(), fact: "name", name: this.name }); + } + for (const [targetId, label] of this.labels) { + if (!copiedIds.has(targetId)) continue; + storage.labels.set(targetId, label); + storage.log.push({ kind: "fact", seq: storage.nextSequence(), fact: "label", targetId, label }); + } + return storage; + } + + async getMetadata(): Promise { + return structuredClone(this.metadata); + } + + async getLanes(): Promise { + return [...this.lanes].map(([lane, leafId]) => ({ lane, leafId })); + } + + async createLane(lane: string, at: string | null): Promise { + if (this.lanes.has(lane)) throw new SessionError("already_exists", `Lane already exists: ${lane}`); + this.validateTarget(at); + this.lanes.set(lane, at); + this.log.push({ kind: "lane", seq: this.nextSequence(), lane, action: "create", leafId: at }); + } + + async moveLane(lane: string, to: string | null): Promise { + this.requireLane(lane); + this.validateTarget(to); + this.lanes.set(lane, to); + this.log.push({ kind: "lane", seq: this.nextSequence(), lane, action: "move", leafId: to }); + } + + async appendEntry(newEntry: ProvisionedEntry, lane: string): Promise { + const parentId = this.requireLane(lane); + this.validateUnusedId(newEntry.id); + const clonedEntry = structuredClone(newEntry); + const entry = provisionEntry(clonedEntry, parentId, this.nextSequence()); + this.usedIds.add(entry.id); + this.entries.push(entry); + this.entriesById.set(entry.id, entry); + this.laneByEntryId.set(entry.id, lane); + this.lanes.set(lane, entry.id); + this.log.push({ kind: "entry", seq: entry.seq, lane, entry }); + if (entry.type === "message") this.stats.messageCount += 1; + return structuredClone(entry); + } + + async appendRecord(newRecord: NewRecord): Promise { + this.requireLane(newRecord.lane); + this.validateUnusedId(newRecord.id); + const clonedRecord = structuredClone(newRecord); + const record = provisionRecord(clonedRecord, this.nextSequence()); + this.usedIds.add(record.id); + this.records.push(record); + this.log.push({ kind: "record", seq: record.seq, record }); + if (record.type === "usage") { + this.stats.cachedTokens += record.usage.cacheRead; + this.stats.uncachedTokens += record.usage.input + record.usage.cacheWrite; + this.stats.totalTokens += record.usage.totalTokens; + this.stats.costTotal += record.usage.cost.total; + } + return structuredClone(record); + } + + async getEntry(id: string): Promise { + const entry = this.entriesById.get(id); + return entry === undefined ? undefined : structuredClone(entry); + } + + async findEntries(query: EntryQuery = {}): Promise { + const results: Entry[] = []; + for (const entry of ordered(this.entries, query.order)) { + if (!this.matchesEntryQuery(entry, query)) continue; + results.push(entry); + if (results.length === query.limit) break; + } + return structuredClone(results); + } + + async findEntriesOnBranch(query: EntryQuery & BranchBounds & { start: string }): Promise { + const results: Entry[] = []; + if (query.order === "oldestFirst") { + for (const entry of [...this.walkToRoot(query.start)].reverse()) { + const reachedBound = entry.id === query.stopAtId || entry.type === query.stopAtType; + if (this.matchesEntryQuery(entry, query)) results.push(entry); + if (reachedBound || results.length === query.limit) break; + } + } else { + for (const entry of this.walkToRoot(query.start, query)) { + if (this.matchesEntryQuery(entry, query)) results.push(entry); + if (results.length === query.limit) break; + } + } + return structuredClone(results); + } + + async findRecords( + query: RecordQuery & { type: K }, + ): Promise[]>; + async findRecords(query?: RecordQuery): Promise; + async findRecords(query: RecordQuery = {}): Promise { + const results: LaneRecord[] = []; + for (const record of ordered(this.records, query.order)) { + if (!this.matchesRecordQuery(record, query)) continue; + results.push(record); + if (results.length === query.limit) break; + } + return structuredClone(results); + } + + async getLog(options: LogOptions = {}): Promise { + const results: LogItem[] = []; + for (const item of this.log) { + if (options.afterSeq !== undefined && item.seq <= options.afterSeq) continue; + results.push(item); + if (results.length === options.limit) break; + } + return structuredClone(results); + } + + async getName(): Promise { + return this.name; + } + + async setName(name: string): Promise { + this.name = name; + this.log.push({ kind: "fact", seq: this.nextSequence(), fact: "name", name }); + } + + async getLabel(id: string): Promise { + return this.labels.get(id); + } + + async setLabel(id: string, label: string | undefined): Promise { + this.validateTarget(id); + if (label === undefined) this.labels.delete(id); + else this.labels.set(id, label); + this.log.push({ kind: "fact", seq: this.nextSequence(), fact: "label", targetId: id, label }); + } + + async getStats(): Promise { + return structuredClone(this.stats); + } + + private nextSequence(): number { + return ++this.sequence; + } + + private requireLane(lane: string): string | null { + const leafId = this.lanes.get(lane); + if (leafId === undefined) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); + return leafId; + } + + private requireEntryLane(entryId: string): string { + const lane = this.laneByEntryId.get(entryId); + if (lane === undefined) { + throw new SessionError("invalid_entry", `Entry has no originating lane: ${entryId}`); + } + return lane; + } + + private validateTarget(targetId: string | null): void { + if (targetId !== null && !this.entriesById.has(targetId)) { + throw new SessionError("not_found", `Entry not found: ${targetId}`); + } + } + + private validateUnusedId(id: string): void { + if (this.usedIds.has(id)) throw new SessionError("already_exists", `Session id already exists: ${id}`); + } + + private *walkToRoot( + start: string | null, + bounds?: Pick, + ): IterableIterator { + if (start === null) return; + const visited = new Set(); + let current = this.entriesById.get(start); + if (!current) throw new SessionError("not_found", `Entry not found: ${start}`); + while (current) { + if (visited.has(current.id)) { + throw new SessionError("invalid_entry", `Session branch contains a cycle at ${current.id}`); + } + visited.add(current.id); + yield current; + if (current.id === bounds?.stopAtId || current.type === bounds?.stopAtType) break; + if (current.parentId === null) break; + const parentId: string = current.parentId; + current = this.entriesById.get(parentId); + if (!current) throw new SessionError("invalid_entry", `Entry not found: ${parentId}`); + } + } + + private matchesEntryQuery(entry: Entry, query: EntryQuery): boolean { + return ( + (query.type === undefined || entry.type === query.type) && + (query.customType === undefined || (entry.type === "custom" && entry.customType === query.customType)) && + (query.cursor === undefined || + (query.order === "oldestFirst" ? entry.seq > query.cursor.afterSeq : entry.seq < query.cursor.afterSeq)) + ); + } + + private matchesRecordQuery(record: LaneRecord, query: RecordQuery): boolean { + return ( + (query.lane === undefined || record.lane === query.lane) && + (query.type === undefined || record.type === query.type) && + (query.runId === undefined || + (record.type === "operation_started" + ? record.id === query.runId + : "runId" in record && record.runId === query.runId)) && + (query.afterSeq === undefined || record.seq > query.afterSeq) + ); + } +} + +export class InMemorySessionRepo implements SessionRepo { + private readonly sessions = new Map(); + + async create(options: SessionCreateOptions = {}): Promise { + const id = options.id ?? uuidv7(); + if (this.sessions.has(id)) throw new SessionError("already_exists", `Session already exists: ${id}`); + const storage = new InMemorySessionStorage({ + id, + createdAt: Date.now(), + parentSessionId: options.parentSessionId, + }); + this.sessions.set(id, storage); + return new Session(storage); + } + + async open(metadata: SessionMetadata): Promise { + return new Session(this.requireStorage(metadata.id)); + } + + async list(): Promise { + return Promise.all([...this.sessions.values()].map((storage) => storage.getMetadata())); + } + + async delete(metadata: SessionMetadata): Promise { + this.sessions.delete(metadata.id); + } + + async fork(source: SessionMetadata, options: ForkOptions & SessionCreateOptions = {}): Promise { + const sourceStorage = this.requireStorage(source.id); + const id = options.id ?? uuidv7(); + if (this.sessions.has(id)) throw new SessionError("already_exists", `Session already exists: ${id}`); + const storage = sourceStorage.fork( + { id, createdAt: Date.now(), parentSessionId: options.parentSessionId ?? source.id }, + options, + ); + this.sessions.set(id, storage); + return new Session(storage); + } + + private requireStorage(id: string): InMemorySessionStorage { + const storage = this.sessions.get(id); + if (!storage) throw new SessionError("not_found", `Session not found: ${id}`); + return storage; + } +} diff --git a/packages/agent/src/harness/experimental/session/session.ts b/packages/agent/src/harness/experimental/session/session.ts new file mode 100644 index 000000000..9d8269feb --- /dev/null +++ b/packages/agent/src/harness/experimental/session/session.ts @@ -0,0 +1,281 @@ +import { uuidv7 } from "@earendil-works/pi-ai"; +import type { AgentMessage } from "../../../types.ts"; +import type { + BranchBounds, + Entry, + EntryQuery, + IdGenerator, + LanePointer, + LaneRecord, + LogItem, + LogOptions, + NewRecord, + ProvisionedEntry, + RecordQuery, + SessionMetadata, + SessionStats, + SessionStorage, + SessionTree, +} from "./types.ts"; +import { SessionError } from "./types.ts"; + +type JsonValidationFrame = { value: unknown } | { exit: object }; + +function invalidPayload(reason: string): never { + throw new SessionError("invalid_payload", `Durable payload ${reason}`); +} + +function assertValidLimit(limit: number | undefined): void { + if (limit !== undefined && (!Number.isInteger(limit) || limit <= 0)) { + throw new SessionError("invalid_query", "limit must be a positive integer"); + } +} + +function assertValidCursor(afterSeq: number | undefined): void { + if (afterSeq !== undefined && (!Number.isInteger(afterSeq) || afterSeq < 0)) { + throw new SessionError("invalid_query", "cursor sequence must be a non-negative integer"); + } +} + +function assertJsonSerializable(value: unknown): void { + const active = new WeakSet(); + const stack: JsonValidationFrame[] = [{ value }]; + while (stack.length > 0) { + const frame = stack.pop()!; + if ("exit" in frame) { + active.delete(frame.exit); + continue; + } + const candidate = frame.value; + if (candidate === null || typeof candidate === "string" || typeof candidate === "boolean") { + continue; + } + if (typeof candidate === "number") { + if (!Number.isFinite(candidate)) invalidPayload("contains a non-finite number"); + continue; + } + if (typeof candidate !== "object") invalidPayload(`contains ${typeof candidate}`); + if (active.has(candidate)) invalidPayload("contains a cycle"); + active.add(candidate); + stack.push({ exit: candidate }); + + if (Array.isArray(candidate)) { + if (Object.getPrototypeOf(candidate) !== Array.prototype) { + invalidPayload("contains a non-standard array"); + } + if ( + Object.getOwnPropertySymbols(candidate).length > 0 || + Object.getOwnPropertyNames(candidate).length !== candidate.length + 1 + ) { + invalidPayload("contains an array with unsupported properties"); + } + for (let index = candidate.length - 1; index >= 0; index--) { + if (!Object.hasOwn(candidate, index)) invalidPayload("contains a sparse array"); + const descriptor = Object.getOwnPropertyDescriptor(candidate, index)!; + if (!("value" in descriptor)) invalidPayload("contains an array accessor"); + stack.push({ value: descriptor.value }); + } + continue; + } + + const prototype = Object.getPrototypeOf(candidate); + if (prototype !== Object.prototype && prototype !== null) { + invalidPayload("contains a non-plain object"); + } + if (Object.getOwnPropertySymbols(candidate).length > 0) { + invalidPayload("contains a symbol-keyed property"); + } + const keys = Object.keys(candidate); + if (Object.getOwnPropertyNames(candidate).length !== keys.length) { + invalidPayload("contains a non-enumerable property"); + } + for (let index = keys.length - 1; index >= 0; index--) { + const descriptor = Object.getOwnPropertyDescriptor(candidate, keys[index]!)!; + if (!("value" in descriptor)) invalidPayload("contains an accessor"); + stack.push({ value: descriptor.value }); + } + } +} + +export class Session implements SessionTree { + private readonly storage: SessionStorage; + readonly idGenerator: IdGenerator; + + constructor(storage: SessionStorage, options: { idGenerator?: IdGenerator } = {}) { + this.storage = storage; + this.idGenerator = options.idGenerator ?? { next: () => uuidv7() }; + } + + async getMetadata(): Promise { + return this.storage.getMetadata(); + } + + view(lane: string): SessionTree { + if (lane === "main") return this; + return { + getLeafId: () => this.getLeafIdForLane(lane), + getEntry: (id) => this.getEntry(id), + getStats: () => this.getStats(), + getName: () => this.getName(), + setName: (name) => this.setName(name), + getLabel: (targetId) => this.getLabel(targetId), + setLabel: (targetId, label) => this.setLabel(targetId, label), + findEntries: (query) => this.queryEntries(query), + findEntry: async (query = {}) => (await this.queryEntries(query, 1))[0], + findEntriesOnBranch: (query) => this.queryBranchEntries(lane, query), + findEntryOnBranch: async (query = {}) => (await this.queryBranchEntries(lane, query, 1))[0], + appendMessage: (message) => this.appendMessageToLane(lane, message), + appendCustomEntry: (customType, data) => this.appendCustomEntryToLane(lane, customType, data), + }; + } + + async getLeafId(): Promise { + return this.getLeafIdForLane("main"); + } + + async getEntry(id: string): Promise { + return this.storage.getEntry(id); + } + + async getStats(): Promise { + return this.storage.getStats(); + } + + async getName(): Promise { + return this.storage.getName(); + } + + async setName(name: string): Promise { + await this.storage.setName(name); + } + + async getLabel(targetId: string): Promise { + return this.storage.getLabel(targetId); + } + + async setLabel(targetId: string, label: string | undefined): Promise { + await this.storage.setLabel(targetId, label); + } + + async findEntries(query?: EntryQuery): Promise { + return this.queryEntries(query); + } + + async findEntry(query: EntryQuery = {}): Promise { + return (await this.queryEntries(query, 1))[0]; + } + + async findEntriesOnBranch(query?: EntryQuery & BranchBounds): Promise { + return this.queryBranchEntries("main", query); + } + + async findEntryOnBranch(query: EntryQuery & BranchBounds = {}): Promise { + return (await this.queryBranchEntries("main", query, 1))[0]; + } + + async appendMessage(message: AgentMessage): Promise { + return this.appendMessageToLane("main", message); + } + + async appendCustomEntry(customType: string, data?: unknown): Promise { + return this.appendCustomEntryToLane("main", customType, data); + } + + async getLanes(): Promise { + return this.storage.getLanes(); + } + + async createLane(lane: string, at: string | null): Promise { + await this.storage.createLane(lane, at); + } + + async moveLane(lane: string, to: string | null): Promise { + await this.storage.moveLane(lane, to); + } + + async appendEntry(entry: ProvisionedEntry, lane: string): Promise { + return this.commitEntry(entry, lane); + } + + async appendRecord( + record: TNewRecord, + ): Promise>; + async appendRecord(record: NewRecord): Promise; + async appendRecord(record: NewRecord): Promise { + return this.commitRecord(record); + } + + async findRecords( + query: RecordQuery & { type: K }, + ): Promise[]>; + async findRecords(query?: RecordQuery): Promise; + async findRecords(query?: RecordQuery): Promise { + return this.queryRecords(query); + } + + async getLog(options?: LogOptions): Promise { + return this.queryLog(options); + } + + private async getLeafIdForLane(lane: string): Promise { + const pointer = (await this.storage.getLanes()).find((candidate) => candidate.lane === lane); + if (!pointer) throw new SessionError("invalid_lane", `Lane not found: ${lane}`); + return pointer.leafId; + } + + private async queryEntries(query: EntryQuery = {}, resultLimit = query.limit): Promise { + assertValidLimit(query.limit); + assertValidCursor(query.cursor?.afterSeq); + return this.storage.findEntries(resultLimit === query.limit ? query : { ...query, limit: resultLimit }); + } + + private async queryBranchEntries( + lane: string, + query: EntryQuery & BranchBounds = {}, + resultLimit = query.limit, + ): Promise { + assertValidLimit(query.limit); + assertValidCursor(query.cursor?.afterSeq); + const start = query.start ?? (await this.getLeafIdForLane(lane)); + if (start === null) return []; + const storageQuery = resultLimit === query.limit ? query : { ...query, limit: resultLimit }; + return this.storage.findEntriesOnBranch({ ...storageQuery, start }); + } + + private async queryRecords(query: RecordQuery = {}): Promise { + assertValidLimit(query.limit); + assertValidCursor(query.afterSeq); + return this.storage.findRecords(query); + } + + private async queryLog(options: LogOptions = {}): Promise { + assertValidLimit(options.limit); + assertValidCursor(options.afterSeq); + return this.storage.getLog(options); + } + + private async appendMessageToLane(lane: string, message: AgentMessage): Promise { + const entry = await this.commitEntry({ type: "message", id: this.idGenerator.next(), message }, lane); + return entry.id; + } + + private async appendCustomEntryToLane(lane: string, customType: string, data?: unknown): Promise { + const entry = await this.commitEntry( + data === undefined + ? { type: "custom", id: this.idGenerator.next(), customType } + : { type: "custom", id: this.idGenerator.next(), customType, data }, + lane, + ); + return entry.id; + } + + private async commitEntry(entry: ProvisionedEntry, lane: string): Promise { + assertJsonSerializable(entry); + return this.storage.appendEntry(entry, lane); + } + + private async commitRecord(record: NewRecord): Promise { + assertJsonSerializable(record); + return this.storage.appendRecord(record); + } +} diff --git a/packages/agent/src/harness/experimental/session/testing/conformance.ts b/packages/agent/src/harness/experimental/session/testing/conformance.ts new file mode 100644 index 000000000..9f1f720ea --- /dev/null +++ b/packages/agent/src/harness/experimental/session/testing/conformance.ts @@ -0,0 +1,830 @@ +import { deepStrictEqual, ok, rejects, strictEqual } from "node:assert/strict"; +import type { AgentMessage } from "../../../../types.ts"; +import type { + CustomEntry, + Entry, + MessageEntry, + NewRecord, + OperationStartedRecord, + SessionErrorCode, + SessionRepo, +} from "../types.ts"; +import type { SessionBackendConformanceCase, SessionBackendFixtureFactory } from "./types.ts"; + +function createUserMessage(text: string): AgentMessage { + return { + role: "user", + content: [{ type: "text", text }], + timestamp: 1, + }; +} + +function createAssistantMessage(text: string): AgentMessage { + return { + role: "assistant", + content: [{ type: "text", text }], + api: "anthropic-messages", + provider: "anthropic", + model: "claude-sonnet-4-5", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: 1, + }; +} + +function operationStarted(id: string, lane = "main"): NewRecord { + return { + type: "operation_started", + id, + lane, + sourceLeafId: null, + intent: { kind: "run", originalPrompt: [], initialMessages: [] }, + }; +} + +async function entryIds(entries: Promise): Promise { + return (await entries).map((entry) => entry.id); +} + +async function rejectsWithCode(operation: Promise, code: SessionErrorCode): Promise { + await rejects( + operation, + (error: unknown) => typeof error === "object" && error !== null && "code" in error && error.code === code, + `Expected SessionError with code ${code}`, + ); +} + +type ConformanceTest = (repository: SessionRepo) => Promise; + +function createCase( + factory: SessionBackendFixtureFactory, + group: string, + name: string, + test: ConformanceTest, +): SessionBackendConformanceCase { + return { + group, + name, + async run() { + await using fixture = await factory(); + await test(fixture.repository); + }, + }; +} + +/** Creates the session backend conformance cases. Each case creates and disposes its own fixture. */ +export function createSessionBackendConformance( + factory: SessionBackendFixtureFactory, +): readonly SessionBackendConformanceCase[] { + return [ + createCase( + factory, + "entries and lanes", + "assigns parents and one sequence across every mutation", + async (repository) => { + const session = await repository.create({ id: "session" }); + const root = await session.appendEntry( + { type: "message", id: "root", message: createUserMessage("root") }, + "main", + ); + await session.createLane("thread", root.id); + const child = await session.appendEntry( + { type: "custom", id: "child", customType: "note", data: { value: 1 } }, + "thread", + ); + const record = await session.appendRecord(operationStarted("run", "thread")); + await session.setName("Example"); + await session.setLabel(root.id, "checkpoint"); + await session.moveLane("main", child.id); + + deepStrictEqual({ parentId: root.parentId, seq: root.seq }, { parentId: null, seq: 1 }); + deepStrictEqual({ parentId: child.parentId, seq: child.seq }, { parentId: "root", seq: 3 }); + strictEqual(record.seq, 4); + for (const timestamp of [root.timestamp, child.timestamp, record.timestamp]) { + ok( + Number.isSafeInteger(timestamp) && timestamp >= 0, + "storage-assigned timestamps must be Unix milliseconds", + ); + } + deepStrictEqual( + (await session.getLog()).map((item) => [item.kind, item.seq]), + [ + ["entry", 1], + ["lane", 2], + ["entry", 3], + ["record", 4], + ["fact", 5], + ["fact", 6], + ["lane", 7], + ], + ); + deepStrictEqual(await session.getLanes(), [ + { lane: "main", leafId: "child" }, + { lane: "thread", leafId: "child" }, + ]); + }, + ), + + createCase( + factory, + "records and log", + "commits records and lane moves as separate mutations", + async (repository) => { + const session = await repository.create({ id: "session" }); + const root = await session.appendEntry( + { type: "message", id: "root", message: createUserMessage("root") }, + "main", + ); + const finished = await session.appendRecord({ + type: "operation_finished", + id: "finish", + lane: "main", + runId: "run", + outcome: "completed", + }); + + strictEqual(finished.seq, 2); + deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: "root" }]); + await session.moveLane("main", null); + deepStrictEqual(await session.getLanes(), [{ lane: "main", leafId: null }]); + deepStrictEqual(await session.getLog(), [ + { kind: "entry", seq: 1, lane: "main", entry: root }, + { kind: "record", seq: 2, record: finished }, + { kind: "lane", seq: 3, lane: "main", action: "move", leafId: null }, + ]); + + await rejectsWithCode(session.moveLane("main", "missing"), "not_found"); + strictEqual((await session.findRecords()).length, 1); + deepStrictEqual( + (await session.getLog()).map((item) => item.seq), + [1, 2, 3], + ); + }, + ), + + createCase(factory, "entries and lanes", "rejects duplicate ids without changing state", async (repository) => { + const session = await repository.create({ id: "session" }); + await session.appendEntry( + { type: "message", id: "shared", message: createUserMessage("root") }, + "main", + ); + await rejectsWithCode(session.appendRecord(operationStarted("shared")), "already_exists"); + await session.appendRecord(operationStarted("run")); + await rejectsWithCode( + session.appendEntry({ type: "custom", id: "run", customType: "note" }, "main"), + "already_exists", + ); + deepStrictEqual( + (await session.getLog()).map((item) => item.seq), + [1, 2], + ); + }), + + createCase(factory, "entries and lanes", "isolates lanes while sharing the tree", async (repository) => { + const session = await repository.create({ id: "session" }); + await session.appendEntry( + { type: "message", id: "root", message: createUserMessage("root") }, + "main", + ); + await session.createLane("thread", "root"); + await session.appendEntry( + { type: "message", id: "main-child", message: createUserMessage("main") }, + "main", + ); + await session.appendEntry( + { type: "message", id: "thread-child", message: createUserMessage("thread") }, + "thread", + ); + + deepStrictEqual(await session.getLanes(), [ + { lane: "main", leafId: "main-child" }, + { lane: "thread", leafId: "thread-child" }, + ]); + deepStrictEqual(await entryIds(session.findEntriesOnBranch({ start: "main-child", order: "oldestFirst" })), [ + "root", + "main-child", + ]); + deepStrictEqual(await entryIds(session.findEntriesOnBranch({ start: "thread-child", order: "oldestFirst" })), [ + "root", + "thread-child", + ]); + }), + + createCase(factory, "queries and facts", "rejects invalid queries before empty reads", async (repository) => { + const session = await repository.create({ id: "invalid-queries" }); + await session.createLane("thread", null); + const thread = session.view("thread"); + + await rejectsWithCode(session.findEntries({ limit: 0 }), "invalid_query"); + await rejectsWithCode(session.findEntry({ limit: 0 }), "invalid_query"); + await rejectsWithCode(session.findEntriesOnBranch({ limit: 0 }), "invalid_query"); + await rejectsWithCode(thread.findEntriesOnBranch({ cursor: { afterSeq: -1 } }), "invalid_query"); + await rejectsWithCode(thread.findEntryOnBranch({ limit: 0 }), "invalid_query"); + await rejectsWithCode(session.findRecords({ limit: 0 }), "invalid_query"); + await rejectsWithCode(session.getLog({ afterSeq: -1 }), "invalid_query"); + }), + + createCase( + factory, + "queries and facts", + "supports bounded filtered and cursor-based queries", + async (repository) => { + const session = await repository.create({ id: "session" }); + await session.appendEntry( + { type: "message", id: "root", message: createUserMessage("root") }, + "main", + ); + await session.appendEntry( + { type: "custom", id: "old-note", customType: "note", data: 1 }, + "main", + ); + await session.appendEntry( + { type: "compaction", id: "compact", summary: "summary", retainedTail: [], tokensBefore: 10 }, + "main", + ); + await session.appendEntry( + { type: "custom", id: "new-note", customType: "note", data: 2 }, + "main", + ); + await session.appendEntry( + { type: "message", id: "tail", message: createAssistantMessage("tail") }, + "main", + ); + + deepStrictEqual(await entryIds(session.findEntries()), ["tail", "new-note", "compact", "old-note", "root"]); + deepStrictEqual( + await entryIds(session.findEntries({ order: "oldestFirst", cursor: { afterSeq: 2 }, limit: 2 })), + ["compact", "new-note"], + ); + deepStrictEqual(await entryIds(session.findEntries({ customType: "note" })), ["new-note", "old-note"]); + deepStrictEqual( + await entryIds(session.findEntriesOnBranch({ start: "tail", customType: "note", limit: 1 })), + ["new-note"], + ); + deepStrictEqual( + await entryIds( + session.findEntriesOnBranch({ start: "tail", stopAtType: "compaction", type: "message" }), + ), + ["tail"], + ); + deepStrictEqual( + await entryIds(session.findEntriesOnBranch({ start: "tail", stopAtId: "tail", type: "custom" })), + [], + ); + deepStrictEqual( + await entryIds( + session.findEntriesOnBranch({ start: "tail", stopAtType: "custom", order: "oldestFirst" }), + ), + ["root", "old-note"], + ); + await rejectsWithCode(session.findEntries({ limit: 0 }), "invalid_query"); + await rejectsWithCode(session.findEntriesOnBranch({ start: "missing" }), "not_found"); + }, + ), + + createCase( + factory, + "records and log", + "keeps lane names permanent with their recovery records", + async (repository) => { + const session = await repository.create({ id: "session" }); + await session.createLane("thread", null); + await session.appendRecord(operationStarted("old-run", "thread")); + await session.appendRecord({ + type: "queue_enqueued", + id: "old-next-run", + lane: "thread", + queue: "nextRun", + target: { type: "message", id: "queued-message", message: createUserMessage("queued") }, + }); + + deepStrictEqual( + (await session.findRecords({ lane: "thread" })).map((record) => record.id), + ["old-next-run", "old-run"], + ); + deepStrictEqual( + (await session.getLog()).flatMap((item) => (item.kind === "record" ? [item.record.id] : [])), + ["old-run", "old-next-run"], + ); + await rejectsWithCode(session.createLane("thread", null), "already_exists"); + }, + ), + + createCase( + factory, + "records and log", + "persists queue cancellation without consuming its target", + async (repository) => { + const session = await repository.create({ id: "session" }); + const enqueued = await session.appendRecord({ + type: "queue_enqueued", + id: "enqueue", + lane: "main", + queue: "nextRun", + target: { type: "message", id: "queued-message", message: createUserMessage("queued") }, + }); + const cancelled = await session.appendRecord({ + type: "queue_cancelled", + id: "cancel", + lane: "main", + entryId: "queued-message", + }); + deepStrictEqual({ seq: cancelled.seq, entryId: cancelled.entryId }, { seq: 2, entryId: "queued-message" }); + strictEqual("runId" in cancelled, false); + strictEqual(await session.getEntry("queued-message"), undefined); + const cancellations = await session.findRecords({ type: "queue_cancelled" }); + strictEqual(cancellations[0]?.entryId, "queued-message"); + deepStrictEqual(cancellations, [cancelled]); + deepStrictEqual(await session.getLog(), [ + { kind: "record", seq: enqueued.seq, record: enqueued }, + { kind: "record", seq: cancelled.seq, record: cancelled }, + ]); + }, + ), + + createCase( + factory, + "records and log", + "filters records by lane type run sequence and order", + async (repository) => { + const session = await repository.create({ id: "session" }); + await session.appendRecord(operationStarted("run-1")); + await session.appendRecord({ + type: "step_attempt", + id: "attempt-1", + lane: "main", + runId: "run-1", + step: "assistant", + attempt: 1, + resultEntryId: "assistant-1", + }); + await session.createLane("thread", null); + await session.appendRecord(operationStarted("run-2", "thread")); + await session.appendRecord({ + type: "step_attempt", + id: "attempt-2", + lane: "thread", + runId: "run-2", + step: "assistant", + attempt: 1, + resultEntryId: "assistant-2", + }); + + deepStrictEqual( + (await session.findRecords({ lane: "thread" })).map((record) => record.id), + ["attempt-2", "run-2"], + ); + deepStrictEqual( + (await session.findRecords({ type: "step_attempt", order: "oldestFirst" })).map((record) => record.id), + ["attempt-1", "attempt-2"], + ); + deepStrictEqual( + (await session.findRecords({ runId: "run-1", afterSeq: 1 })).map((record) => record.id), + ["attempt-1"], + ); + deepStrictEqual( + (await session.findRecords({ limit: 1 })).map((record) => record.id), + ["attempt-2"], + ); + }, + ), + + createCase( + factory, + "queries and facts", + "keeps latest-value facts and computes ledger statistics across lanes", + async (repository) => { + const session = await repository.create({ id: "session" }); + const assistant = createAssistantMessage("answer"); + if (assistant.role !== "assistant") throw new Error("Expected assistant message"); + assistant.usage = { + input: 10, + output: 5, + cacheRead: 3, + cacheWrite: 2, + totalTokens: 20, + cost: { input: 1, output: 2, cacheRead: 3, cacheWrite: 4, total: 10 }, + }; + await session.appendEntry( + { type: "message", id: "user", message: createUserMessage("question") }, + "main", + ); + await session.appendEntry({ type: "message", id: "assistant", message: assistant }, "main"); + await session.appendRecord({ + type: "usage", + id: "assistant-usage", + lane: "main", + cause: "assistant", + runId: "run", + entryId: "assistant", + attempt: 1, + stopReason: "stop", + usage: assistant.usage, + }); + await session.appendRecord({ + type: "usage", + id: "deferred-usage", + lane: "main", + cause: "deferred_fetch", + runId: "run", + entryId: "deferred-result", + attempt: 1, + stopReason: "deferred", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + }); + await session.createLane("thread", "assistant"); + await session.appendRecord({ + type: "usage", + id: "correction", + lane: "thread", + cause: "adjustment", + details: { reason: "provider correction" }, + usage: { + input: -2, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: -2, + cost: { input: -0.5, output: 0, cacheRead: 0, cacheWrite: 0, total: -0.5 }, + }, + }); + await session.setName("First"); + await session.setName("Second"); + await session.setLabel("user", "keep"); + await session.setLabel("user", undefined); + + strictEqual(await session.getName(), "Second"); + strictEqual(await session.getLabel("user"), undefined); + const usageRecords = await session.findRecords({ type: "usage", order: "oldestFirst" }); + deepStrictEqual( + usageRecords.map((record) => record.cause), + ["assistant", "deferred_fetch", "adjustment"], + ); + const deferredUsage = usageRecords.find((record) => record.cause === "deferred_fetch"); + if (deferredUsage?.cause !== "deferred_fetch") throw new Error("Expected deferred usage record"); + strictEqual(deferredUsage.stopReason, "deferred"); + deepStrictEqual(await session.getStats(), { + messageCount: 2, + cachedTokens: 3, + uncachedTokens: 10, + totalTokens: 18, + costTotal: 9.5, + }); + }, + ), + + createCase(factory, "validation and immutability", "returns immutable copies from reads", async (repository) => { + const session = await repository.create({ id: "immutable" }); + const metadata = await session.getMetadata(); + const data = { nested: { value: 1 } }; + await session.appendEntry({ type: "custom", id: "custom", customType: "note", data }, "main"); + data.nested.value = 50; + const read = await session.getEntry("custom"); + if (read?.type !== "custom") throw new Error("Expected custom entry"); + (read.data as { nested: { value: number } }).nested.value = 99; + const readMetadata = await session.getMetadata(); + readMetadata.id = "changed"; + const log = await session.getLog(); + if (log[0]?.kind !== "entry" || log[0].entry.type !== "custom") throw new Error("Expected entry log"); + (log[0].entry.data as { nested: { value: number } }).nested.value = 100; + + deepStrictEqual(await session.getMetadata(), metadata); + deepStrictEqual(await session.getEntry("custom"), { + type: "custom", + id: "custom", + customType: "note", + data: { nested: { value: 1 } }, + parentId: null, + seq: 1, + timestamp: read.timestamp, + }); + }), + + createCase(factory, "entries and lanes", "validates lane lifecycle and targets", async (repository) => { + const session = await repository.create({ id: "session" }); + await rejectsWithCode(session.createLane("main", null), "already_exists"); + await rejectsWithCode(session.createLane("thread", "missing"), "not_found"); + await rejectsWithCode(session.moveLane("missing", null), "invalid_lane"); + }), + + createCase(factory, "entries and lanes", "binds lane views without caching leaves", async (repository) => { + const session = await repository.create({ id: "session" }); + const root = await session.appendMessage(createUserMessage("root")); + await session.createLane("thread", root); + const thread = session.view("thread"); + const [mainChild, threadChild] = await Promise.all([ + session.appendMessage(createUserMessage("main")), + thread.appendMessage(createUserMessage("thread")), + ]); + + strictEqual(await session.getLeafId(), mainChild); + strictEqual(await thread.getLeafId(), threadChild); + deepStrictEqual(await entryIds(session.findEntriesOnBranch({ order: "oldestFirst" })), [root, mainChild]); + deepStrictEqual(await entryIds(thread.findEntriesOnBranch({ order: "oldestFirst" })), [root, threadChild]); + const empty = await repository.create({ id: "empty" }); + deepStrictEqual(await empty.findEntriesOnBranch(), []); + }), + + createCase( + factory, + "entries and lanes", + "appends provisioned entries with their existing ids", + async (repository) => { + const session = await repository.create({ id: "session" }); + const entry = await session.appendEntry( + { type: "custom", id: "provisioned", customType: "note", data: { value: 1 } }, + "main", + ); + + strictEqual(entry.customType, "note"); + deepStrictEqual( + { id: entry.id, parentId: entry.parentId, seq: entry.seq }, + { id: "provisioned", parentId: null, seq: 1 }, + ); + strictEqual(await session.getLeafId(), "provisioned"); + }, + ), + + createCase(factory, "entries and lanes", "persists tool-result termination decisions", async (repository) => { + const session = await repository.create({ id: "session" }); + const entry = await session.appendEntry( + { + type: "message", + id: "tool-result", + message: { + role: "toolResult", + toolCallId: "call-1", + toolName: "example", + content: [{ type: "text", text: "done" }], + isError: false, + timestamp: 1, + }, + terminate: true, + }, + "main", + ); + + strictEqual(entry.terminate, true); + const stored = await session.getEntry(entry.id); + if (stored?.type !== "message") throw new Error("Expected message entry"); + strictEqual(stored.terminate, true); + deepStrictEqual(await session.findEntries(), [entry]); + deepStrictEqual(await session.getLog(), [{ kind: "entry", seq: entry.seq, lane: "main", entry }]); + }), + + createCase( + factory, + "validation and immutability", + "rejects non-JSON entries before storage mutation", + async (repository) => { + const session = await repository.create({ id: "session" }); + const cyclic: { self?: unknown } = {}; + cyclic.self = cyclic; + + for (const data of [ + { value: undefined }, + [undefined], + { value: 1n }, + { value: Number.NaN }, + { value: new Map() }, + cyclic, + ]) { + await rejectsWithCode(session.appendCustomEntry("invalid", data), "invalid_payload"); + } + + strictEqual(await session.getLeafId(), null); + deepStrictEqual(await session.findEntries(), []); + deepStrictEqual(await session.getLog(), []); + const validId = await session.appendCustomEntry("valid", { value: 1 }); + strictEqual((await session.getEntry(validId))?.seq, 1); + }, + ), + + createCase( + factory, + "validation and immutability", + "rejects non-JSON records before storage mutation", + async (repository) => { + const session = await repository.create({ id: "session" }); + for (const [id, value] of [ + ["undefined-record", undefined], + ["bigint-record", 1n], + ] as const) { + await rejectsWithCode( + session.appendRecord({ + type: "tool_started", + id, + lane: "main", + runId: "run", + assistantEntryId: "assistant", + toolIndex: 0, + toolCallId: "call", + toolName: "example", + effectiveArgs: { value }, + resultEntryId: "result", + replay: "never", + }), + "invalid_payload", + ); + } + + deepStrictEqual(await session.findRecords(), []); + deepStrictEqual(await session.getLog(), []); + strictEqual((await session.appendRecord(operationStarted("valid-record"))).seq, 1); + }, + ), + + createCase(factory, "entries and lanes", "linearizes concurrent writes across two lanes", async (repository) => { + const session = await repository.create({ id: "session" }); + await session.appendEntry( + { type: "message", id: "root", message: createUserMessage("root") }, + "main", + ); + await session.createLane("thread", "root"); + const completionOrder: string[] = []; + const writes = [ + session.appendEntry({ type: "custom", id: "main-1", customType: "note" }, "main"), + session.appendEntry({ type: "custom", id: "thread-1", customType: "note" }, "thread"), + session.appendEntry({ type: "custom", id: "main-2", customType: "note" }, "main"), + session.appendEntry({ type: "custom", id: "thread-2", customType: "note" }, "thread"), + ].map((write) => + write.then((entry) => { + completionOrder.push(entry.id); + return entry; + }), + ); + const entries = await Promise.all(writes); + const commitOrder = [...entries].sort((left, right) => left.seq - right.seq).map((entry) => entry.id); + + strictEqual(new Set(entries.map((entry) => entry.seq)).size, entries.length); + deepStrictEqual(completionOrder, commitOrder); + const concurrentIds = new Set(entries.map((entry) => entry.id)); + deepStrictEqual( + (await session.getLog()).flatMap((item) => + item.kind === "entry" && concurrentIds.has(item.entry.id) ? [item.entry.id] : [], + ), + commitOrder, + ); + const sequences = (await session.getLog()).map((item) => item.seq); + deepStrictEqual( + sequences, + [...sequences].sort((left, right) => left - right), + ); + }), + + createCase(factory, "repository and forks", "creates lists and opens sessions", async (repository) => { + const session = await repository.create({ id: "one" }); + const entryId = await session.appendMessage(createUserMessage("persisted")); + const metadata = await session.getMetadata(); + + deepStrictEqual(await repository.list(), [metadata]); + deepStrictEqual(await entryIds((await repository.open(metadata)).findEntries()), [entryId]); + await rejectsWithCode(repository.create({ id: "one" }), "already_exists"); + }), + + createCase(factory, "repository and forks", "deletes sessions idempotently", async (repository) => { + const session = await repository.create({ id: "one" }); + const metadata = await session.getMetadata(); + + await repository.delete(metadata); + await rejectsWithCode(repository.open(metadata), "not_found"); + await repository.delete(metadata); + }), + + createCase( + factory, + "repository and forks", + "forks one branch with selected facts and no records", + async (repository) => { + const source = await repository.create({ id: "source" }); + const root = await source.appendMessage(createUserMessage("root")); + const shared = await source.appendMessage(createAssistantMessage("shared")); + await source.createLane("thread", shared); + const threadChild = await source.view("thread").appendMessage(createUserMessage("thread")); + const mainChild = await source.appendMessage(createUserMessage("main")); + await source.setName("Source"); + await source.setLabel(shared, "copied"); + await source.setLabel(threadChild, "excluded"); + await source.appendRecord(operationStarted("run")); + await source.appendRecord({ + type: "usage", + id: "source-usage", + lane: "main", + cause: "adjustment", + usage: { + input: 10, + output: 5, + cacheRead: 3, + cacheWrite: 2, + totalTokens: 20, + cost: { input: 1, output: 2, cacheRead: 3, cacheWrite: 4, total: 10 }, + }, + }); + + const fork = await repository.fork(await source.getMetadata(), { + scope: "branch", + entryId: mainChild, + position: "at", + id: "branch-fork", + }); + + deepStrictEqual(await entryIds(fork.findEntries({ order: "oldestFirst" })), [root, shared, mainChild]); + deepStrictEqual(await fork.getLanes(), [{ lane: "main", leafId: mainChild }]); + strictEqual(await fork.getName(), "Source"); + strictEqual(await fork.getLabel(shared), "copied"); + strictEqual(await fork.getLabel(threadChild), undefined); + deepStrictEqual(await fork.findRecords(), []); + deepStrictEqual(await fork.getStats(), { + messageCount: 0, + cachedTokens: 0, + uncachedTokens: 0, + totalTokens: 0, + costTotal: 0, + }); + const metadata = await fork.getMetadata(); + deepStrictEqual( + { id: metadata.id, parentSessionId: metadata.parentSessionId }, + { id: "branch-fork", parentSessionId: "source" }, + ); + }, + ), + + createCase(factory, "repository and forks", "forks a complete tree with lanes and facts", async (repository) => { + const source = await repository.create({ id: "source" }); + const root = await source.appendMessage(createUserMessage("root")); + await source.createLane("thread", root); + const mainChild = await source.appendMessage(createUserMessage("main")); + const threadChild = await source.view("thread").appendMessage(createUserMessage("thread")); + await source.setLabel(threadChild, "thread-tip"); + + const fork = await repository.fork(await source.getMetadata(), { scope: "tree", id: "tree-fork" }); + deepStrictEqual(await entryIds(fork.findEntries({ order: "oldestFirst" })), [root, mainChild, threadChild]); + deepStrictEqual(await fork.getLanes(), [ + { lane: "main", leafId: mainChild }, + { lane: "thread", leafId: threadChild }, + ]); + strictEqual(await fork.getLabel(threadChild), "thread-tip"); + deepStrictEqual( + (await fork.getLog()).filter((item) => item.kind === "lane"), + [ + { kind: "lane", seq: 4, lane: "main", action: "move", leafId: mainChild }, + { kind: "lane", seq: 5, lane: "thread", action: "create", leafId: threadChild }, + ], + ); + }), + + createCase( + factory, + "repository and forks", + "forks before an entry without modifying the source", + async (repository) => { + const source = await repository.create({ id: "source" }); + const root = await source.appendMessage(createUserMessage("root")); + const tail = await source.appendMessage(createUserMessage("tail")); + const fork = await repository.fork(await source.getMetadata(), { entryId: tail, id: "fork" }); + + deepStrictEqual(await entryIds(fork.findEntries({ order: "oldestFirst" })), [root]); + strictEqual(await fork.getLeafId(), root); + strictEqual(await source.getLeafId(), tail); + const beforeDefaultTarget = await repository.fork(await source.getMetadata(), { + position: "before", + id: "before-default-target", + }); + deepStrictEqual(await entryIds(beforeDefaultTarget.findEntries({ order: "oldestFirst" })), [root]); + strictEqual(await beforeDefaultTarget.getLeafId(), root); + + const atDefaultTarget = await repository.fork(await source.getMetadata(), { + position: "at", + id: "at-default-target", + }); + deepStrictEqual(await entryIds(atDefaultTarget.findEntries({ order: "oldestFirst" })), [root, tail]); + strictEqual(await atDefaultTarget.getLeafId(), tail); + await rejectsWithCode( + repository.fork(await source.getMetadata(), { entryId: "missing" }), + "invalid_fork_target", + ); + }, + ), + + createCase(factory, "repository and forks", "validates the default fork target", async (repository) => { + const source = await repository.create({ id: "source-with-custom-leaf" }); + await source.appendCustomEntry("not-a-message"); + + await rejectsWithCode(repository.fork(await source.getMetadata(), { id: "fork" }), "invalid_fork_target"); + }), + ]; +} diff --git a/packages/agent/src/harness/experimental/session/testing/index.ts b/packages/agent/src/harness/experimental/session/testing/index.ts new file mode 100644 index 000000000..8812edc63 --- /dev/null +++ b/packages/agent/src/harness/experimental/session/testing/index.ts @@ -0,0 +1,6 @@ +export { createSessionBackendConformance } from "./conformance.ts"; +export type { + SessionBackendConformanceCase, + SessionBackendFixture, + SessionBackendFixtureFactory, +} from "./types.ts"; diff --git a/packages/agent/src/harness/experimental/session/testing/types.ts b/packages/agent/src/harness/experimental/session/testing/types.ts new file mode 100644 index 000000000..83e8d091f --- /dev/null +++ b/packages/agent/src/harness/experimental/session/testing/types.ts @@ -0,0 +1,16 @@ +import type { SessionRepo } from "../types.ts"; + +/** A fresh backend instance owned by one conformance case. */ +export interface SessionBackendFixture extends AsyncDisposable { + readonly repository: SessionRepo; +} + +/** Creates an isolated fixture for one conformance case. */ +export type SessionBackendFixtureFactory = () => Promise; + +/** A runner-independent conformance case that can be registered with any test framework. */ +export interface SessionBackendConformanceCase { + readonly group: string; + readonly name: string; + run(): Promise; +} diff --git a/packages/agent/src/harness/experimental/session/types.ts b/packages/agent/src/harness/experimental/session/types.ts new file mode 100644 index 000000000..acba16e06 --- /dev/null +++ b/packages/agent/src/harness/experimental/session/types.ts @@ -0,0 +1,361 @@ +import type { StopReason, Usage } from "@earendil-works/pi-ai"; +import type { AgentMessage } from "../../../types.ts"; +import type { Session } from "./session.ts"; + +export type JsonValue = null | boolean | number | string | JsonValue[] | { [key: string]: JsonValue }; + +export type SessionStopReason = Exclude | "deferred"; + +export interface IdGenerator { + next(): string; +} + +export interface EntryBase { + type: string; + id: string; + seq: number; + parentId: string | null; + timestamp: number; +} + +export interface MessageEntry extends EntryBase { + type: "message"; + message: AgentMessage; + terminate?: true; +} + +export interface ModelChangeEntry extends EntryBase { + type: "model_change"; + provider: string; + modelId: string; +} + +export interface ThinkingLevelChangeEntry extends EntryBase { + type: "thinking_level_change"; + thinkingLevel: string; +} + +export interface ActiveToolsChangeEntry extends EntryBase { + type: "active_tools_change"; + activeToolNames: string[]; +} + +export interface CompactionEntry extends EntryBase { + type: "compaction"; + summary: string; + retainedTail: AgentMessage[]; + tokensBefore: number; + details?: unknown; + usage?: Usage; +} + +export interface BranchSummaryEntry extends EntryBase { + type: "branch_summary"; + fromId: string; + summary: string; + details?: unknown; + usage?: Usage; +} + +export interface CustomEntry extends EntryBase { + type: "custom"; + customType: string; + data?: unknown; +} + +export type Entry = + | MessageEntry + | ModelChangeEntry + | ThinkingLevelChangeEntry + | ActiveToolsChangeEntry + | CompactionEntry + | BranchSummaryEntry + | CustomEntry; + +export type ProvisionedEntry = TEntry extends Entry + ? Omit + : never; + +export interface RecordBase { + id: string; + seq: number; + lane: string; + timestamp: number; +} + +export interface OperationStartedRecord extends RecordBase { + type: "operation_started"; + sourceLeafId: string | null; + intent: + | { + kind: "run"; + /** Normalized caller input before before_run; kept for suspended operations and before_resume. */ + originalPrompt: AgentMessage[]; + /** Captured nextRun items, then the prompt, then before_run injections. */ + initialMessages: ProvisionedEntry[]; + systemPromptOverride?: string; + resumeData?: { [extensionId: string]: JsonValue }; + } + | { + kind: "compaction"; + customInstructions?: string; + resultEntryId: string; + } + | { + kind: "navigation"; + targetId: string | null; + summarize: boolean; + customInstructions?: string; + label?: string; + summaryEntryId?: string; + }; +} + +export interface AbortRequestedRecord extends RecordBase { + type: "abort_requested"; + runId: string; + reason: "user" | "shutdown"; +} + +export interface OperationFinishedRecord extends RecordBase { + type: "operation_finished"; + runId: string; + outcome: "completed" | "aborted" | "failed" | "declined"; + error?: { code: string; message: string }; +} + +export type CompactionReason = "manual" | "threshold" | "overflow"; + +export type StepAttemptRecord = RecordBase & + ( + | { + type: "step_attempt"; + runId: string; + step: "assistant" | "branch_summary"; + attempt: number; + resultEntryId: string; + compactionReason?: never; + } + | { + type: "step_attempt"; + runId: string; + step: "compaction"; + attempt: number; + resultEntryId: string; + /** Persists why compaction summary generation started so recovery resumes the same work. */ + compactionReason: CompactionReason; + } + ); + +export interface ToolStartedRecord extends RecordBase { + type: "tool_started"; + runId: string; + assistantEntryId: string; + toolIndex: number; + toolCallId: string; + toolName: string; + effectiveArgs: { [key: string]: unknown }; + resultEntryId: string; + replay: "never" | "safe"; +} + +export type QueueEnqueuedRecord = RecordBase & + ( + | { + type: "queue_enqueued"; + queue: "steer" | "followUp"; + runId: string; + target: ProvisionedEntry; + } + | { + type: "queue_enqueued"; + queue: "nextRun"; + runId?: never; + target: ProvisionedEntry; + } + ); + +export interface QueueCancelledRecord extends RecordBase { + type: "queue_cancelled"; + runId?: string; + entryId: string; +} + +export interface WriteDeferredRecord extends RecordBase { + type: "write_deferred"; + runId: string; + target: ProvisionedEntry; +} + +export type UsageRecord = RecordBase & { type: "usage"; usage: Usage } & ( + | { + cause: "assistant" | "compaction" | "branch_summary" | "deferred_fetch"; + runId: string; + entryId: string; + attempt: number; + stopReason: SessionStopReason; + } + | { cause: "tool"; runId: string; entryId: string; toolCallId: string } + | { cause: "hook"; runId: string; entryId: string } + | { cause: "adjustment"; runId?: string; entryId?: string; details?: JsonValue } + ); + +export type LaneRecord = + | OperationStartedRecord + | AbortRequestedRecord + | OperationFinishedRecord + | StepAttemptRecord + | ToolStartedRecord + | QueueEnqueuedRecord + | QueueCancelledRecord + | WriteDeferredRecord + | UsageRecord; +export type NewRecord = TRecord extends LaneRecord + ? Omit + : never; + +export type EntryOrder = "newestFirst" | "oldestFirst"; + +export interface EntryCursor { + afterSeq: number; +} + +export interface EntryQuery { + type?: Entry["type"]; + customType?: string; + order?: EntryOrder; + limit?: number; + cursor?: EntryCursor; +} + +export interface BranchBounds { + start?: string; + stopAtType?: Entry["type"]; + stopAtId?: string; +} + +export interface RecordQuery { + lane?: string; + type?: LaneRecord["type"]; + runId?: string; + afterSeq?: number; + order?: EntryOrder; + limit?: number; +} + +export interface SessionMetadata { + id: string; + createdAt: number; + parentSessionId?: string; +} + +export interface SessionStats { + messageCount: number; + cachedTokens: number; + uncachedTokens: number; + totalTokens: number; + costTotal: number; +} + +export interface LanePointer { + lane: string; + leafId: string | null; +} + +export type LogItem = + | { kind: "entry"; seq: number; lane: string; entry: Entry } + | { kind: "record"; seq: number; record: LaneRecord } + | { kind: "lane"; seq: number; lane: string; action: "create" | "move"; leafId: string | null } + | { kind: "fact"; seq: number; fact: "name"; name: string } + | { kind: "fact"; seq: number; fact: "label"; targetId: string; label: string | undefined }; + +export interface LogOptions { + afterSeq?: number; + limit?: number; +} + +export interface SessionStorage { + getMetadata(): Promise; + + // Lanes + getLanes(): Promise<{ lane: string; leafId: string | null }[]>; + createLane(lane: string, at: string | null): Promise; + moveLane(lane: string, to: string | null): Promise; + + // Entries and Records + appendEntry(entry: ProvisionedEntry, lane: string): Promise; + appendRecord(record: NewRecord): Promise; + + // Reads + getEntry(id: string): Promise; + findEntries(query?: EntryQuery): Promise; + /** start is mandatory here; defaulting to a lane's leaf is view sugar. */ + findEntriesOnBranch(query: EntryQuery & BranchBounds & { start: string }): Promise; + findRecords( + query: RecordQuery & { type: K }, + ): Promise[]>; + findRecords(query?: RecordQuery): Promise; + getLog(options?: { afterSeq?: number; limit?: number }): Promise; + + // Global facts + getName(): Promise; + setName(name: string): Promise; + getLabel(id: string): Promise; + setLabel(id: string, label: string | undefined): Promise; + getStats(): Promise; +} + +export interface SessionTree { + getLeafId(): Promise; + getEntry(id: string): Promise; + getStats(): Promise; + getName(): Promise; + setName(name: string): Promise; + getLabel(targetId: string): Promise; + setLabel(targetId: string, label: string | undefined): Promise; + findEntries(query?: EntryQuery): Promise; + findEntry(query?: EntryQuery): Promise; + findEntriesOnBranch(query?: EntryQuery & BranchBounds): Promise; + findEntryOnBranch(query?: EntryQuery & BranchBounds): Promise; + appendMessage(message: AgentMessage): Promise; + appendCustomEntry(customType: string, data?: unknown): Promise; +} + +export interface SessionCreateOptions { + id?: string; + parentSessionId?: string; +} + +export type ForkOptions = { scope?: "branch"; entryId?: string; position?: "before" | "at" } | { scope: "tree" }; + +export interface SessionRepo< + TMetadata extends SessionMetadata = SessionMetadata, + TCreateOptions extends SessionCreateOptions = SessionCreateOptions, + TListOptions = void, +> { + create(options: TCreateOptions): Promise>; + open(metadata: TMetadata): Promise>; + list(options?: TListOptions): Promise; + delete(metadata: TMetadata): Promise; + fork(source: TMetadata, options: ForkOptions & TCreateOptions): Promise>; +} + +export type SessionErrorCode = + | "not_found" + | "already_exists" + | "invalid_entry" + | "invalid_payload" + | "invalid_lane" + | "invalid_query" + | "invalid_fork_target" + | "storage"; + +export class SessionError extends Error { + readonly code: SessionErrorCode; + + constructor(code: SessionErrorCode, message: string, cause?: Error) { + super(message, cause === undefined ? undefined : { cause }); + this.name = "SessionError"; + this.code = code; + } +} diff --git a/packages/agent/test/harness/experimental/session/memory.test.ts b/packages/agent/test/harness/experimental/session/memory.test.ts new file mode 100644 index 000000000..ae280dbb2 --- /dev/null +++ b/packages/agent/test/harness/experimental/session/memory.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from "vitest"; +import { InMemorySessionRepo, InMemorySessionStorage, Session } from "../../../../src/experimental.ts"; +import { + createSessionBackendConformance, + type SessionBackendFixture, +} from "../../../../src/harness/experimental/session/testing/index.ts"; + +const conformance = createSessionBackendConformance(() => + Promise.resolve({ + repository: new InMemorySessionRepo(), + [Symbol.asyncDispose]: () => Promise.resolve(), + }), +); + +describe("InMemorySessionRepo conformance", () => { + for (const group of new Set(conformance.map((testCase) => testCase.group))) { + describe(group, () => { + for (const testCase of conformance.filter((candidate) => candidate.group === group)) { + it(testCase.name, () => testCase.run()); + } + }); + } +}); + +describe("Session with in-memory storage", () => { + it("uses one injectable id generator across lane views", async () => { + let nextId = 0; + const session = new Session(new InMemorySessionStorage({ id: "session", createdAt: 1 }), { + idGenerator: { next: () => `generated-${++nextId}` }, + }); + const mainId = await session.appendCustomEntry("note"); + await session.createLane("thread", mainId); + const threadId = await session.view("thread").appendCustomEntry("note"); + + expect(mainId).toBe("generated-1"); + expect(threadId).toBe("generated-2"); + }); +});