fix(agent): harden bounded branch queries

This commit is contained in:
Christian Klotz 2026-08-01 20:54:36 +03:00
parent 9b75872017
commit 081d7fbfca
No known key found for this signature in database
5 changed files with 443 additions and 25 deletions

View file

@ -51,7 +51,7 @@ export function createArraySessionReader<TMetadata extends SessionMetadata>(
throw new RangeError("Session branch query limit must be a positive integer");
}
if (query.start === null) return [];
const path: SessionTreeEntry[] = [];
const pathFromStart: SessionTreeEntry[] = [];
const visited = new Set<string>();
let current = byId.get(query.start);
if (!current) throw new SessionError("not_found", `Entry ${query.start} not found`);
@ -60,15 +60,22 @@ export function createArraySessionReader<TMetadata extends SessionMetadata>(
throw new SessionError("invalid_session", `Session branch contains a cycle at ${current.id}`);
}
visited.add(current.id);
path.push(current);
if (current.id === query.stopAtId || current.type === query.stopAtType) break;
pathFromStart.push(current);
if (query.order !== "oldestFirst" && (current.id === query.stopAtId || current.type === query.stopAtType)) {
break;
}
if (!current.parentId) break;
const parent = byId.get(current.parentId);
if (!parent) throw new SessionError("invalid_session", `Entry ${current.parentId} not found`);
current = parent;
}
if (query.order === "oldestFirst") path.reverse();
const entries = path.filter(
const traversal = query.order === "oldestFirst" ? pathFromStart.reverse() : pathFromStart;
const stopIndex =
query.order === "oldestFirst"
? traversal.findIndex((entry) => entry.id === query.stopAtId || entry.type === query.stopAtType)
: -1;
const bounded = stopIndex === -1 ? traversal : traversal.slice(0, stopIndex + 1);
const entries = bounded.filter(
(entry) =>
(query.type === undefined || entry.type === query.type) &&
(query.customType === undefined || (entry.type === "custom" && entry.customType === query.customType)),

View file

@ -21,6 +21,7 @@ async function verifyBranchQueries(session: Session): Promise<{ tail: string; fu
const compaction = await session.appendCompaction("summary", child, 100, undefined, undefined, undefined, [
createAssistantMessage("child"),
]);
const recentCustom = await session.appendCustomEntry("note", { value: 2 });
const tail = await session.appendMessage(createUserMessage("tail"));
await session.moveTo(root);
const sibling = await session.appendMessage(createUserMessage("sibling"));
@ -32,11 +33,12 @@ async function verifyBranchQueries(session: Session): Promise<{ tail: string; fu
custom,
child,
compaction,
recentCustom,
tail,
]);
expect(
(await session.findEntriesOnBranch({ start: tail, stopAtType: "compaction" })).map((entry) => entry.id),
).toEqual([tail, compaction]);
).toEqual([tail, recentCustom, compaction]);
expect(
(await session.findEntriesOnBranch({ start: tail, stopAtType: "compaction", type: "message" })).map(
(entry) => entry.id,
@ -46,13 +48,27 @@ async function verifyBranchQueries(session: Session): Promise<{ tail: string; fu
(await session.findEntriesOnBranch({ start: tail, stopAtId: child, order: "oldestFirst" })).map(
(entry) => entry.id,
),
).toEqual([child, compaction, tail]);
).toEqual([root, custom, child]);
expect((await session.findEntriesOnBranch({ start: tail, stopAtType: "custom" })).map((entry) => entry.id)).toEqual([
tail,
recentCustom,
]);
expect(
(
await session.findEntriesOnBranch({
start: tail,
stopAtType: "custom",
order: "oldestFirst",
})
).map((entry) => entry.id),
).toEqual([root, custom]);
expect(
(await session.findEntriesOnBranch({ start: tail, type: "message", order: "oldestFirst" })).map(
(entry) => entry.id,
),
).toEqual([root, child, tail]);
expect((await session.findEntriesOnBranch({ start: tail, customType: "note" })).map((entry) => entry.id)).toEqual([
recentCustom,
custom,
]);
expect((await session.findEntriesOnBranch({ start: tail, limit: 1 })).map((entry) => entry.id)).toEqual([tail]);
@ -69,7 +85,7 @@ async function verifyBranchQueries(session: Session): Promise<{ tail: string; fu
expect(await session.findEntryOnBranch({ start: tail, type: "compaction" })).toMatchObject({ id: compaction });
await expect(session.findEntriesOnBranch({ start: "missing" })).rejects.toMatchObject({ code: "not_found" });
await expect(session.findEntriesOnBranch({ limit: 0 })).rejects.toThrow("limit must be a positive integer");
return { tail, fullPath: [root, custom, child, compaction, tail] };
return { tail, fullPath: [root, custom, child, compaction, recentCustom, tail] };
}
describe("bounded session branch queries", () => {
@ -97,6 +113,12 @@ describe("bounded session branch queries", () => {
message: createUserMessage("orphan"),
});
expect(
(await reader.findEntriesOnBranch({ start: "orphan", stopAtId: "orphan" })).map((entry) => entry.id),
).toEqual(["orphan"]);
expect(
(await reader.findEntriesOnBranch({ start: "orphan", stopAtType: "message" })).map((entry) => entry.id),
).toEqual(["orphan"]);
await expect(reader.findEntriesOnBranch({ start: "orphan" })).rejects.toMatchObject({
code: "invalid_session",
message: "Entry missing-parent not found",
@ -134,6 +156,172 @@ describe("bounded session branch queries", () => {
).toEqual(expected.fullPath);
});
it("does not decode SQLite branch entries outside query bounds", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const sqlite = createNodeSqliteFactory();
const store = createSqliteSessionStore({
env: new NodeExecutionEnv({ cwd: root }),
sqlite,
databasePath,
});
ownedStores.push(store);
const repo = createSessionRepository({ store });
const session = await repo.create({ id: "bounded-sqlite", cwd: root });
const rootId = await session.appendMessage(createUserMessage("root"));
const middleId = await session.appendMessage(createAssistantMessage("middle"));
const tailId = await session.appendMessage(createUserMessage("tail"));
const db = await sqlite.open(databasePath);
try {
await db
.prepare("UPDATE session_entries SET payload = ? WHERE session_id = ? AND id = ?")
.run("not json", "bounded-sqlite", middleId);
const branch = await db
.prepare("SELECT branch_id FROM branch_entries WHERE session_id = ? AND entry_id = ?")
.get<{ branch_id: string }>("bounded-sqlite", tailId);
if (!branch) throw new Error("Missing branch cache for bounded SQLite query test");
await db
.prepare("DELETE FROM branch_entries WHERE session_id = ? AND branch_id = ? AND entry_id = ?")
.run("bounded-sqlite", branch.branch_id, middleId);
} finally {
await db.close();
}
expect((await session.findEntriesOnBranch({ start: tailId, stopAtId: tailId })).map((entry) => entry.id)).toEqual(
[tailId],
);
const inspection = await sqlite.open(databasePath);
try {
const repaired = await inspection
.prepare(
`SELECT entry_id FROM branch_entries
WHERE session_id = ? AND branch_id = (
SELECT branch_id FROM branch_entries WHERE session_id = ? AND entry_id = ? LIMIT 1
)
ORDER BY entry_seq`,
)
.all<{ entry_id: string }>("bounded-sqlite", "bounded-sqlite", tailId);
expect(repaired.map((row) => row.entry_id)).toEqual([rootId, tailId]);
} finally {
await inspection.close();
}
expect(
(
await session.findEntriesOnBranch({
start: tailId,
stopAtId: rootId,
order: "oldestFirst",
limit: 1,
})
).map((entry) => entry.id),
).toEqual([rootId]);
await expect(session.findEntriesOnBranch({ start: tailId, limit: 2 })).rejects.toMatchObject({
code: "invalid_entry",
message: expect.stringContaining(`failed to decode entry ${middleId}`),
});
});
it("validates SQLite entries before filtering and limiting branch results", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const sqlite = createNodeSqliteFactory();
const store = createSqliteSessionStore({
env: new NodeExecutionEnv({ cwd: root }),
sqlite,
databasePath,
});
ownedStores.push(store);
const repo = createSessionRepository({ store });
const session = await repo.create({ id: "invalid-filtered-sqlite", cwd: root });
await session.appendMessage(createUserMessage("root"));
const customId = await session.appendCustomEntry("note", { value: 1 });
const tailId = await session.appendMessage(createAssistantMessage("tail"));
const db = await sqlite.open(databasePath);
try {
await db
.prepare("UPDATE session_entries SET payload = ? WHERE session_id = ? AND id = ?")
.run("{}", "invalid-filtered-sqlite", customId);
} finally {
await db.close();
}
await expect(session.findEntriesOnBranch({ start: tailId, type: "message", limit: 1 })).rejects.toMatchObject({
code: "invalid_entry",
message: expect.stringContaining(`failed to decode entry ${customId}`),
});
const invalidJsonDb = await sqlite.open(databasePath);
try {
await invalidJsonDb
.prepare("UPDATE session_entries SET payload = ? WHERE session_id = ? AND id = ?")
.run("not json", "invalid-filtered-sqlite", customId);
} finally {
await invalidJsonDb.close();
}
await expect(session.findEntriesOnBranch({ start: tailId, customType: "other" })).rejects.toMatchObject({
code: "invalid_entry",
message: expect.stringContaining(`failed to decode entry ${customId}`),
});
});
it("does not validate SQLite ancestors beyond newest-first stop bounds", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const sqlite = createNodeSqliteFactory();
const store = createSqliteSessionStore({
env: new NodeExecutionEnv({ cwd: root }),
sqlite,
databasePath,
});
ownedStores.push(store);
const repo = createSessionRepository({ store });
const session = await repo.create({ id: "bounded-corrupt-sqlite", cwd: root });
const rootId = await session.appendMessage(createUserMessage("root"));
const childId = await session.appendMessage(createAssistantMessage("child"));
const db = await sqlite.open(databasePath);
try {
await db
.prepare("UPDATE session_entries SET parent_id = ? WHERE session_id = ? AND id = ?")
.run("missing-parent", "bounded-corrupt-sqlite", childId);
} finally {
await db.close();
}
expect(
(await session.findEntriesOnBranch({ start: childId, stopAtId: childId })).map((entry) => entry.id),
).toEqual([childId]);
expect(
(await session.findEntriesOnBranch({ start: childId, stopAtType: "message" })).map((entry) => entry.id),
).toEqual([childId]);
await expect(session.findEntriesOnBranch({ start: childId })).rejects.toMatchObject({
code: "invalid_session",
message: expect.stringContaining("Entry missing-parent not found"),
});
const cycleDb = await sqlite.open(databasePath);
try {
await cycleDb
.prepare("UPDATE session_entries SET parent_id = ? WHERE session_id = ? AND id = ?")
.run(rootId, "bounded-corrupt-sqlite", childId);
await cycleDb
.prepare("UPDATE session_entries SET parent_id = ? WHERE session_id = ? AND id = ?")
.run(childId, "bounded-corrupt-sqlite", rootId);
} finally {
await cycleDb.close();
}
expect(
(await session.findEntriesOnBranch({ start: childId, stopAtId: childId })).map((entry) => entry.id),
).toEqual([childId]);
expect(
(await session.findEntriesOnBranch({ start: childId, stopAtType: "message" })).map((entry) => entry.id),
).toEqual([childId]);
await expect(session.findEntriesOnBranch({ start: childId })).rejects.toMatchObject({
code: "invalid_session",
message: expect.stringContaining(`cycle in parent chain at entry ${childId}`),
});
});
it("provides identical SQLite query semantics", async () => {
const root = createTempDir();
const store = createSqliteSessionStore({

View file

@ -199,6 +199,47 @@ describe("SQLite branch cache", () => {
expect((await fork.getEntries()).map((entry) => entry.id)).toEqual([rootId, childId]);
});
it("repairs a stale branch cache from canonical parent links", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");
const env = new NodeExecutionEnv({ cwd: root });
const sqlite = createNodeSqliteFactory();
const repo = createSessionRepository({ store: createSqliteSessionStore({ env, sqlite, databasePath }) });
const session = await repo.create({ cwd: root, id: "session-1" });
const rootId = await session.appendMessage(createUserMessage("root"));
const staleId = await session.appendMessage(createAssistantMessage("stale"));
const leafId = await session.appendMessage(createUserMessage("leaf"));
const db = await sqlite.open(databasePath);
try {
await db
.prepare("UPDATE session_entries SET parent_id = ? WHERE session_id = ? AND id = ?")
.run(rootId, "session-1", leafId);
} finally {
await db.close();
}
expect(
(await session.findEntriesOnBranch({ start: leafId, order: "oldestFirst" })).map((entry) => entry.id),
).toEqual([rootId, leafId]);
const inspection = await sqlite.open(databasePath);
try {
const rows = await inspection
.prepare(
`SELECT entry_id FROM branch_entries
WHERE session_id = ? AND branch_id = (
SELECT branch_id FROM branch_entries WHERE session_id = ? AND entry_id = ? LIMIT 1
)
ORDER BY entry_seq`,
)
.all<{ entry_id: string }>("session-1", "session-1", leafId);
expect(rows.map((row) => row.entry_id)).toEqual([rootId, leafId]);
expect(rows.map((row) => row.entry_id)).not.toContain(staleId);
} finally {
await inspection.close();
}
});
it("deletes branch entries and tips with the session", async () => {
const root = createTempDir();
const databasePath = join(root, "sessions.sqlite");

View file

@ -9,6 +9,12 @@ export interface CachedBranch {
leafSeq: number;
}
export interface CachedBranchQuery {
stopAtType?: SessionEntryRow["type"];
stopAtId?: string;
order?: "newestFirst" | "oldestFirst";
}
export async function readCachedBranch(
db: SqliteDatabase,
sessionId: string,
@ -23,6 +29,78 @@ export async function readCachedBranch(
return { branchId: membership.branch_id, leafSeq: membership.entry_seq };
}
export async function isCachedBranchValid(
db: SqliteDatabase,
sessionId: string,
branch: CachedBranch,
leafId: string,
startSeq = 0,
): Promise<boolean> {
const result = await db
.prepare(
`WITH path AS (
SELECT
b.entry_id,
b.entry_seq,
e.id AS stored_entry_id,
e.parent_id,
LAG(b.entry_id) OVER (ORDER BY b.entry_seq) AS previous_entry_id
FROM branch_entries AS b
LEFT JOIN session_entries AS e ON e.session_id = b.session_id AND e.id = b.entry_id
WHERE b.session_id = ? AND b.branch_id = ? AND b.entry_seq BETWEEN ? AND ?
)
SELECT
COUNT(*) AS row_count,
COALESCE(SUM(
stored_entry_id IS NULL OR
(? = 0 AND previous_entry_id IS NULL AND parent_id IS NOT NULL) OR
(previous_entry_id IS NOT NULL AND parent_id IS NOT previous_entry_id)
), 0) AS invalid_count,
COALESCE(MAX(entry_seq = ? AND entry_id = ?), 0) AS contains_leaf
FROM path`,
)
.get<{ row_count: number; invalid_count: number; contains_leaf: number }>(
sessionId,
branch.branchId,
startSeq,
branch.leafSeq,
startSeq,
branch.leafSeq,
leafId,
);
return result?.row_count !== 0 && result?.invalid_count === 0 && result.contains_leaf === 1;
}
export async function readNewestCachedStopSeq(
db: SqliteDatabase,
sessionId: string,
branch: CachedBranch,
stopAtType: SessionEntryRow["type"] | undefined,
stopAtId: string | undefined,
): Promise<number | undefined> {
const predicates: string[] = [];
const params: unknown[] = [sessionId, branch.branchId, branch.leafSeq];
if (stopAtType !== undefined) {
predicates.push("e.type = ?");
params.push(stopAtType);
}
if (stopAtId !== undefined) {
predicates.push("b.entry_id = ?");
params.push(stopAtId);
}
if (predicates.length === 0) return undefined;
const row = await db
.prepare(
`SELECT MAX(b.entry_seq) AS entry_seq
FROM branch_entries AS b
JOIN session_entries AS e ON e.session_id = b.session_id AND e.id = b.entry_id
WHERE b.session_id = ? AND b.branch_id = ? AND b.entry_seq <= ?
AND (${predicates.join(" OR ")})`,
)
.get<{ entry_seq: number | null }>(...params);
return row?.entry_seq ?? undefined;
}
export async function readCachedBranchRows(
db: SqliteDatabase,
sessionId: string,
@ -40,6 +118,51 @@ export async function readCachedBranchRows(
.all<SessionEntryRow>(sessionId, branch.branchId, startSeq, branch.leafSeq);
}
export async function queryCachedBranchRows(
db: SqliteDatabase,
sessionId: string,
branch: CachedBranch,
query: CachedBranchQuery,
): Promise<SessionEntryRow[]> {
const oldestFirst = query.order === "oldestFirst";
const boundaryParams: unknown[] = [sessionId, branch.branchId, branch.leafSeq];
const stopPredicates: string[] = [];
if (query.stopAtType !== undefined) {
stopPredicates.push("stop_entry.type = ?");
boundaryParams.push(query.stopAtType);
}
if (query.stopAtId !== undefined) {
stopPredicates.push("stop.entry_id = ?");
boundaryParams.push(query.stopAtId);
}
const boundary = stopPredicates.length
? `WITH boundary AS (
SELECT ${oldestFirst ? "MIN" : "MAX"}(stop.entry_seq) AS entry_seq
FROM branch_entries AS stop
JOIN session_entries AS stop_entry
ON stop_entry.session_id = stop.session_id AND stop_entry.id = stop.entry_id
WHERE stop.session_id = ? AND stop.branch_id = ? AND stop.entry_seq <= ?
AND (${stopPredicates.join(" OR ")})
)`
: "";
const range = stopPredicates.length
? `AND b.entry_seq ${oldestFirst ? "<=" : ">="} COALESCE(
(SELECT entry_seq FROM boundary), ${oldestFirst ? branch.leafSeq : 0}
)`
: "";
const sql = `${boundary}
SELECT e.session_id, e.id, e.entry_seq, e.parent_id, e.type, e.timestamp, e.payload
FROM branch_entries AS b
JOIN session_entries AS e ON e.session_id = b.session_id AND e.id = b.entry_id
WHERE b.session_id = ? AND b.branch_id = ? AND b.entry_seq <= ?
${range}
ORDER BY b.entry_seq ${oldestFirst ? "ASC" : "DESC"}`;
const params = [...(stopPredicates.length === 0 ? [] : boundaryParams), sessionId, branch.branchId, branch.leafSeq];
return db.prepare(sql).all<SessionEntryRow>(...params);
}
export async function readCachedEntryRowsByType(
db: SqliteDatabase,
sessionId: string,

View file

@ -8,10 +8,14 @@ import { SessionError, toError } from "@earendil-works/pi-agent-core";
import type { SqliteDatabase, SqliteSessionMetadata } from "../types.ts";
import {
appendEntryToBranchCache,
type CachedBranch,
isCachedBranchValid,
queryCachedBranchRows,
readCachedBranch,
readCachedBranchRows,
readCachedEntryRowsByType,
readCachedEntrySeq,
readNewestCachedStopSeq,
rebuildCachedBranch,
} from "./branch-cache.ts";
import { decodeEntry, encodeEntry, type SessionEntryRow } from "./session-entries.ts";
@ -81,32 +85,87 @@ export class SqliteSessionConnection implements SessionReader<SqliteSessionMetad
throw new RangeError("Session branch query limit must be a positive integer");
}
if (query.start === null) return [];
const fullPath = await this.readPathToRoot(query.start);
const path: SessionTreeEntry[] = [];
for (let index = fullPath.length - 1; index >= 0; index--) {
const entry = fullPath[index]!;
path.push(entry);
if (entry.id === query.stopAtId || entry.type === query.stopAtType) break;
const startId = query.start;
let cached = await readCachedBranch(this.db, this.metadata.id, startId);
const validationStartSeq =
cached && query.order !== "oldestFirst"
? await readNewestCachedStopSeq(this.db, this.metadata.id, cached, query.stopAtType, query.stopAtId)
: undefined;
if (!cached || !(await isCachedBranchValid(this.db, this.metadata.id, cached, startId, validationStartSeq))) {
if (query.order !== "oldestFirst" && (query.stopAtId !== undefined || query.stopAtType !== undefined)) {
return this.findEntriesOnCanonicalBranch({ ...query, start: startId });
}
cached = await this.repairBranchCacheForQuery(startId, cached?.branchId);
}
if (query.order === "oldestFirst") path.reverse();
const entries = path.filter(
const decoded = decodeEntryRows(await queryCachedBranchRows(this.db, this.metadata.id, cached, query));
const filtered = decoded.filter(
(entry) =>
(query.type === undefined || entry.type === query.type) &&
(query.customType === undefined || (entry.type === "custom" && entry.customType === query.customType)),
);
return query.limit === undefined ? entries : entries.slice(0, query.limit);
const entries = query.limit === undefined ? filtered : filtered.slice(0, query.limit);
for (const entry of entries) this.byId.set(entry.id, entry);
return entries;
}
private async readPathToRoot(leafId: string): Promise<SessionTreeEntry[]> {
const cached = await readCachedBranch(this.db, this.metadata.id, leafId);
if (cached) {
const entries = decodeEntryRows(await readCachedBranchRows(this.db, this.metadata.id, cached, 0));
if (this.isValidCachedPath(entries, leafId, null)) {
for (const entry of entries) this.byId.set(entry.id, entry);
return entries;
private async findEntriesOnCanonicalBranch(
query: SessionBranchQuery & { start: string },
): Promise<SessionTreeEntry[]> {
const rows: SessionEntryRow[] = [];
const visited = new Set<string>();
let currentId: string | null = query.start;
while (currentId !== null) {
if (visited.has(currentId)) throw invalidSession(`cycle in parent chain at entry ${currentId}`);
visited.add(currentId);
const row: SessionEntryRow | undefined = await this.db
.prepare(
"SELECT session_id, id, entry_seq, parent_id, type, timestamp, payload FROM session_entries WHERE session_id = ? AND id = ?",
)
.get<SessionEntryRow>(this.metadata.id, currentId);
if (!row) {
if (currentId === query.start) throw new SessionError("not_found", `Entry ${query.start} not found`);
throw invalidSession(`Entry ${currentId} not found`);
}
rows.push(row);
if (row.id === query.stopAtId || row.type === query.stopAtType) break;
currentId = row.parent_id;
}
return this.repairBranchCache(leafId, cached?.branchId);
const entries = decodeEntryRows(rows).filter(
(entry) =>
(query.type === undefined || entry.type === query.type) &&
(query.customType === undefined || (entry.type === "custom" && entry.customType === query.customType)),
);
const limited = query.limit === undefined ? entries : entries.slice(0, query.limit);
for (const entry of limited) this.byId.set(entry.id, entry);
return limited;
}
private async repairBranchCacheForQuery(leafId: string, branchIdToReplace?: string): Promise<CachedBranch> {
const visited = new Set<string>();
let currentId: string | null = leafId;
while (currentId !== null) {
if (visited.has(currentId)) throw invalidSession(`cycle in parent chain at entry ${currentId}`);
visited.add(currentId);
const row: { parent_id: string | null } | undefined = await this.db
.prepare("SELECT parent_id FROM session_entries WHERE session_id = ? AND id = ?")
.get<{ parent_id: string | null }>(this.metadata.id, currentId);
if (!row) {
if (currentId === leafId) throw new SessionError("not_found", `Entry ${leafId} not found`);
throw invalidSession(`Entry ${currentId} not found`);
}
currentId = row.parent_id;
}
try {
await rebuildCachedBranch(this.db, this.metadata.id, leafId, branchIdToReplace);
} catch (error) {
if (error instanceof SessionError) throw error;
throw new SessionError("storage", `Failed to rebuild SQLite branch cache at entry ${leafId}`, toError(error));
}
const cached = await readCachedBranch(this.db, this.metadata.id, leafId);
if (!cached || !(await isCachedBranchValid(this.db, this.metadata.id, cached, leafId))) {
throw invalidSession(`branch cache repair did not produce a valid path to entry ${leafId}`);
}
return cached;
}
async readPathToRootOrCompaction(leafId: string | null): Promise<SessionTreeEntry[]> {