alexion
4def357d09
Imported as a linear vendor commit so forge rebases do not replay the upstream root tree. git-subtree-dir: third_party/pi git-subtree-split: d981de1229ef899957bbe968bc8dcda02a21f477
945 lines
33 KiB
TypeScript
945 lines
33 KiB
TypeScript
import { access, mkdtemp, realpath, rm, writeFile } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import { dirname, isAbsolute, join, relative } from "node:path";
|
|
import * as storedValues from "@earendil-works/pi-agent-core";
|
|
import * as sessionWrites from "@earendil-works/pi-agent-core";
|
|
import { BACKGROUND_CONTEXT } from "@earendil-works/pi-agent-core";
|
|
import { describe, expect, it } from "vitest";
|
|
import type { SqliteDatabase, SqliteDatabaseFactory, SqliteStatement } from "../src/index.ts";
|
|
import { createNodeSqliteFactory, SqliteSessionRepo, sql } from "../src/index.ts";
|
|
|
|
const TEST_LANE_CONFIGURATION = {
|
|
model: { provider: "test", modelId: "test" },
|
|
thinkingLevel: "off",
|
|
activeToolNames: [],
|
|
} satisfies storedValues.LaneConfiguration;
|
|
const IDLE_LANE_STATE = {
|
|
currentOperationId: null,
|
|
lastOperationId: null,
|
|
inbox: [],
|
|
} satisfies storedValues.LaneState;
|
|
|
|
async function withTempDir<T>(run: (directory: string) => Promise<T>): Promise<T> {
|
|
const directory = await mkdtemp(join(tmpdir(), "pi-sqlite-session-"));
|
|
try {
|
|
return await run(directory);
|
|
} finally {
|
|
await rm(directory, { recursive: true, force: true });
|
|
}
|
|
}
|
|
|
|
async function withDb<T>(path: string, run: (db: SqliteDatabase) => Promise<T> | T): Promise<T> {
|
|
const db = await createNodeSqliteFactory().open(path);
|
|
try {
|
|
return await run(db);
|
|
} finally {
|
|
db.close();
|
|
}
|
|
}
|
|
|
|
async function pathExists(path: string): Promise<boolean> {
|
|
return access(path).then(
|
|
() => true,
|
|
() => false,
|
|
);
|
|
}
|
|
|
|
class ForwardingDatabase implements SqliteDatabase {
|
|
readonly source: SqliteDatabase;
|
|
|
|
constructor(source: SqliteDatabase) {
|
|
this.source = source;
|
|
}
|
|
|
|
exec(query: string): void {
|
|
this.source.exec(query);
|
|
}
|
|
|
|
prepare(query: string): SqliteStatement {
|
|
return this.source.prepare(query);
|
|
}
|
|
|
|
transaction<T>(callback: () => T): T {
|
|
return this.source.transaction(callback);
|
|
}
|
|
|
|
close(): void {
|
|
this.source.close();
|
|
}
|
|
}
|
|
|
|
class SnapshotBoundaryStatement implements SqliteStatement {
|
|
readonly source: SqliteStatement;
|
|
readonly afterGet: () => void;
|
|
private called = false;
|
|
|
|
constructor(source: SqliteStatement, afterGet: () => void) {
|
|
this.source = source;
|
|
this.afterGet = afterGet;
|
|
}
|
|
|
|
run(...params: unknown[]) {
|
|
return this.source.run(...params);
|
|
}
|
|
|
|
get<TRow extends object>(...params: unknown[]): TRow | undefined {
|
|
const row = this.source.get<TRow>(...params);
|
|
if (!this.called) {
|
|
this.called = true;
|
|
this.afterGet();
|
|
}
|
|
return row;
|
|
}
|
|
|
|
all<TRow extends object>(...params: unknown[]): TRow[] {
|
|
return this.source.all<TRow>(...params);
|
|
}
|
|
|
|
iterate<TRow extends object>(...params: unknown[]): Iterable<TRow> {
|
|
return this.source.iterate<TRow>(...params);
|
|
}
|
|
}
|
|
|
|
class SnapshotBoundaryDatabase extends ForwardingDatabase {
|
|
readonly afterSnapshotEstablished: () => void;
|
|
|
|
constructor(source: SqliteDatabase, afterSnapshotEstablished: () => void) {
|
|
super(source);
|
|
this.afterSnapshotEstablished = afterSnapshotEstablished;
|
|
}
|
|
|
|
override prepare(query: string): SqliteStatement {
|
|
const statement = this.source.prepare(query);
|
|
return query.includes("FROM sessions")
|
|
? new SnapshotBoundaryStatement(statement, this.afterSnapshotEstablished)
|
|
: statement;
|
|
}
|
|
}
|
|
|
|
class SnapshotBoundaryFactory implements SqliteDatabaseFactory {
|
|
readonly source = createNodeSqliteFactory();
|
|
readonly afterSnapshotEstablished: () => void;
|
|
readOnlyOpenCount = 0;
|
|
|
|
constructor(afterSnapshotEstablished: () => void) {
|
|
this.afterSnapshotEstablished = afterSnapshotEstablished;
|
|
}
|
|
|
|
open(path: string): Promise<SqliteDatabase> {
|
|
return this.source.open(path);
|
|
}
|
|
|
|
openExisting(path: string): Promise<SqliteDatabase> {
|
|
return this.source.openExisting(path);
|
|
}
|
|
|
|
async openReadOnly(path: string): Promise<SqliteDatabase> {
|
|
this.readOnlyOpenCount++;
|
|
return new SnapshotBoundaryDatabase(await this.source.openReadOnly(path), this.afterSnapshotEstablished);
|
|
}
|
|
}
|
|
|
|
function deferred(): { promise: Promise<void>; resolve: () => void } {
|
|
let resolve!: () => void;
|
|
const promise = new Promise<void>((resolvePromise) => {
|
|
resolve = resolvePromise;
|
|
});
|
|
return { promise, resolve };
|
|
}
|
|
|
|
class GatedOpenExistingFactory implements SqliteDatabaseFactory {
|
|
readonly source = createNodeSqliteFactory();
|
|
readonly entered = deferred();
|
|
readonly release = deferred();
|
|
private gated = false;
|
|
|
|
arm(): void {
|
|
this.gated = true;
|
|
}
|
|
|
|
open(path: string): Promise<SqliteDatabase> {
|
|
return this.source.open(path);
|
|
}
|
|
|
|
async openExisting(path: string): Promise<SqliteDatabase> {
|
|
if (this.gated) {
|
|
this.gated = false;
|
|
this.entered.resolve();
|
|
await this.release.promise;
|
|
}
|
|
return this.source.openExisting(path);
|
|
}
|
|
|
|
openReadOnly(path: string): Promise<SqliteDatabase> {
|
|
return this.source.openReadOnly(path);
|
|
}
|
|
}
|
|
|
|
class CloseTrackingDatabase extends ForwardingDatabase {
|
|
closeAttempts = 0;
|
|
closeError: Error | undefined;
|
|
|
|
override close(): void {
|
|
this.closeAttempts++;
|
|
this.source.close();
|
|
if (this.closeError !== undefined) throw this.closeError;
|
|
}
|
|
}
|
|
|
|
class CloseTrackingFactory implements SqliteDatabaseFactory {
|
|
readonly source = createNodeSqliteFactory();
|
|
readonly writableConnections: CloseTrackingDatabase[] = [];
|
|
|
|
async open(path: string): Promise<SqliteDatabase> {
|
|
const db = new CloseTrackingDatabase(await this.source.open(path));
|
|
this.writableConnections.push(db);
|
|
return db;
|
|
}
|
|
|
|
async openExisting(path: string): Promise<SqliteDatabase> {
|
|
const db = new CloseTrackingDatabase(await this.source.openExisting(path));
|
|
this.writableConnections.push(db);
|
|
return db;
|
|
}
|
|
|
|
openReadOnly(path: string): Promise<SqliteDatabase> {
|
|
return this.source.openReadOnly(path);
|
|
}
|
|
}
|
|
|
|
function commitLaterSourceState(db: SqliteDatabase): void {
|
|
const tip = storedValues.branchTip("main");
|
|
const name = storedValues.sessionName;
|
|
const label = storedValues.entryLabel("root");
|
|
db.transaction(() => {
|
|
sql`INSERT INTO entries (session_id, id, parent_id, seq, type, custom_type, timestamp, payload)
|
|
VALUES (${"source"}, ${"child"}, ${"root"}, ${7}, ${"message"}, ${null}, ${2}, ${JSON.stringify({ message: { role: "user", content: "after", timestamp: 2 } })})`.run(
|
|
db,
|
|
);
|
|
sql`INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type)
|
|
VALUES (${"source"}, ${"root"}, ${"child"}, ${7}, ${"message"})`.run(db);
|
|
sql`UPDATE branch_meta SET tip_entry_id = ${"child"}, tip_seq = ${7}
|
|
WHERE session_id = ${"source"} AND branch_id = ${"root"}`.run(db);
|
|
for (const [namespace, key, seq, value] of [
|
|
[tip.namespace, tip.key, 8, "child"],
|
|
[name.namespace, name.key, 9, "after"],
|
|
[label.namespace, label.key, 10, "after-label"],
|
|
] as const) {
|
|
sql`INSERT INTO scalar_values (session_id, namespace, key, seq, value)
|
|
VALUES (${"source"}, ${namespace}, ${key}, ${seq}, ${JSON.stringify(value)})
|
|
ON CONFLICT(session_id, namespace, key) DO UPDATE SET seq = excluded.seq, value = excluded.value`.run(db);
|
|
}
|
|
sql`UPDATE sessions SET message_count = ${1}, next_seq = ${11} WHERE id = ${"source"}`.run(db);
|
|
});
|
|
}
|
|
|
|
describe("SqliteSessionRepo", () => {
|
|
it("creates one branchless initialized session file", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1_700_000_000_000,
|
|
});
|
|
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
const metadata = session.metadata;
|
|
expect(metadata).toMatchObject({
|
|
id: "session",
|
|
createdAt: 1_700_000_000_000,
|
|
storageVersion: 1,
|
|
});
|
|
expect(metadata.path).toBe(await realpath(join(directory, "session.sqlite")));
|
|
|
|
await withDb(metadata.path, (db) => {
|
|
expect(sql`SELECT COUNT(*) AS count FROM sessions`.get<{ count: number }>(db)).toEqual({ count: 1 });
|
|
expect(
|
|
sql`SELECT message_count, usage_payload, next_seq FROM sessions WHERE id = ${"session"}`.get(db),
|
|
).toEqual({
|
|
message_count: 0,
|
|
usage_payload: JSON.stringify({
|
|
input: 0,
|
|
output: 0,
|
|
cacheRead: 0,
|
|
cacheWrite: 0,
|
|
totalTokens: 0,
|
|
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
|
}),
|
|
next_seq: 1,
|
|
});
|
|
expect(
|
|
sql`SELECT namespace, key, seq, value FROM scalar_values WHERE session_id = ${"session"} ORDER BY seq`.all(
|
|
db,
|
|
),
|
|
).toEqual([]);
|
|
expect(sql`SELECT COUNT(*) AS count FROM list_values WHERE session_id = ${"session"}`.get(db)).toEqual({
|
|
count: 0,
|
|
});
|
|
});
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("exposes explicit branch scans through the open-session facade", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1_700_000_000_000,
|
|
});
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
await session.mutate(
|
|
(mutator) =>
|
|
mutator.commit(
|
|
[
|
|
sessionWrites.insertEntry({ id: "root", parentId: null, type: "custom", customType: "root" }),
|
|
sessionWrites.insertEntry({ id: "child", parentId: "root", type: "custom", customType: "child" }),
|
|
],
|
|
BACKGROUND_CONTEXT,
|
|
),
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
|
|
expect(await session.scanBranch({ start: "child", order: "oldestFirst" }, BACKGROUND_CONTEXT)).toMatchObject([
|
|
{ id: "root" },
|
|
{ id: "child" },
|
|
]);
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
await expect(session.scanBranch({ start: "child" }, BACKGROUND_CONTEXT)).rejects.toThrow("Session is closed");
|
|
});
|
|
});
|
|
|
|
it("commits an explicit mutation through the open-session facade", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1_700_000_000_000,
|
|
});
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
const mutation = await session.beginMutation(BACKGROUND_CONTEXT);
|
|
const result = await mutation.commit(
|
|
[storedValues.setValue(storedValues.sessionName, "explicit")],
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
|
|
expect(result.seqs).toHaveLength(1);
|
|
expect(await mutation.getValue(storedValues.sessionName, BACKGROUND_CONTEXT)).toMatchObject({
|
|
value: "explicit",
|
|
});
|
|
await mutation.end(BACKGROUND_CONTEXT);
|
|
expect(await session.getName(BACKGROUND_CONTEXT)).toBe("explicit");
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("rejects duplicate create without deleting the existing database", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
const { metadata } = session;
|
|
|
|
await expect(repo.create({ id: "session" }, BACKGROUND_CONTEXT)).rejects.toThrow();
|
|
await withDb(metadata.path, (db) => {
|
|
expect(sql`SELECT COUNT(*) AS count FROM sessions`.get<{ count: number }>(db)).toEqual({ count: 1 });
|
|
});
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("lists an open session without storage-layer ownership state", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
const { metadata } = session;
|
|
|
|
await expect(repo.list(undefined, BACKGROUND_CONTEXT)).resolves.toMatchObject([
|
|
{ id: "session", path: metadata.path },
|
|
]);
|
|
await withDb(metadata.path, (db) => {
|
|
expect(
|
|
sql`SELECT name FROM sqlite_master WHERE type = ${"table"} AND name = ${"writer_lease"}`.get(db),
|
|
).toBeUndefined();
|
|
});
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("ignores a stale writer_lease table from a pre-WP07 WIP database", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const created = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
await created.close(BACKGROUND_CONTEXT);
|
|
await withDb(created.metadata.path, (db) => {
|
|
db.exec(
|
|
"CREATE TABLE writer_lease (session_id TEXT PRIMARY KEY, owner_id TEXT NOT NULL, fence INTEGER NOT NULL, expires_at_ms INTEGER NOT NULL) WITHOUT ROWID",
|
|
);
|
|
sql`INSERT INTO writer_lease (session_id, owner_id, fence, expires_at_ms)
|
|
VALUES (${"session"}, ${"stale"}, ${7}, ${999})`.run(db);
|
|
});
|
|
|
|
const reopened = await repo.open(created.metadata, BACKGROUND_CONTEXT);
|
|
await reopened.setName("works", BACKGROUND_CONTEXT);
|
|
await reopened.close(BACKGROUND_CONTEXT);
|
|
await withDb(created.metadata.path, (db) => {
|
|
expect(sql`SELECT owner_id, fence FROM writer_lease WHERE session_id = ${"session"}`.get(db)).toEqual({
|
|
owner_id: "stale",
|
|
fence: 7,
|
|
});
|
|
});
|
|
});
|
|
});
|
|
|
|
it("skips corrupt and incompatible files during list discovery", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
await writeFile(join(directory, "corrupt.sqlite"), "not a sqlite database");
|
|
await withDb(session.metadata.path, (db) => {
|
|
sql`UPDATE sessions SET storage_version = ${999} WHERE id = ${"session"}`.run(db);
|
|
});
|
|
|
|
expect(await repo.list(undefined, BACKGROUND_CONTEXT)).toEqual([]);
|
|
});
|
|
});
|
|
|
|
it("does not remove a pre-existing non-database file when create fails", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const path = join(directory, "session.sqlite");
|
|
await writeFile(path, "not a sqlite database");
|
|
|
|
await expect(repo.create({ id: "session" }, BACKGROUND_CONTEXT)).rejects.toThrow();
|
|
await expect(access(path)).resolves.toBeUndefined();
|
|
});
|
|
});
|
|
|
|
it("rejects delete for missing files and deletes a closed session", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
const { metadata } = session;
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
|
|
await expect(
|
|
repo.delete({ ...metadata, path: join(directory, "missing.sqlite") }, BACKGROUND_CONTEXT),
|
|
).rejects.toThrow();
|
|
await repo.delete(metadata, BACKGROUND_CONTEXT);
|
|
await expect(access(metadata.path)).rejects.toThrow();
|
|
});
|
|
});
|
|
|
|
it("closes open sessions through repo close and rejects later operations", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
|
|
await repo.close(BACKGROUND_CONTEXT);
|
|
|
|
await expect(session.getStats(BACKGROUND_CONTEXT)).rejects.toThrow("Session is closed");
|
|
await expect(repo.list(undefined, BACKGROUND_CONTEXT)).rejects.toThrow("SqliteSessionRepo is closed");
|
|
await expect(repo.create({ id: "other" }, BACKGROUND_CONTEXT)).rejects.toThrow("SqliteSessionRepo is closed");
|
|
await expect(repo.close(BACKGROUND_CONTEXT)).resolves.toBeUndefined();
|
|
});
|
|
});
|
|
|
|
it("opens a session through the version gate and rejects a duplicate local handle", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const created = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
const { metadata } = created;
|
|
await created.close(BACKGROUND_CONTEXT);
|
|
|
|
const opened = await repo.open(metadata, BACKGROUND_CONTEXT);
|
|
expect(opened.metadata).toMatchObject({ id: "session", path: metadata.path });
|
|
await expect(repo.open(metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already open");
|
|
await opened.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("isolates sessions stored in one shared SQLite container", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const databasePath = join(directory, "sessions.sqlite");
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databasePath,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1_700_000_000_000,
|
|
});
|
|
const left = await repo.create({ id: "left" }, BACKGROUND_CONTEXT);
|
|
const right = await repo.create({ id: "right" }, BACKGROUND_CONTEXT);
|
|
|
|
expect(left.metadata.path).toBe(await realpath(databasePath));
|
|
expect(right.metadata.path).toBe(await realpath(databasePath));
|
|
await left.mutate(
|
|
(mutator) =>
|
|
mutator.commit(
|
|
[
|
|
sessionWrites.insertEntry({ id: "left-root", parentId: null, type: "custom", customType: "left" }),
|
|
storedValues.setValue(storedValues.sessionName, "left-name"),
|
|
],
|
|
BACKGROUND_CONTEXT,
|
|
),
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
await right.mutate(
|
|
(mutator) =>
|
|
mutator.commit(
|
|
[
|
|
sessionWrites.insertEntry({
|
|
id: "right-root",
|
|
parentId: null,
|
|
type: "custom",
|
|
customType: "right",
|
|
}),
|
|
storedValues.setValue(storedValues.sessionName, "right-name"),
|
|
],
|
|
BACKGROUND_CONTEXT,
|
|
),
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
|
|
expect((await repo.list(undefined, BACKGROUND_CONTEXT)).map((metadata) => metadata.id).sort()).toEqual([
|
|
"left",
|
|
"right",
|
|
]);
|
|
expect(await left.getValue(storedValues.sessionName, BACKGROUND_CONTEXT)).toMatchObject({
|
|
value: "left-name",
|
|
});
|
|
expect(await right.getValue(storedValues.sessionName, BACKGROUND_CONTEXT)).toMatchObject({
|
|
value: "right-name",
|
|
});
|
|
expect((await left.getEntries(["left-root", "right-root"], BACKGROUND_CONTEXT)).has("right-root")).toBe(false);
|
|
expect((await right.getEntries(["left-root", "right-root"], BACKGROUND_CONTEXT)).has("left-root")).toBe(false);
|
|
|
|
await Promise.all([left.close(BACKGROUND_CONTEXT), right.close(BACKGROUND_CONTEXT)]);
|
|
});
|
|
});
|
|
|
|
it("does not copy usage ledger rows when forking", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1_700_000_000_000,
|
|
});
|
|
const source = await repo.create({ id: "source" }, BACKGROUND_CONTEXT);
|
|
await source.mutate(
|
|
(mutator) =>
|
|
mutator.commit(
|
|
[
|
|
sessionWrites.insertEntry({ id: "root", parentId: null, type: "custom", customType: "root" }),
|
|
sessionWrites.insertEntry({
|
|
id: "child",
|
|
parentId: "root",
|
|
type: "message",
|
|
message: { role: "user", content: "child", timestamp: 1 },
|
|
}),
|
|
storedValues.setValue(storedValues.branchTip("main"), "child"),
|
|
storedValues.setValue(storedValues.laneConfig("main"), TEST_LANE_CONFIGURATION),
|
|
storedValues.setValue(storedValues.laneState("main"), IDLE_LANE_STATE),
|
|
sessionWrites.insertUsage({
|
|
id: "usage",
|
|
adjustment: false,
|
|
usage: {
|
|
input: 1,
|
|
output: 1,
|
|
cacheRead: 0,
|
|
cacheWrite: 0,
|
|
totalTokens: 2,
|
|
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
|
|
},
|
|
}),
|
|
],
|
|
BACKGROUND_CONTEXT,
|
|
),
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
|
|
await source.close(BACKGROUND_CONTEXT);
|
|
const fork = await repo.fork(
|
|
source.metadata,
|
|
{ id: "fork", scope: "branch", branch: "main", entryId: "child" },
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
|
|
await withDb(fork.metadata.path, (db) => {
|
|
expect(
|
|
sql`SELECT COUNT(*) AS count FROM usage_ledger WHERE session_id = ${"fork"}`.get<{ count: number }>(db),
|
|
).toEqual({ count: 0 });
|
|
});
|
|
await Promise.all([source.close(BACKGROUND_CONTEXT), fork.close(BACKGROUND_CONTEXT)]);
|
|
});
|
|
});
|
|
|
|
it("does not create databases for missing open, list, fork, or delete targets", async () => {
|
|
await withTempDir(async (root) => {
|
|
const directory = join(root, "missing-directory");
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const missing = {
|
|
id: "missing",
|
|
createdAt: 1,
|
|
storageVersion: 1,
|
|
path: join(directory, "missing.sqlite"),
|
|
};
|
|
|
|
await expect(repo.list(undefined, BACKGROUND_CONTEXT)).resolves.toEqual([]);
|
|
expect(await pathExists(directory)).toBe(false);
|
|
await expect(repo.open(missing, BACKGROUND_CONTEXT)).rejects.toThrow();
|
|
await expect(repo.delete(missing, BACKGROUND_CONTEXT)).rejects.toThrow();
|
|
await expect(repo.fork(missing, { id: "fork", scope: "tree" }, BACKGROUND_CONTEXT)).rejects.toThrow();
|
|
expect(await pathExists(missing.path)).toBe(false);
|
|
expect(await pathExists(join(directory, "fork.sqlite"))).toBe(false);
|
|
});
|
|
});
|
|
|
|
it("reserves deletion against local create, open, and fork destinations", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const databaseFactory = new GatedOpenExistingFactory();
|
|
const repo = new SqliteSessionRepo({ directory, databaseFactory, now: () => 1 });
|
|
const target = await repo.create({ id: "target" }, BACKGROUND_CONTEXT);
|
|
const source = await repo.create({ id: "source" }, BACKGROUND_CONTEXT);
|
|
await expect(repo.delete(target.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already open");
|
|
// The host closes a worker before deletion; only same-repository exclusion is promised here.
|
|
await target.close(BACKGROUND_CONTEXT);
|
|
databaseFactory.arm();
|
|
|
|
const deleting = repo.delete(target.metadata, BACKGROUND_CONTEXT);
|
|
await databaseFactory.entered.promise;
|
|
await expect(repo.create({ id: "target" }, BACKGROUND_CONTEXT)).rejects.toThrow("already open");
|
|
await expect(repo.open(target.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("already open");
|
|
await expect(repo.fork(source.metadata, { id: "target", scope: "tree" }, BACKGROUND_CONTEXT)).rejects.toThrow(
|
|
"already open",
|
|
);
|
|
databaseFactory.release.resolve();
|
|
await deleting;
|
|
await source.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("deletes only the selected Session from a shared container", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const databasePath = join(directory, "shared.sqlite");
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databasePath,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const removed = await repo.create({ id: "removed" }, BACKGROUND_CONTEXT);
|
|
const retained = await repo.create({ id: "retained" }, BACKGROUND_CONTEXT);
|
|
await removed.setName("removed", BACKGROUND_CONTEXT);
|
|
await retained.setName("retained", BACKGROUND_CONTEXT);
|
|
await removed.close(BACKGROUND_CONTEXT);
|
|
|
|
await repo.delete(removed.metadata, BACKGROUND_CONTEXT);
|
|
|
|
expect((await repo.list(undefined, BACKGROUND_CONTEXT)).map(({ id }) => id)).toEqual(["retained"]);
|
|
expect(await retained.getName(BACKGROUND_CONTEXT)).toBe("retained");
|
|
await expect(repo.open(removed.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("Unknown SQLite session");
|
|
await retained.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("removes per-file WAL and SHM sidecars", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
await writeFile(`${session.metadata.path}-wal`, "");
|
|
await writeFile(`${session.metadata.path}-shm`, "");
|
|
|
|
await repo.delete(session.metadata, BACKGROUND_CONTEXT);
|
|
|
|
await expect(access(session.metadata.path)).rejects.toThrow();
|
|
await expect(access(`${session.metadata.path}-wal`)).rejects.toThrow();
|
|
await expect(access(`${session.metadata.path}-shm`)).rejects.toThrow();
|
|
});
|
|
});
|
|
|
|
it("creates the parent of a custom shared-container path", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const databasePath = join(directory, "nested", "containers", "sessions.sqlite");
|
|
const repo = new SqliteSessionRepo({
|
|
directory: join(directory, "unrelated"),
|
|
databasePath,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
|
|
const session = await repo.create({ id: "session" }, BACKGROUND_CONTEXT);
|
|
|
|
expect(session.metadata.path).toBe(await realpath(databasePath));
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("encodes unsafe explicit IDs inside the repository directory and round-trips them", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const repo = new SqliteSessionRepo({
|
|
directory,
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const ids = ["../escape", "slash/id", "back\\slash", "percent%id", "..", "dots...id", "ユニコード"];
|
|
const metadata = [];
|
|
for (const id of ids) {
|
|
const session = await repo.create({ id }, BACKGROUND_CONTEXT);
|
|
metadata.push(session.metadata);
|
|
await session.close(BACKGROUND_CONTEXT);
|
|
}
|
|
const canonicalDirectory = await realpath(directory);
|
|
for (const stored of metadata) {
|
|
const fromDirectory = relative(canonicalDirectory, stored.path);
|
|
expect(isAbsolute(fromDirectory)).toBe(false);
|
|
expect(fromDirectory.startsWith(".."), stored.id).toBe(false);
|
|
expect(dirname(stored.path)).toBe(canonicalDirectory);
|
|
}
|
|
expect((await repo.list(undefined, BACKGROUND_CONTEXT)).map(({ id }) => id).sort()).toEqual([...ids].sort());
|
|
for (const stored of metadata) {
|
|
const reopened = await repo.open(stored, BACKGROUND_CONTEXT);
|
|
expect(reopened.metadata.id).toBe(stored.id);
|
|
await reopened.close(BACKGROUND_CONTEXT);
|
|
}
|
|
const fork = await repo.fork(metadata[0]!, { id: "fork/../%", scope: "tree" }, BACKGROUND_CONTEXT);
|
|
expect(fork.metadata).toMatchObject({ id: "fork/../%", parentSessionId: "../escape" });
|
|
expect(dirname(fork.metadata.path)).toBe(canonicalDirectory);
|
|
await fork.close(BACKGROUND_CONTEXT);
|
|
});
|
|
});
|
|
|
|
it("uses exact physical identity for foreign fork sources and rejects foreign writable metadata", async () => {
|
|
await withTempDir(async (root) => {
|
|
const leftRepo = new SqliteSessionRepo({
|
|
directory: join(root, "left"),
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const rightRepo = new SqliteSessionRepo({
|
|
directory: join(root, "right"),
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const left = await leftRepo.create({ id: "same" }, BACKGROUND_CONTEXT);
|
|
const right = await rightRepo.create({ id: "same" }, BACKGROUND_CONTEXT);
|
|
await left.mutate(
|
|
(mutator) =>
|
|
mutator.commit(
|
|
[
|
|
sessionWrites.insertEntry({ id: "left-root", parentId: null, type: "custom", customType: "left" }),
|
|
storedValues.setValue(storedValues.branchTip("main"), "left-root"),
|
|
storedValues.setValue(storedValues.laneConfig("main"), TEST_LANE_CONFIGURATION),
|
|
storedValues.setValue(storedValues.laneState("main"), IDLE_LANE_STATE),
|
|
storedValues.setValue(storedValues.sessionName, "left"),
|
|
],
|
|
BACKGROUND_CONTEXT,
|
|
),
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
await right.mutate(
|
|
(mutator) =>
|
|
mutator.commit(
|
|
[
|
|
sessionWrites.insertEntry({
|
|
id: "right-root",
|
|
parentId: null,
|
|
type: "custom",
|
|
customType: "right",
|
|
}),
|
|
storedValues.setValue(storedValues.branchTip("main"), "right-root"),
|
|
storedValues.setValue(storedValues.sessionName, "right"),
|
|
],
|
|
BACKGROUND_CONTEXT,
|
|
),
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
|
|
const fork = await rightRepo.fork(
|
|
left.metadata,
|
|
{ id: "fork-left", scope: "branch", branch: "main" },
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
expect((await fork.findEntries({ order: "asc" }, BACKGROUND_CONTEXT)).map(({ id }) => id)).toEqual([
|
|
"left-root",
|
|
]);
|
|
expect(await fork.getName(BACKGROUND_CONTEXT)).toBe("left");
|
|
await right.close(BACKGROUND_CONTEXT);
|
|
await expect(rightRepo.open(left.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("outside this repository");
|
|
await expect(rightRepo.delete(left.metadata, BACKGROUND_CONTEXT)).rejects.toThrow("outside this repository");
|
|
await Promise.all([left.close(BACKGROUND_CONTEXT), fork.close(BACKGROUND_CONTEXT)]);
|
|
});
|
|
});
|
|
|
|
for (const layout of ["per-file", "shared"] as const) {
|
|
it(`forks a live ${layout} source through one coherent read-only WAL snapshot`, async () => {
|
|
await withTempDir(async (directory) => {
|
|
const databasePath = layout === "shared" ? join(directory, "shared.sqlite") : undefined;
|
|
const workerRepo = new SqliteSessionRepo({
|
|
directory,
|
|
...(databasePath === undefined ? {} : { databasePath }),
|
|
databaseFactory: createNodeSqliteFactory(),
|
|
now: () => 1,
|
|
});
|
|
const source = await workerRepo.create({ id: "source" }, BACKGROUND_CONTEXT);
|
|
await source.mutate(
|
|
(mutator) =>
|
|
mutator.commit(
|
|
[
|
|
sessionWrites.insertEntry({ id: "root", parentId: null, type: "custom", customType: "root" }),
|
|
storedValues.setValue(storedValues.branchTip("main"), "root"),
|
|
storedValues.setValue(storedValues.laneConfig("main"), TEST_LANE_CONFIGURATION),
|
|
storedValues.setValue(storedValues.laneState("main"), IDLE_LANE_STATE),
|
|
storedValues.setValue(storedValues.sessionName, "before"),
|
|
storedValues.setValue(storedValues.entryLabel("root"), "before-label"),
|
|
],
|
|
BACKGROUND_CONTEXT,
|
|
),
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
const writerDb = await createNodeSqliteFactory().openExisting(source.metadata.path);
|
|
writerDb.exec("PRAGMA busy_timeout = 5000;");
|
|
let laterCommitCompleted = false;
|
|
const databaseFactory = new SnapshotBoundaryFactory(() => {
|
|
if (laterCommitCompleted) return;
|
|
commitLaterSourceState(writerDb);
|
|
laterCommitCompleted = true;
|
|
});
|
|
const serverRepo = new SqliteSessionRepo({
|
|
directory,
|
|
...(databasePath === undefined ? {} : { databasePath }),
|
|
databaseFactory,
|
|
now: () => 2,
|
|
});
|
|
|
|
const first = await serverRepo.fork(
|
|
source.metadata,
|
|
{ id: "first-fork", scope: "branch", branch: "main" },
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
|
|
expect(laterCommitCompleted).toBe(true);
|
|
expect(databaseFactory.readOnlyOpenCount).toBe(1);
|
|
expect((await first.findEntries({ order: "asc" }, BACKGROUND_CONTEXT)).map(({ id }) => id)).toEqual([
|
|
"root",
|
|
]);
|
|
expect(await first.getName(BACKGROUND_CONTEXT)).toBe("before");
|
|
expect(await first.getLabel("root", BACKGROUND_CONTEXT)).toBe("before-label");
|
|
await expect((await first.branch("main", BACKGROUND_CONTEXT))?.getTipId(BACKGROUND_CONTEXT)).resolves.toBe(
|
|
"root",
|
|
);
|
|
expect((await first.getStats(BACKGROUND_CONTEXT)).messageCount).toBe(0);
|
|
|
|
const second = await serverRepo.fork(
|
|
source.metadata,
|
|
{ id: "second-fork", scope: "branch", branch: "main" },
|
|
BACKGROUND_CONTEXT,
|
|
);
|
|
expect(databaseFactory.readOnlyOpenCount).toBe(2);
|
|
expect((await second.findEntries({ order: "asc" }, BACKGROUND_CONTEXT)).map(({ id }) => id)).toEqual([
|
|
"root",
|
|
"child",
|
|
]);
|
|
expect(await second.getName(BACKGROUND_CONTEXT)).toBe("after");
|
|
expect(await second.getLabel("root", BACKGROUND_CONTEXT)).toBe("after-label");
|
|
await expect((await second.branch("main", BACKGROUND_CONTEXT))?.getTipId(BACKGROUND_CONTEXT)).resolves.toBe(
|
|
"child",
|
|
);
|
|
expect((await second.getStats(BACKGROUND_CONTEXT)).messageCount).toBe(1);
|
|
|
|
writerDb.close();
|
|
await Promise.all([
|
|
source.close(BACKGROUND_CONTEXT),
|
|
first.close(BACKGROUND_CONTEXT),
|
|
second.close(BACKGROUND_CONTEXT),
|
|
]);
|
|
});
|
|
});
|
|
}
|
|
|
|
it("waits for every Session close and reports one or multiple failures after settlement", async () => {
|
|
await withTempDir(async (directory) => {
|
|
const multipleFactory = new CloseTrackingFactory();
|
|
const multipleRepo = new SqliteSessionRepo({
|
|
directory: join(directory, "multiple"),
|
|
databaseFactory: multipleFactory,
|
|
now: () => 1,
|
|
});
|
|
await multipleRepo.create({ id: "first" }, BACKGROUND_CONTEXT);
|
|
await multipleRepo.create({ id: "second" }, BACKGROUND_CONTEXT);
|
|
await multipleRepo.create({ id: "third" }, BACKGROUND_CONTEXT);
|
|
const firstError = new Error("first close failed");
|
|
const secondError = new Error("second close failed");
|
|
multipleFactory.writableConnections[0]!.closeError = firstError;
|
|
multipleFactory.writableConnections[1]!.closeError = secondError;
|
|
|
|
const multipleClose = multipleRepo.close(BACKGROUND_CONTEXT);
|
|
expect(multipleRepo.close(BACKGROUND_CONTEXT)).toBe(multipleClose);
|
|
let multipleFailure: unknown;
|
|
try {
|
|
await multipleClose;
|
|
} catch (error) {
|
|
multipleFailure = error;
|
|
}
|
|
expect(multipleFailure).toBeInstanceOf(AggregateError);
|
|
if (!(multipleFailure instanceof AggregateError)) throw new Error("Expected AggregateError");
|
|
expect(multipleFailure.errors).toEqual([firstError, secondError]);
|
|
expect(multipleFactory.writableConnections.map(({ closeAttempts }) => closeAttempts)).toEqual([1, 1, 1]);
|
|
|
|
const singleFactory = new CloseTrackingFactory();
|
|
const singleRepo = new SqliteSessionRepo({
|
|
directory: join(directory, "single"),
|
|
databaseFactory: singleFactory,
|
|
now: () => 1,
|
|
});
|
|
await singleRepo.create({ id: "first" }, BACKGROUND_CONTEXT);
|
|
await singleRepo.create({ id: "second" }, BACKGROUND_CONTEXT);
|
|
const singleError = new Error("single close failed");
|
|
singleFactory.writableConnections[0]!.closeError = singleError;
|
|
|
|
await expect(singleRepo.close(BACKGROUND_CONTEXT)).rejects.toBe(singleError);
|
|
expect(singleFactory.writableConnections.map(({ closeAttempts }) => closeAttempts)).toEqual([1, 1]);
|
|
});
|
|
});
|
|
});
|