From 081d7fbfca102aec2ee9f052c7804637cebebc4c Mon Sep 17 00:00:00 2001 From: Christian Klotz Date: Sat, 1 Aug 2026 20:54:36 +0300 Subject: [PATCH] fix(agent): harden bounded branch queries --- .../harness/session/array-session-reader.ts | 17 +- .../agent/test/harness/branch-query.test.ts | 194 +++++++++++++++++- .../test/harness/sqlite-branch-cache.test.ts | 41 ++++ .../src/sqlite/storage/branch-cache.ts | 123 +++++++++++ .../sqlite-node/src/sqlite/storage/index.ts | 93 +++++++-- 5 files changed, 443 insertions(+), 25 deletions(-) diff --git a/packages/agent/src/harness/session/array-session-reader.ts b/packages/agent/src/harness/session/array-session-reader.ts index c7c532aa2..696a7b14f 100644 --- a/packages/agent/src/harness/session/array-session-reader.ts +++ b/packages/agent/src/harness/session/array-session-reader.ts @@ -51,7 +51,7 @@ export function createArraySessionReader( 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(); 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( 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)), diff --git a/packages/agent/test/harness/branch-query.test.ts b/packages/agent/test/harness/branch-query.test.ts index 816374326..8810a27fa 100644 --- a/packages/agent/test/harness/branch-query.test.ts +++ b/packages/agent/test/harness/branch-query.test.ts @@ -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({ diff --git a/packages/agent/test/harness/sqlite-branch-cache.test.ts b/packages/agent/test/harness/sqlite-branch-cache.test.ts index 200b7a635..10b8989e2 100644 --- a/packages/agent/test/harness/sqlite-branch-cache.test.ts +++ b/packages/agent/test/harness/sqlite-branch-cache.test.ts @@ -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"); diff --git a/packages/storage/sqlite-node/src/sqlite/storage/branch-cache.ts b/packages/storage/sqlite-node/src/sqlite/storage/branch-cache.ts index 9c60f226f..9d33efc8d 100644 --- a/packages/storage/sqlite-node/src/sqlite/storage/branch-cache.ts +++ b/packages/storage/sqlite-node/src/sqlite/storage/branch-cache.ts @@ -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 { + 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 { + 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(sessionId, branch.branchId, startSeq, branch.leafSeq); } +export async function queryCachedBranchRows( + db: SqliteDatabase, + sessionId: string, + branch: CachedBranch, + query: CachedBranchQuery, +): Promise { + 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(...params); +} + export async function readCachedEntryRowsByType( db: SqliteDatabase, sessionId: string, diff --git a/packages/storage/sqlite-node/src/sqlite/storage/index.ts b/packages/storage/sqlite-node/src/sqlite/storage/index.ts index 4bfdf0e7b..72fbf162f 100644 --- a/packages/storage/sqlite-node/src/sqlite/storage/index.ts +++ b/packages/storage/sqlite-node/src/sqlite/storage/index.ts @@ -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= 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 { - 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 { + const rows: SessionEntryRow[] = []; + const visited = new Set(); + 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(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 { + const visited = new Set(); + 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 {