diff --git a/packages/agent/docs/harness-v2.md b/packages/agent/docs/harness-v2.md index b88e5678b..b27a5a83a 100644 --- a/packages/agent/docs/harness-v2.md +++ b/packages/agent/docs/harness-v2.md @@ -1764,6 +1764,13 @@ Stale branches (no lane resolves through them) are kept. Every restore query is an index seek plus a bounded scan: a lane's open operation via `(lane, type, seq)`, its last run-kind start via `(lane, type, op_kind, seq)`, its records above the operation via the same index, its own entries via the read plan from its leaf. No query touches another lane's traffic. +SQLite implementation follow-ups: + +- Finish search backend work now in progress. +- Add limit and cursor support to search results. +- Route `findEntries` through indexed/search-backed query paths where possible instead of decoding and filtering all session entries. +- Re-audit SQLite query plans after search and `findEntries` changes to see whether further index or query-shape improvements are warranted. + ## 14. Agent-loop building blocks `agent-loop.ts` exposes building blocks that own no durable state and know nothing about sessions, records, or lanes. The harness composes them and inserts durability writes between their phases. diff --git a/packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql b/packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql index 74cbf3490..706b0b04d 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql +++ b/packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql @@ -8,7 +8,6 @@ CREATE TABLE IF NOT EXISTS sessions ( CREATE INDEX IF NOT EXISTS idx_sessions_created_at ON sessions(created_at DESC); CREATE INDEX IF NOT EXISTS idx_sessions_cwd_created_at ON sessions(cwd, created_at DESC); -CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id); CREATE TABLE IF NOT EXISTS entries ( session_id TEXT NOT NULL, @@ -22,7 +21,6 @@ CREATE TABLE IF NOT EXISTS entries ( UNIQUE (session_id, seq) ); -CREATE INDEX IF NOT EXISTS idx_entries_session_seq ON entries(session_id, seq); CREATE INDEX IF NOT EXISTS idx_entries_session_parent ON entries(session_id, parent_id); CREATE INDEX IF NOT EXISTS idx_entries_session_type_seq ON entries(session_id, type, seq); @@ -79,7 +77,6 @@ CREATE TABLE IF NOT EXISTS records ( UNIQUE (session_id, seq) ) WITHOUT ROWID; -CREATE INDEX IF NOT EXISTS idx_records_session_seq ON records(session_id, seq); CREATE INDEX IF NOT EXISTS idx_records_session_lane_seq ON records(session_id, lane, seq); CREATE INDEX IF NOT EXISTS idx_records_session_type_seq ON records(session_id, type, seq); CREATE INDEX IF NOT EXISTS idx_records_session_type_op_kind_seq ON records(session_id, type, op_kind, seq); @@ -95,7 +92,6 @@ CREATE TABLE IF NOT EXISTS lane_moves ( PRIMARY KEY (session_id, seq) ) WITHOUT ROWID; -CREATE INDEX IF NOT EXISTS idx_lane_moves_session_lane_seq ON lane_moves(session_id, lane, seq); CREATE TABLE IF NOT EXISTS facts ( session_id TEXT NOT NULL, @@ -110,8 +106,8 @@ CREATE INDEX IF NOT EXISTS idx_facts_session_kind_key_seq ON facts(session_id, k CREATE TABLE IF NOT EXISTS branch_tips ( session_id TEXT NOT NULL, - tip_id TEXT NOT NULL, branch_id TEXT NOT NULL, + tip_id TEXT NOT NULL, PRIMARY KEY (session_id, tip_id), UNIQUE (session_id, branch_id) ) WITHOUT ROWID; diff --git a/packages/session-backends/sqlite-node/src/sqlite/repo.ts b/packages/session-backends/sqlite-node/src/sqlite/repo.ts index 5bd2f2591..da4b13c01 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/repo.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/repo.ts @@ -532,7 +532,14 @@ class SqliteSessionStorage implements SessionStorage { } async findEntries(query: EntryQuery = {}): Promise { - const rows = readEntryRows(this.db, this.metadata.id, { order: query.order }); + const sqlType = query.type ?? (query.customType === undefined ? undefined : "custom"); + const sqlLimit = query.customType === undefined ? query.limit : undefined; + const rows = readEntryRows(this.db, this.metadata.id, { + cursor: query.cursor, + limit: sqlLimit, + order: query.order, + type: sqlType, + }); const entries = rows.map(decodeEntry).filter((entry) => matchesEntryQuery(entry, query)); return query.limit === undefined ? entries : entries.slice(0, query.limit); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts index 58807897d..76016a71c 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts @@ -1,4 +1,4 @@ -import type { Entry } from "@earendil-works/pi-agent-core"; +import { type Entry, SessionError } from "@earendil-works/pi-agent-core"; import { joinSqlFragments, sql } from "../sql.ts"; import type { SqliteDatabase } from "../types.ts"; @@ -28,6 +28,14 @@ export interface CachedBranchQuery { limit?: number; } +interface BranchPathEntryRow { + id: string; + seq: number; + parent_id: string | null; + type: Entry["type"]; + payload: string; +} + export function readCachedBranch(db: SqliteDatabase, sessionId: string, leafId: string) { const membership = sql`SELECT branch_id, entry_seq FROM branch_entries @@ -56,14 +64,12 @@ export function queryCachedBranchRows( const boundary = stopPredicates.length === 0 ? sql`` - : sql`WITH boundary AS ( - SELECT ${aggregate}(stop.entry_seq) AS entry_seq + : sql`SELECT ${aggregate}(stop.entry_seq) FROM branch_entries AS stop WHERE stop.session_id = ${sessionId} AND stop.branch_id = ${branch.branchId} AND stop.entry_seq <= ${branch.leafSeq} - AND (${joinSqlFragments(stopPredicates, " OR ")}) - )`; + AND (${joinSqlFragments(stopPredicates, " OR ")})`; const predicates = [ sql`b.session_id = ${sessionId}`, @@ -71,17 +77,16 @@ export function queryCachedBranchRows( sql`b.entry_seq <= ${branch.leafSeq}`, ]; if (stopPredicates.length > 0) { - predicates.push(sql`b.entry_seq ${boundaryComparison} COALESCE( - (SELECT entry_seq FROM boundary), ${oldestFirst ? branch.leafSeq : 0} - )`); + predicates.push( + sql`b.entry_seq ${boundaryComparison} COALESCE((${boundary}), ${oldestFirst ? branch.leafSeq : 0})`, + ); } if (query.cursor !== undefined) predicates.push(sql`b.entry_seq ${cursorComparison} ${query.cursor.afterSeq}`); if (query.type !== undefined) predicates.push(sql`b.entry_type = ${query.type}`); if (query.customType !== undefined) predicates.push(sql`b.custom_type = ${query.customType}`); const limit = query.limit === undefined ? sql`` : sql` LIMIT ${query.limit}`; - return sql`${boundary} - SELECT e.session_id, e.id, e.seq AS entry_seq, e.parent_id, e.type, e.timestamp, e.payload + return sql`SELECT e.session_id, e.id, e.seq AS entry_seq, e.parent_id, e.type, e.timestamp, e.payload FROM branch_entries AS b JOIN entries AS e ON e.session_id = b.session_id AND e.id = b.entry_id WHERE ${joinSqlFragments(predicates, " AND ")} @@ -106,21 +111,44 @@ export function insertBranchEntry( VALUES (${sessionId}, ${branchId}, ${entryId}, ${entrySeq}, ${entryType}, ${customType})`.run(db); } +function customTypeFromPayload(row: BranchPathEntryRow): string | null { + if (row.type !== "custom") return null; + try { + const payload = JSON.parse(row.payload) as unknown; + if (typeof payload !== "object" || payload === null || Array.isArray(payload)) { + throw new Error("Payload is not an object"); + } + const customType = (payload as { customType?: unknown }).customType; + if (typeof customType !== "string") throw new Error("Invalid custom payload"); + return customType; + } catch (error) { + throw new SessionError( + "invalid_entry", + `Invalid SQLite session entry ${row.id}: failed to decode entry ${row.id}`, + error instanceof Error ? error : undefined, + ); + } +} + export function insertBranchEntriesForPath(db: SqliteDatabase, sessionId: string, branchId: string, leafId: string) { - sql`WITH RECURSIVE path(id, entry_seq, parent_id, type, custom_type) AS ( - SELECT id, seq, parent_id, type, - CASE WHEN type = 'custom' THEN json_extract(payload, '$.customType') ELSE NULL END + const path: BranchPathEntryRow[] = []; + const seen = new Set(); + let entryId: string | null = leafId; + + while (entryId !== null) { + if (seen.has(entryId)) throw new SessionError("invalid_entry", `Entry parent cycle at ${entryId}`); + seen.add(entryId); + const row: BranchPathEntryRow | undefined = sql`SELECT id, seq, parent_id, type, payload FROM entries - WHERE session_id = ${sessionId} AND id = ${leafId} - UNION ALL - SELECT parent.id, parent.seq, parent.parent_id, parent.type, - CASE WHEN parent.type = 'custom' THEN json_extract(parent.payload, '$.customType') ELSE NULL END - FROM entries AS parent - JOIN path AS child ON child.parent_id = parent.id - WHERE parent.session_id = ${sessionId} - ) - INSERT INTO branch_entries (session_id, branch_id, entry_id, entry_seq, entry_type, custom_type) - SELECT ${sessionId}, ${branchId}, id, entry_seq, type, custom_type FROM path`.run(db); + WHERE session_id = ${sessionId} AND id = ${entryId}`.get(db); + if (!row) throw new SessionError("invalid_entry", `Entry ${entryId} not found`); + path.push(row); + entryId = row.parent_id; + } + + for (const row of path.reverse()) { + insertBranchEntry(db, sessionId, branchId, row.id, row.seq, row.type, customTypeFromPayload(row)); + } } export function readBranchContainingEntry(db: SqliteDatabase, sessionId: string, entryId: string) { diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts index c2a8b625e..e76ef3cf6 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts @@ -42,14 +42,28 @@ export function readEntryRow(db: SqliteDatabase, sessionId: string, entryId: str export function readEntryRows( db: SqliteDatabase, sessionId: string, - options: { afterSeq?: number; order?: EntryOrder; limit?: number } = {}, + options: { + afterSeq?: number; + cursor?: { afterSeq: number }; + type?: Entry["type"]; + order?: EntryOrder; + limit?: number; + } = {}, ) { + const oldestFirst = options.order === "oldestFirst"; const after = options.afterSeq === undefined ? sql`` : sql` AND seq > ${options.afterSeq}`; - const direction = options.order === "oldestFirst" ? sql`ASC` : sql`DESC`; + const cursor = + options.cursor === undefined + ? sql`` + : oldestFirst + ? sql` AND seq > ${options.cursor.afterSeq}` + : sql` AND seq < ${options.cursor.afterSeq}`; + const type = options.type === undefined ? sql`` : sql` AND type = ${options.type}`; + const direction = oldestFirst ? sql`ASC` : sql`DESC`; const limit = options.limit === undefined ? sql`` : sql` LIMIT ${options.limit}`; return sql`SELECT session_id, seq, id, parent_id, type, timestamp, payload FROM entries - WHERE session_id = ${sessionId}${after} + WHERE session_id = ${sessionId}${after}${cursor}${type} ORDER BY seq ${direction}${limit}`.all(db); } diff --git a/packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts b/packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts index a9654aede..19fa76e75 100644 --- a/packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts +++ b/packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts @@ -31,17 +31,19 @@ export function readLatestFact(db: SqliteDatabase, sessionId: string, kind: stri } export function readLatestLabelFacts(db: SqliteDatabase, sessionId: string) { - return sql`WITH latest AS ( - SELECT key, MAX(seq) AS seq - FROM facts INDEXED BY idx_facts_session_kind_key_seq - WHERE session_id = ${sessionId} AND kind = 'label' - GROUP BY key - ) - SELECT latest.key, f.value - FROM latest - JOIN facts AS f ON f.session_id = ${sessionId} AND f.seq = latest.seq - WHERE f.value IS NOT NULL - ORDER BY latest.key`.all<{ key: string; value: string }>(db); + return sql`SELECT f.key, f.value + FROM facts AS f INDEXED BY idx_facts_session_kind_key_seq + WHERE f.session_id = ${sessionId} + AND f.kind = 'label' + AND f.value IS NOT NULL + AND f.seq = ( + SELECT MAX(candidate.seq) + FROM facts AS candidate INDEXED BY idx_facts_session_kind_key_seq + WHERE candidate.session_id = f.session_id + AND candidate.kind = f.kind + AND candidate.key IS f.key + ) + ORDER BY f.key`.all<{ key: string; value: string }>(db); } export function readFactRows( diff --git a/packages/session-backends/sqlite-node/test/migrations.test.ts b/packages/session-backends/sqlite-node/test/migrations.test.ts index ae8f4f352..fb1631100 100644 --- a/packages/session-backends/sqlite-node/test/migrations.test.ts +++ b/packages/session-backends/sqlite-node/test/migrations.test.ts @@ -36,8 +36,11 @@ describe("SQLite migrations", () => { expect(sessionColumns.map((column) => column.name)).not.toContain("leaf_id"); const sessionIndexes = db.prepare("PRAGMA index_list(sessions)").all<{ name: string }>(); expect(sessionIndexes.map((index) => index.name)).toContain("idx_sessions_cwd_created_at"); + expect(sessionIndexes.map((index) => index.name)).not.toContain("idx_sessions_parent"); const laneColumns = db.prepare("PRAGMA table_info(lanes)").all<{ name: string }>(); expect(laneColumns.map((column) => column.name)).toContain("open_operation_id"); + const entryIndexes = db.prepare("PRAGMA index_list(entries)").all<{ name: string }>(); + expect(entryIndexes.map((index) => index.name)).not.toContain("idx_entries_session_seq"); const branchEntryIndexes = db.prepare("PRAGMA index_list(branch_entries)").all<{ name: string }>(); expect(branchEntryIndexes.map((index) => index.name)).toContain("idx_branch_entries_session_entry"); const recordIndexes = db.prepare("PRAGMA index_list(records)").all<{ name: string }>(); @@ -48,6 +51,9 @@ describe("SQLite migrations", () => { "idx_records_session_type_op_kind_seq", ]), ); + expect(recordIndexes.map((index) => index.name)).not.toContain("idx_records_session_seq"); + const laneMoveIndexes = db.prepare("PRAGMA index_list(lane_moves)").all<{ name: string }>(); + expect(laneMoveIndexes.map((index) => index.name)).not.toContain("idx_lane_moves_session_lane_seq"); } finally { db.close(); }