diff --git a/packages/mnemopi/CHANGELOG.md b/packages/mnemopi/CHANGELOG.md index 513734245..133adfc96 100644 --- a/packages/mnemopi/CHANGELOG.md +++ b/packages/mnemopi/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed one-shot prepared statements never being released, which kept the SQLite connection alive after `close()`. Every `db.prepare(...)` in `beam/store.ts`, `annotations.ts`, `memory.ts`, `beam/consolidate.ts`, `beam/helpers.ts`, and the e6 migration now either goes through `db.run()` or is bound with `using`, so the statement is finalized when it leaves scope. On Windows this left the memory DB and its `-wal`/`-shm` sidecars locked, so the file could not be deleted, moved, or rotated after mnemopi closed it; on POSIX the handle leaked silently because open files can be unlinked. + ## [17.0.8] - 2026-07-22 ### Changed diff --git a/packages/mnemopi/src/core/annotations.ts b/packages/mnemopi/src/core/annotations.ts index b582c36ba..c6718249b 100644 --- a/packages/mnemopi/src/core/annotations.ts +++ b/packages/mnemopi/src/core/annotations.ts @@ -110,7 +110,7 @@ interface StatementRunResult { readonly lastInsertRowid: number | bigint; } -interface WritableStatement { +interface WritableStatement extends Disposable { run(...params: SqlValue[]): StatementRunResult; } @@ -249,11 +249,10 @@ export class AnnotationStore { } add(memoryId: string, kind: string, value: string, source = "", confidence = 1.0): number { - const result = this.db - .prepare( - "INSERT OR IGNORE INTO annotations (memory_id, kind, value, source, confidence) VALUES (?, ?, ?, ?, ?)", - ) - .run(memoryId, kind, value, source, confidence); + const result = this.db.run( + "INSERT OR IGNORE INTO annotations (memory_id, kind, value, source, confidence) VALUES (?, ?, ?, ?, ?)", + [memoryId, kind, value, source, confidence], + ); return Number(result.lastInsertRowid); } @@ -267,7 +266,7 @@ export class AnnotationStore { if (!values || values.length === 0) return 0; const rows = values.filter(value => value.length > 0 && value.trim().length > 0); if (rows.length === 0) return 0; - const insert = this.db.prepare( + using insert = this.db.prepare( "INSERT OR IGNORE INTO annotations (memory_id, kind, value, source, confidence) VALUES (?, ?, ?, ?, ?)", ); transaction(this.db, () => { @@ -280,10 +279,8 @@ export class AnnotationStore { kind === null || kind === undefined ? "SELECT * FROM annotations WHERE memory_id = ? ORDER BY created_at ASC, id ASC" : "SELECT * FROM annotations WHERE memory_id = ? AND kind = ? ORDER BY created_at ASC, id ASC"; - const rows = - kind === null || kind === undefined - ? this.db.prepare(sql).all(memoryId) - : this.db.prepare(sql).all(memoryId, kind); + using statement = this.db.prepare(sql); + const rows = kind === null || kind === undefined ? statement.all(memoryId) : statement.all(memoryId, kind); return (rows as AnnotationRow[]).map(normalizeRow); } queryByKind( @@ -307,23 +304,24 @@ export class AnnotationStore { conditions.push("memory_id = ?"); params.push(memoryId); } - const rows = this.db - .prepare(`SELECT * FROM annotations WHERE ${conditions.join(" AND ")} ORDER BY created_at ASC, id ASC`) - .all(...params) as AnnotationRow[]; + using statement = this.db.prepare( + `SELECT * FROM annotations WHERE ${conditions.join(" AND ")} ORDER BY created_at ASC, id ASC`, + ); + const rows = statement.all(...params) as AnnotationRow[]; const normalized = rows.map(normalizeRow); const filterNoise = options.filter_noise ?? options.filterNoise ?? true; return filterNoise && kind === "mentions" ? filterCleanMentions(normalized) : normalized; } getDistinctValues(kind: string): string[] { - const rows = this.db - .prepare("SELECT DISTINCT value FROM annotations WHERE kind = ? ORDER BY value") - .all(kind) as { value: string }[]; + using statement = this.db.prepare("SELECT DISTINCT value FROM annotations WHERE kind = ? ORDER BY value"); + const rows = statement.all(kind) as { value: string }[]; return rows.map(row => row.value); } exportAll(): AnnotationRow[] { - const rows = this.db - .prepare("SELECT id, memory_id, kind, value, source, confidence, created_at FROM annotations ORDER BY id") - .all() as AnnotationRow[]; + using statement = this.db.prepare( + "SELECT id, memory_id, kind, value, source, confidence, created_at FROM annotations ORDER BY id", + ); + const rows = statement.all() as AnnotationRow[]; return rows.map(normalizeRow); } importAll(annotations: readonly AnnotationInput[], force = false): AnnotationImportStats { @@ -346,19 +344,20 @@ export class AnnotationStore { } transaction(this.db, () => { - const existingRows = this.db - .prepare("SELECT id, memory_id, kind, value, source, confidence, created_at FROM annotations") - .all() as AnnotationRow[]; + using existingStatement = this.db.prepare( + "SELECT id, memory_id, kind, value, source, confidence, created_at FROM annotations", + ); + const existingRows = existingStatement.all() as AnnotationRow[]; const existing = new Map(); for (const row of existingRows) existing.set(Number(row.id), normalizeRow(row)); - const insertWithId = this.db.prepare( + using insertWithId = this.db.prepare( "INSERT INTO annotations (id, memory_id, kind, value, source, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", ) as WritableStatement; - const insertWithoutId = this.db.prepare( + using insertWithoutId = this.db.prepare( "INSERT INTO annotations (memory_id, kind, value, source, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?)", ) as WritableStatement; - const deleteById = this.db.prepare("DELETE FROM annotations WHERE id = ?"); + using deleteById = this.db.prepare("DELETE FROM annotations WHERE id = ?"); for (const item of annotations) { const id = rowId(item.id); diff --git a/packages/mnemopi/src/core/beam/consolidate.ts b/packages/mnemopi/src/core/beam/consolidate.ts index c1623cb64..1a2a3da89 100644 --- a/packages/mnemopi/src/core/beam/consolidate.ts +++ b/packages/mnemopi/src/core/beam/consolidate.ts @@ -828,8 +828,8 @@ function extractKeySignal(content: string, maxChars: number): string { } function invalidateEpisodicVectors(beam: BeamMemoryState, memoryId: string): void { - beam.db.prepare("DELETE FROM memory_embeddings WHERE memory_id = ?").run(memoryId); - beam.db.prepare("UPDATE episodic_memory SET binary_vector = NULL WHERE id = ?").run(memoryId); + beam.db.run("DELETE FROM memory_embeddings WHERE memory_id = ?", [memoryId]); + beam.db.run("UPDATE episodic_memory SET binary_vector = NULL WHERE id = ?", [memoryId]); } export function degradeEpisodic(beam: BeamMemoryState, dryRun = false): Record { diff --git a/packages/mnemopi/src/core/beam/helpers.ts b/packages/mnemopi/src/core/beam/helpers.ts index 8bf0d51ac..a1bbc2fa2 100644 --- a/packages/mnemopi/src/core/beam/helpers.ts +++ b/packages/mnemopi/src/core/beam/helpers.ts @@ -782,7 +782,7 @@ async function runEmbedding(beam: BeamMemoryState, items: readonly EmbedItem[]): const matrix = await embed(items.map(item => item.content)); if (matrix === null) return; const model = currentEmbeddingModel(); - const insertEmbedding = beam.db.prepare( + using insertEmbedding = beam.db.prepare( "INSERT OR REPLACE INTO memory_embeddings(memory_id, embedding_json, model) VALUES (?, ?, ?)", ); const insertMany = beam.db.transaction((rows: readonly EmbedItem[]) => { diff --git a/packages/mnemopi/src/core/beam/store.ts b/packages/mnemopi/src/core/beam/store.ts index d18a64ce5..aba798881 100644 --- a/packages/mnemopi/src/core/beam/store.ts +++ b/packages/mnemopi/src/core/beam/store.ts @@ -157,18 +157,16 @@ function invalidateCaches(beam: BeamMemoryState): void { } function findDuplicate(beam: BeamMemoryState, content: string): string | null { - const row = beam.db - .prepare("SELECT id FROM working_memory WHERE content = ? AND session_id = ? LIMIT 1") - .get(content, beam.sessionId) as { id: string } | null; + using statement = beam.db.prepare("SELECT id FROM working_memory WHERE content = ? AND session_id = ? LIMIT 1"); + const row = statement.get(content, beam.sessionId) as { id: string } | null; return row?.id ?? null; } function tableExists(db: BeamMemoryState["db"], table: string): boolean { - return ( - db - .prepare("SELECT 1 FROM sqlite_master WHERE type IN ('table','virtual table') AND name = ? LIMIT 1") - .get(table) !== null + using statement = db.prepare( + "SELECT 1 FROM sqlite_master WHERE type IN ('table','virtual table') AND name = ? LIMIT 1", ); + return statement.get(table) !== null; } /** Tables whose rows point back to a `working_memory` id via `source_memory_id`. */ @@ -197,29 +195,30 @@ function purgeWorkingMemoryArtifacts(db: BeamMemoryState["db"], ids: readonly st const graphRefs = new Set(ids); for (const id of ids) graphRefs.add(`gist_${id}`); if (tableExists(db, "facts")) { - const factRows = db.prepare(`SELECT fact_id FROM facts WHERE source_msg_id IN (${placeholders})`).all(...ids) as { + using factStatement = db.prepare(`SELECT fact_id FROM facts WHERE source_msg_id IN (${placeholders})`); + const factRows = factStatement.all(...ids) as { fact_id: string; }[]; for (const row of factRows) graphRefs.add(row.fact_id); - db.prepare(`DELETE FROM facts WHERE source_msg_id IN (${placeholders})`).run(...ids); + db.run(`DELETE FROM facts WHERE source_msg_id IN (${placeholders})`, [...ids]); } - db.prepare(`DELETE FROM annotations WHERE memory_id IN (${placeholders})`).run(...ids); - db.prepare(`DELETE FROM memory_embeddings WHERE memory_id IN (${placeholders})`).run(...ids); + db.run(`DELETE FROM annotations WHERE memory_id IN (${placeholders})`, [...ids]); + db.run(`DELETE FROM memory_embeddings WHERE memory_id IN (${placeholders})`, [...ids]); for (const table of MEMORIA_SOURCE_TABLES) { - db.prepare(`DELETE FROM ${table} WHERE source_memory_id IN (${placeholders})`).run(...ids); + db.run(`DELETE FROM ${table} WHERE source_memory_id IN (${placeholders})`, [...ids]); } if (tableExists(db, "gists")) { - db.prepare(`DELETE FROM gists WHERE memory_id IN (${placeholders})`).run(...ids); + db.run(`DELETE FROM gists WHERE memory_id IN (${placeholders})`, [...ids]); } if (tableExists(db, "graph_edges")) { const refs = [...graphRefs]; const refPlaceholders = refs.map(() => "?").join(", "); - db.prepare(`DELETE FROM graph_edges WHERE source IN (${refPlaceholders}) OR target IN (${refPlaceholders})`).run( + db.run(`DELETE FROM graph_edges WHERE source IN (${refPlaceholders}) OR target IN (${refPlaceholders})`, [ ...refs, ...refs, - ); + ]); } } @@ -237,30 +236,30 @@ function trimWorkingMemory(beam: BeamMemoryState): void { const ttlHours = beam.config.workingMemoryTtlHours; const cutoff = toUtcIso(new Date(Date.now() - ttlHours * 3_600_000)); transaction(beam.db, () => { - const ids = ( - beam.db - .prepare(` + using selectStatement = beam.db.prepare(` + SELECT id FROM working_memory + WHERE session_id = ? + AND consolidated_at IS NULL + AND trust_tier IS NOT 'IMPORTED' + AND ( + timestamp < ? OR + id NOT IN ( SELECT id FROM working_memory - WHERE session_id = ? - AND consolidated_at IS NULL - AND trust_tier IS NOT 'IMPORTED' - AND ( - timestamp < ? OR - id NOT IN ( - SELECT id FROM working_memory - WHERE session_id = ? AND consolidated_at IS NULL AND trust_tier IS NOT 'IMPORTED' - ORDER BY timestamp DESC - LIMIT ? - ) - ) - `) - .all(beam.sessionId, cutoff, beam.sessionId, limit) as { id: string }[] - ).map(row => row.id); + WHERE session_id = ? AND consolidated_at IS NULL AND trust_tier IS NOT 'IMPORTED' + ORDER BY timestamp DESC + LIMIT ? + ) + ) + `); + const ids = (selectStatement.all(beam.sessionId, cutoff, beam.sessionId, limit) as { id: string }[]).map( + row => row.id, + ); if (ids.length === 0) return; const placeholders = ids.map(() => "?").join(", "); - beam.db - .prepare(`DELETE FROM working_memory WHERE id IN (${placeholders}) AND session_id = ?`) - .run(...ids, beam.sessionId); + beam.db.run(`DELETE FROM working_memory WHERE id IN (${placeholders}) AND session_id = ?`, [ + ...ids, + beam.sessionId, + ]); purgeWorkingMemoryArtifacts(beam.db, ids); }); } @@ -396,11 +395,11 @@ export function reconcileEmbeddingModel(beam: BeamMemoryState): void { .all() as EmbedItem[]; transaction(beam.db, () => { - beam.db.prepare("DELETE FROM memory_embeddings").run(); - beam.db.prepare("UPDATE episodic_memory SET binary_vector = NULL").run(); + beam.db.run("DELETE FROM memory_embeddings"); + beam.db.run("UPDATE episodic_memory SET binary_vector = NULL"); if (vecAvailable(beam.db)) { try { - beam.db.prepare("DELETE FROM vec_episodes").run(); + beam.db.run("DELETE FROM vec_episodes"); } catch { // sqlite-vec cleanup is best-effort; rebuild correctness takes precedence. } @@ -451,8 +450,8 @@ export function remember(beam: BeamMemoryState, content: string, options: StoreR const existingId = findDuplicate(beam, content); if (existingId !== null) { - beam.db - .prepare(` + beam.db.run( + ` UPDATE working_memory SET importance = MAX(importance, ?), timestamp = ?, source = ?, valid_until = COALESCE(?, valid_until), @@ -466,8 +465,8 @@ export function remember(beam: BeamMemoryState, content: string, options: StoreR embed_text = COALESCE(?, embed_text), consolidated_at = NULL WHERE id = ? AND session_id = ? - `) - .run( + `, + [ importance, timestamp, source, @@ -483,7 +482,8 @@ export function remember(beam: BeamMemoryState, content: string, options: StoreR storedEmbeddingText(content, embedText), existingId, beam.sessionId, - ); + ], + ); emitEvent(beam, "MEMORY_UPDATED", { memoryId: existingId, content, @@ -497,14 +497,14 @@ export function remember(beam: BeamMemoryState, content: string, options: StoreR } const memoryId = options.memoryId ?? options.memory_id ?? generateId(content, new Date(timestamp)); - beam.db - .prepare(` + beam.db.run( + ` INSERT INTO working_memory (id, content, embed_text, source, timestamp, session_id, importance, metadata_json, valid_until, scope, author_id, author_type, channel_id, veracity, memory_type, trust_tier) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `) - .run( + `, + [ memoryId, content, storedEmbeddingText(content, embedText), @@ -521,7 +521,8 @@ export function remember(beam: BeamMemoryState, content: string, options: StoreR veracity, memoryType, trustTier, - ); + ], + ); addTemporalAnnotations(beam, memoryId, timestamp, source); // `extractText` lets a caller decouple "what gets stored" from "what facts are // mined". coding-agent retains full multi-author transcripts but wants @@ -560,7 +561,7 @@ export function rememberBatch( const trustTier = normalizeTrustTier(options.trustTier ?? "IMPORTED", "imported"); transaction(beam.db, () => { - const statement = beam.db.prepare(` + using statement = beam.db.prepare(` INSERT INTO working_memory (id, content, embed_text, source, timestamp, session_id, importance, metadata_json, author_id, author_type, channel_id, memory_type, veracity, trust_tier, scope) @@ -625,41 +626,40 @@ export function rememberBatch( export function getContext(beam: BeamMemoryState, limit = 10): Row[] { const now = toUtcIso(); - return ( - beam.db - .prepare(` - SELECT id, content, source, timestamp, importance, scope - FROM working_memory - WHERE (session_id = ? OR scope = 'global') - AND (valid_until IS NULL OR valid_until > ?) - AND superseded_by IS NULL - ORDER BY - CASE WHEN scope = 'global' THEN 0 ELSE 1 END, - importance DESC, - timestamp DESC - LIMIT ? - `) - .all(beam.sessionId, now, limit) as Row[] - ).map(rowToDict); + using statement = beam.db.prepare(` + SELECT id, content, source, timestamp, importance, scope + FROM working_memory + WHERE (session_id = ? OR scope = 'global') + AND (valid_until IS NULL OR valid_until > ?) + AND superseded_by IS NULL + ORDER BY + CASE WHEN scope = 'global' THEN 0 ELSE 1 END, + importance DESC, + timestamp DESC + LIMIT ? + `); + return (statement.all(beam.sessionId, now, limit) as Row[]).map(rowToDict); } export function invalidate(beam: BeamMemoryState, memoryId: string, replacementId: string | null = null): boolean { const now = toUtcIso(); - const working = beam.db - .prepare(` + const working = beam.db.run( + ` UPDATE working_memory SET valid_until = ?, superseded_by = ? WHERE id = ? AND (session_id = ? OR scope = 'global') - `) - .run(now, replacementId, memoryId, beam.sessionId); + `, + [now, replacementId, memoryId, beam.sessionId], + ); if (working.changes > 0) return true; - const episodic = beam.db - .prepare(` + const episodic = beam.db.run( + ` UPDATE episodic_memory SET valid_until = ?, superseded_by = ? WHERE id = ? AND (session_id = ? OR scope = 'global') - `) - .run(now, replacementId, memoryId, beam.sessionId); + `, + [now, replacementId, memoryId, beam.sessionId], + ); return episodic.changes > 0; } @@ -684,12 +684,14 @@ export function getWorkingStats( params.push(channelId); } const where = clauses.length === 0 ? "" : ` WHERE ${clauses.join(" AND ")}`; - const total = beam.db.prepare(`SELECT COUNT(*) AS total FROM working_memory${where}`).get(...params) as { + using totalStatement = beam.db.prepare(`SELECT COUNT(*) AS total FROM working_memory${where}`); + const total = totalStatement.get(...params) as { total: number; }; - const last = beam.db - .prepare(`SELECT timestamp FROM working_memory${where} ORDER BY timestamp DESC LIMIT 1`) - .get(...params) as { timestamp: string | null } | null; + using lastStatement = beam.db.prepare( + `SELECT timestamp FROM working_memory${where} ORDER BY timestamp DESC LIMIT 1`, + ); + const last = lastStatement.get(...params) as { timestamp: string | null } | null; return { total: total.total, count: total.total, last: last?.timestamp ?? null }; } @@ -715,9 +717,10 @@ export function updateWorking( } if (assignments.length === 0) return false; params.push(memoryId, beam.sessionId); - const result = beam.db - .prepare(`UPDATE working_memory SET ${assignments.join(", ")} WHERE id = ? AND session_id = ?`) - .run(...params); + const result = beam.db.run( + `UPDATE working_memory SET ${assignments.join(", ")} WHERE id = ? AND session_id = ?`, + params, + ); if (result.changes > 0) { invalidateCaches(beam); if (content !== null) scheduleEmbedding(beam, [{ memoryId, content }]); @@ -726,24 +729,22 @@ export function updateWorking( } export function get(beam: BeamMemoryState, memoryId: string): Row | null { - const working = beam.db - .prepare(` - SELECT id, content, source, timestamp, session_id, - importance, metadata_json, veracity, created_at - FROM working_memory - WHERE id = ? - `) - .get(memoryId) as Row | null | undefined; + using workingStatement = beam.db.prepare(` + SELECT id, content, source, timestamp, session_id, + importance, metadata_json, veracity, created_at + FROM working_memory + WHERE id = ? + `); + const working = workingStatement.get(memoryId) as Row | null | undefined; if (working != null) return { ...working, metadata: working.metadata_json, memory_store: "working" }; - const episodic = beam.db - .prepare(` - SELECT id, content, source, timestamp, session_id, - importance, metadata_json, veracity, created_at - FROM episodic_memory - WHERE id = ? AND (session_id = ? OR scope = 'global') - `) - .get(memoryId, beam.sessionId) as Row | null | undefined; + using episodicStatement = beam.db.prepare(` + SELECT id, content, source, timestamp, session_id, + importance, metadata_json, veracity, created_at + FROM episodic_memory + WHERE id = ? AND (session_id = ? OR scope = 'global') + `); + const episodic = episodicStatement.get(memoryId, beam.sessionId) as Row | null | undefined; if (episodic != null) return { ...episodic, metadata: episodic.metadata_json, memory_store: "episodic" }; return getFact(beam, memoryId); @@ -762,7 +763,8 @@ export function get(beam: BeamMemoryState, memoryId: string): Row | null { * path mutates `facts`. */ function getFact(beam: BeamMemoryState, memoryId: string): Row | null { - const fact = beam.db.prepare("SELECT * FROM facts WHERE fact_id = ?").get(memoryId) as Row | null | undefined; + using statement = beam.db.prepare("SELECT * FROM facts WHERE fact_id = ?"); + const fact = statement.get(memoryId) as Row | null | undefined; if (fact == null) return null; if (fact.session_id !== beam.sessionId && fact.scope !== "global") return null; const subject = typeof fact.subject === "string" ? fact.subject : ""; @@ -789,9 +791,10 @@ function getFact(beam: BeamMemoryState, memoryId: string): Row | null { export function forgetWorking(beam: BeamMemoryState, memoryId: string): boolean { let deleted = 0; transaction(beam.db, () => { - const result = beam.db - .prepare("DELETE FROM working_memory WHERE id = ? AND session_id = ?") - .run(memoryId, beam.sessionId); + const result = beam.db.run("DELETE FROM working_memory WHERE id = ? AND session_id = ?", [ + memoryId, + beam.sessionId, + ]); deleted = result.changes; if (deleted > 0) { purgeWorkingMemoryArtifacts(beam.db, [memoryId]); @@ -804,36 +807,65 @@ export function forgetWorking(beam: BeamMemoryState, memoryId: string): boolean export function scratchpadWrite(beam: BeamMemoryState, content: string): string { const padId = generateId(content); const timestamp = toUtcIso(); - beam.db - .prepare(` + beam.db.run( + ` INSERT INTO scratchpad (id, content, session_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET content = excluded.content, updated_at = excluded.updated_at - `) - .run(padId, content, beam.sessionId, timestamp, timestamp); + `, + [padId, content, beam.sessionId, timestamp, timestamp], + ); return padId; } export function scratchpadRead(beam: BeamMemoryState): Row[] { + using statement = beam.db.prepare(` + SELECT id, content, created_at, updated_at + FROM scratchpad + WHERE session_id = ? + ORDER BY updated_at DESC + LIMIT ? + `); return ( - beam.db - .prepare(` - SELECT id, content, created_at, updated_at - FROM scratchpad - WHERE session_id = ? - ORDER BY updated_at DESC - LIMIT ? - `) - .all(beam.sessionId, Number.isFinite(SCRATCHPAD_MAX_ITEMS) ? SCRATCHPAD_MAX_ITEMS : 1000) as Row[] + statement.all(beam.sessionId, Number.isFinite(SCRATCHPAD_MAX_ITEMS) ? SCRATCHPAD_MAX_ITEMS : 1000) as Row[] ).map(rowToDict); } export function scratchpadClear(beam: BeamMemoryState): void { - beam.db.prepare("DELETE FROM scratchpad WHERE session_id = ?").run(beam.sessionId); + beam.db.run("DELETE FROM scratchpad WHERE session_id = ?", [beam.sessionId]); } export function exportToDict(beam: BeamMemoryState): Record { const db = beam.db; + using workingStatement = db.prepare(` + SELECT id, content, source, timestamp, session_id, importance, + embed_text, + metadata_json, valid_until, superseded_by, scope, + recall_count, last_recalled, created_at, veracity, consolidated_at, + memory_type, author_id, author_type, channel_id, trust_tier, + event_date, event_date_precision, temporal_tags + FROM working_memory + ORDER BY session_id, timestamp + `); + using episodicStatement = db.prepare(` + SELECT rowid, id, content, source, timestamp, session_id, importance, + metadata_json, summary_of, valid_until, superseded_by, scope, + recall_count, last_recalled, created_at, veracity, memory_type, + author_id, author_type, channel_id, trust_tier, + event_date, event_date_precision, temporal_tags + FROM episodic_memory + ORDER BY session_id, timestamp + `); + using scratchpadStatement = db.prepare(` + SELECT id, content, session_id, created_at, updated_at + FROM scratchpad + ORDER BY session_id, updated_at + `); + using consolidationStatement = db.prepare(` + SELECT id, session_id, items_consolidated, summary_preview, created_at + FROM consolidation_log + ORDER BY session_id, created_at + `); return { mnemopi_export: { version: "1.0", @@ -841,44 +873,11 @@ export function exportToDict(beam: BeamMemoryState): Record { source_db: beam.dbPath ?? ":memory:", component: "beam", }, - working_memory: db - .prepare(` - SELECT id, content, source, timestamp, session_id, importance, - embed_text, - metadata_json, valid_until, superseded_by, scope, - recall_count, last_recalled, created_at, veracity, consolidated_at, - memory_type, author_id, author_type, channel_id, trust_tier, - event_date, event_date_precision, temporal_tags - FROM working_memory - ORDER BY session_id, timestamp - `) - .all(), - episodic_memory: db - .prepare(` - SELECT rowid, id, content, source, timestamp, session_id, importance, - metadata_json, summary_of, valid_until, superseded_by, scope, - recall_count, last_recalled, created_at, veracity, memory_type, - author_id, author_type, channel_id, trust_tier, - event_date, event_date_precision, temporal_tags - FROM episodic_memory - ORDER BY session_id, timestamp - `) - .all(), + working_memory: workingStatement.all(), + episodic_memory: episodicStatement.all(), episodic_embeddings: [], - scratchpad: db - .prepare(` - SELECT id, content, session_id, created_at, updated_at - FROM scratchpad - ORDER BY session_id, updated_at - `) - .all(), - consolidation_log: db - .prepare(` - SELECT id, session_id, items_consolidated, summary_preview, created_at - FROM consolidation_log - ORDER BY session_id, created_at - `) - .all(), + scratchpad: scratchpadStatement.all(), + consolidation_log: consolidationStatement.all(), }; } @@ -901,50 +900,54 @@ export function importFromDict(beam: BeamMemoryState, data: Record { const counts: Record = {}; - for (const row of db - .prepare("SELECT source, COUNT(*) AS total FROM working_memory GROUP BY source") - .all() as Row[]) { + using statement = db.prepare("SELECT source, COUNT(*) AS total FROM working_memory GROUP BY source"); + for (const row of statement.all() as Row[]) { counts[String(row.source ?? "") || "conversation"] = Number(row.total ?? 0); } return counts; @@ -482,7 +482,8 @@ export class Mnemopi { const episodic = this.#withRuntimeOptions(() => this.beam.getEpisodicStats(authorId, authorType, channelId)); const totalMemories = countRows(this.conn, "SELECT COUNT(*) AS total FROM working_memory"); const totalSessions = countRows(this.conn, "SELECT COUNT(DISTINCT session_id) AS total FROM working_memory"); - const last = this.conn.prepare("SELECT timestamp FROM working_memory ORDER BY timestamp DESC LIMIT 1").get() as { + using lastStatement = this.conn.prepare("SELECT timestamp FROM working_memory ORDER BY timestamp DESC LIMIT 1"); + const last = lastStatement.get() as { timestamp: string | null; } | null; const tripleTotal = countRows(this.conn, "SELECT COUNT(*) AS total FROM triples"); diff --git a/packages/mnemopi/src/core/migrations/e6-triplestore-split.ts b/packages/mnemopi/src/core/migrations/e6-triplestore-split.ts index 4353b926e..8ea1cbb09 100644 --- a/packages/mnemopi/src/core/migrations/e6-triplestore-split.ts +++ b/packages/mnemopi/src/core/migrations/e6-triplestore-split.ts @@ -121,7 +121,7 @@ function kindCounts(rows: readonly TripleCandidateRow[]): Record function migrateRows(db: Database, rows: readonly TripleCandidateRow[]): number { if (rows.length === 0) return 0; - const insert = db.prepare(` + using insert = db.prepare(` INSERT OR IGNORE INTO annotations (memory_id, kind, value, source, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?) `); diff --git a/packages/mnemopi/test/statement-lifetime.test.ts b/packages/mnemopi/test/statement-lifetime.test.ts new file mode 100644 index 000000000..be79b3bec --- /dev/null +++ b/packages/mnemopi/test/statement-lifetime.test.ts @@ -0,0 +1,72 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { BeamMemory } from "@oh-my-pi/pi-mnemopi/core/beam"; +import { openDatabase } from "@oh-my-pi/pi-mnemopi/db"; + +const cleanup: string[] = []; + +function tempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "mnemopi-statement-lifetime-")); + cleanup.push(dir); + return dir; +} + +afterEach(() => { + while (cleanup.length > 0) { + const dir = cleanup.pop(); + if (dir) rmSync(dir, { recursive: true, force: true }); + } +}); + +describe("prepared statement lifetime", () => { + // An unfinalized statement keeps the SQLite connection open, so close() is a + // no-op and the file stays locked. close(true) is the direct assertion: + // closeQuietly() swallows the same error, which is why the leak went unseen. + it("leaves no statement holding the connection after the store paths run", () => { + const dbPath = join(tempDir(), "mnemopi.db"); + const beam = new BeamMemory({ sessionId: "lifetime", dbPath }); + const id = beam.remember("statement lifetime check", { source: "test", importance: 0.5 }); + beam.get(id); + beam.scratchpadWrite("pad entry"); + beam.scratchpadRead(); + beam.updateWorking(id, "statement lifetime check, edited"); + beam.getWorkingStats(); + beam.exportToDict(); + beam.forgetWorking(id); + beam.scratchpadClear(); + + expect(() => beam.db.close(true)).not.toThrow(); + }); + + // SQLite removes the -wal/-shm sidecars when the last connection closes + // cleanly, so their presence after close() means the connection outlived it. + // On Windows that also makes the bank undeletable, which is the user-visible + // half of the bug; on POSIX an open file can still be unlinked. + it("releases the database file so a closed bank can be deleted", () => { + const dbPath = join(tempDir(), "mnemopi.db"); + const beam = new BeamMemory({ sessionId: "lifetime", dbPath }); + const id = beam.remember("deletable after close", { source: "test" }); + beam.get(id); + beam.forgetWorking(id); + beam.close(); + + expect(existsSync(`${dbPath}-wal`)).toBe(false); + expect(existsSync(`${dbPath}-shm`)).toBe(false); + expect(() => rmSync(dbPath)).not.toThrow(); + expect(existsSync(dbPath)).toBe(false); + }); + + it("leaves no statement holding the connection after an annotation import round-trip", () => { + const dbPath = join(tempDir(), "mnemopi.db"); + const beam = new BeamMemory({ sessionId: "lifetime", dbPath }); + beam.remember("annotated row", { source: "test" }); + const exported = beam.exportToDict(); + beam.close(); + + const db = openDatabase(dbPath); + expect(() => db.close(true)).not.toThrow(); + expect(exported.working_memory).toBeDefined(); + }); +});