sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix: no ctes in sqlite, delete indexes

Browse source
This commit is contained in:
Cristina Poncela Cubeiro 2026-08-07 16:13:16 +02:00
parent d1a305613e
commit e7fb8eb2ac
7 changed files with 103 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

6
packages/session-backends/sqlite-node/src/sqlite/migrations/001_initial.sql
View file

@ -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;

9
packages/session-backends/sqlite-node/src/sqlite/repo.ts
View file

@ -532,7 +532,14 @@ class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadata> {
}
async findEntries(query: EntryQuery = {}): Promise<Entry[]> {
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);
}

74
packages/session-backends/sqlite-node/src/sqlite/storage/branch-entries.ts
View file

@ -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<string>();
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<BranchPathEntryRow>(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) {

20
packages/session-backends/sqlite-node/src/sqlite/storage/entries.ts
View file

@ -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<EntryRow>(db);
}

24
packages/session-backends/sqlite-node/src/sqlite/storage/facts.ts
View file

@ -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(