smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(mnemopi): release one-shot prepared statements

Browse Source 
Every `db.prepare(...)` in the package was a one-shot: prepared, stepped
once via run/get/all, then dropped without `finalize()`. There were 65 such
sites and no `finalize()` call anywhere. An unfinalized statement keeps the
SQLite connection alive, so `BeamMemory.close()` -> `closeQuietly(db)` ->
`db.close()` never released the file. `closeQuietly` swallows the error;
calling `db.close(true)` instead surfaces "database is locked".

One-shot writes now go through `db.run(sql, params)`, which prepares and
finalizes internally. Statements that are read from, or reused across a
loop, are bound with `using` so they are finalized when they leave scope.

On Windows the effect is user-visible beyond tests: the memory DB and its
`-wal`/`-shm` sidecars stay locked after mnemopi closes them, so the file
cannot be deleted, moved, or rotated. On POSIX the leak is silent because
open files can be unlinked.

The new suite pins both halves: that `db.close(true)` does not throw after
the store paths run, and that a closed bank leaves no `-wal`/`-shm` behind
and can be deleted. All three cases fail on the unfixed sources.

Same class as 14252e71c and #6762, neither of which covered this package.
This commit is contained in:
Cyrus
2026-07-30 19:53:57 +08:00
parent 8db0228f4d
commit 0c9cc54561
8 changed files with 342 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/mnemopi/CHANGELOG.md
+4
View File
@@ -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
packages/mnemopi/src/core/annotations.ts
+25 -26
View File
@@ -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<number, StoredAnnotationContent>();
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);
packages/mnemopi/src/core/beam/consolidate.ts
+2 -2
View File
@@ -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<string, JsonValue> {
packages/mnemopi/src/core/beam/helpers.ts
+1 -1
View File
@@ -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[]) => {
packages/mnemopi/src/core/beam/store.ts
+231 -225
View File
@@ -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<string>(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<string, unknown> {
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<string, unknown> {
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<string, unkno
const item = jsonObject(raw);
const id = String(item.id ?? "");
if (id.length === 0) continue;
const exists = db.prepare("SELECT 1 FROM working_memory WHERE id = ?").get(id) !== null;
using existsStatement = db.prepare("SELECT 1 FROM working_memory WHERE id = ?");
const exists = existsStatement.get(id) !== null;
if (exists && !force) {
stats.working_memory.skipped++;
continue;
}
if (exists) {
db.prepare("DELETE FROM working_memory WHERE id = ?").run(id);
db.run("DELETE FROM working_memory WHERE id = ?", [id]);
purgeWorkingMemoryArtifacts(db, [id]);
stats.working_memory.overwritten++;
} else {
stats.working_memory.inserted++;
}
db.prepare(`
db.run(
`
INSERT INTO working_memory
(id, content, source, timestamp, session_id, importance, metadata_json,
valid_until, superseded_by, scope, recall_count, last_recalled, created_at,
veracity, consolidated_at, memory_type, embed_text, author_id, author_type, channel_id,
trust_tier, event_date, event_date_precision, temporal_tags)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
id,
sqlBinding(item.content, ""),
sqlBinding(item.source, null),
sqlBinding(item.timestamp, null),
sqlBinding(item.session_id, "default"),
sqlBinding(item.importance, 0.5),
sqlBinding(item.metadata_json, "{}"),
sqlBinding(item.valid_until, null),
sqlBinding(item.superseded_by, null),
sqlBinding(item.scope, "session"),
sqlBinding(item.recall_count, 0),
sqlBinding(item.last_recalled, null),
sqlBinding(item.created_at, null),
clampVeracity(item.veracity),
item.consolidated_at == null ? importedAt : sqlBinding(item.consolidated_at, importedAt),
sqlBinding(item.memory_type, "unknown"),
sqlBinding(item.embed_text, null),
sqlBinding(item.author_id, null),
sqlBinding(item.author_type, null),
sqlBinding(item.channel_id, null),
sqlBinding(item.trust_tier, "STATED"),
sqlBinding(item.event_date, null),
sqlBinding(item.event_date_precision, "unknown"),
sqlBinding(item.temporal_tags, "[]"),
`,
[
id,
sqlBinding(item.content, ""),
sqlBinding(item.source, null),
sqlBinding(item.timestamp, null),
sqlBinding(item.session_id, "default"),
sqlBinding(item.importance, 0.5),
sqlBinding(item.metadata_json, "{}"),
sqlBinding(item.valid_until, null),
sqlBinding(item.superseded_by, null),
sqlBinding(item.scope, "session"),
sqlBinding(item.recall_count, 0),
sqlBinding(item.last_recalled, null),
sqlBinding(item.created_at, null),
clampVeracity(item.veracity),
item.consolidated_at == null ? importedAt : sqlBinding(item.consolidated_at, importedAt),
sqlBinding(item.memory_type, "unknown"),
sqlBinding(item.embed_text, null),
sqlBinding(item.author_id, null),
sqlBinding(item.author_type, null),
sqlBinding(item.channel_id, null),
sqlBinding(item.trust_tier, "STATED"),
sqlBinding(item.event_date, null),
sqlBinding(item.event_date_precision, "unknown"),
sqlBinding(item.temporal_tags, "[]"),
],
);
}
@@ -952,61 +955,66 @@ export function importFromDict(beam: BeamMemoryState, data: Record<string, unkno
const item = jsonObject(raw);
const id = String(item.id ?? "");
if (id.length === 0) continue;
const exists = db.prepare("SELECT 1 FROM episodic_memory WHERE id = ?").get(id) !== null;
using existsStatement = db.prepare("SELECT 1 FROM episodic_memory WHERE id = ?");
const exists = existsStatement.get(id) !== null;
if (exists && !force) {
stats.episodic_memory.skipped++;
continue;
}
using rowidStatement = db.prepare("SELECT rowid FROM episodic_memory WHERE id = ?");
if (exists) {
const existingRow = db.prepare("SELECT rowid FROM episodic_memory WHERE id = ?").get(id) as {
const existingRow = rowidStatement.get(id) as {
rowid: number;
} | null;
if (existingRow !== null && vecAvailable(db)) {
try {
db.prepare("DELETE FROM vec_episodes WHERE rowid = ?").run(existingRow.rowid);
db.run("DELETE FROM vec_episodes WHERE rowid = ?", [existingRow.rowid]);
} catch {
// sqlite-vec cleanup is best-effort; import correctness takes precedence.
}
}
db.prepare("DELETE FROM episodic_memory WHERE id = ?").run(id);
db.run("DELETE FROM episodic_memory WHERE id = ?", [id]);
stats.episodic_memory.overwritten++;
} else {
stats.episodic_memory.inserted++;
}
db.prepare(`
db.run(
`
INSERT INTO episodic_memory
(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)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
id,
sqlBinding(item.content, ""),
sqlBinding(item.source, null),
sqlBinding(item.timestamp, null),
sqlBinding(item.session_id, "default"),
sqlBinding(item.importance, 0.5),
sqlBinding(item.metadata_json, "{}"),
sqlBinding(item.summary_of, ""),
sqlBinding(item.valid_until, null),
sqlBinding(item.superseded_by, null),
sqlBinding(item.scope, "session"),
sqlBinding(item.recall_count, 0),
sqlBinding(item.last_recalled, null),
sqlBinding(item.created_at, null),
clampVeracity(item.veracity),
sqlBinding(item.memory_type, "unknown"),
sqlBinding(item.author_id, null),
sqlBinding(item.author_type, null),
sqlBinding(item.channel_id, null),
sqlBinding(item.trust_tier, "STATED"),
sqlBinding(item.event_date, null),
sqlBinding(item.event_date_precision, "unknown"),
sqlBinding(item.temporal_tags, "[]"),
`,
[
id,
sqlBinding(item.content, ""),
sqlBinding(item.source, null),
sqlBinding(item.timestamp, null),
sqlBinding(item.session_id, "default"),
sqlBinding(item.importance, 0.5),
sqlBinding(item.metadata_json, "{}"),
sqlBinding(item.summary_of, ""),
sqlBinding(item.valid_until, null),
sqlBinding(item.superseded_by, null),
sqlBinding(item.scope, "session"),
sqlBinding(item.recall_count, 0),
sqlBinding(item.last_recalled, null),
sqlBinding(item.created_at, null),
clampVeracity(item.veracity),
sqlBinding(item.memory_type, "unknown"),
sqlBinding(item.author_id, null),
sqlBinding(item.author_type, null),
sqlBinding(item.channel_id, null),
sqlBinding(item.trust_tier, "STATED"),
sqlBinding(item.event_date, null),
sqlBinding(item.event_date_precision, "unknown"),
sqlBinding(item.temporal_tags, "[]"),
],
);
const oldRowid = Number(item.rowid);
const newRow = db.prepare("SELECT rowid FROM episodic_memory WHERE id = ?").get(id) as {
const newRow = rowidStatement.get(id) as {
rowid: number;
} | null;
if (Number.isFinite(oldRowid) && newRow !== null) oldToNewRowid.set(oldRowid, newRow.rowid);
@@ -1033,41 +1041,39 @@ export function importFromDict(beam: BeamMemoryState, data: Record<string, unkno
const item = jsonObject(raw);
const id = String(item.id ?? "");
if (id.length === 0) continue;
const exists = db.prepare("SELECT 1 FROM scratchpad WHERE id = ?").get(id) !== null;
using existsStatement = db.prepare("SELECT 1 FROM scratchpad WHERE id = ?");
const exists = existsStatement.get(id) !== null;
if (exists) {
db.prepare(
"UPDATE scratchpad SET content = ?, session_id = ?, created_at = ?, updated_at = ? WHERE id = ?",
).run(
db.run("UPDATE scratchpad SET content = ?, session_id = ?, created_at = ?, updated_at = ? WHERE id = ?", [
sqlBinding(item.content, ""),
sqlBinding(item.session_id, "default"),
sqlBinding(item.created_at, null),
sqlBinding(item.updated_at, null),
id,
);
]);
stats.scratchpad.updated++;
} else {
db.prepare(
"INSERT INTO scratchpad (id, content, session_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?)",
).run(
db.run("INSERT INTO scratchpad (id, content, session_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", [
id,
sqlBinding(item.content, ""),
sqlBinding(item.session_id, "default"),
sqlBinding(item.created_at, null),
sqlBinding(item.updated_at, null),
);
]);
stats.scratchpad.inserted++;
}
}
for (const raw of Array.isArray(data.consolidation_log) ? data.consolidation_log : []) {
const item = jsonObject(raw);
db.prepare(
db.run(
"INSERT INTO consolidation_log (session_id, items_consolidated, summary_preview, created_at) VALUES (?, ?, ?, ?)",
).run(
sqlBinding(item.session_id, "default"),
sqlBinding(item.items_consolidated, 0),
sqlBinding(item.summary_preview, ""),
sqlBinding(item.created_at, null),
[
sqlBinding(item.session_id, "default"),
sqlBinding(item.items_consolidated, 0),
sqlBinding(item.summary_preview, ""),
sqlBinding(item.created_at, null),
],
);
stats.consolidation_log.inserted++;
}
packages/mnemopi/src/core/memory.ts
+6 -5
View File
@@ -329,7 +329,8 @@ function toRecallOptions(options: RecallFacadeOptions): BeamRecallFacadeOptions
}
function countRows(db: Database, sql: string, ...params: (string | number | null)[]): number {
const row = db.prepare(sql).get(...params) as { total?: number; count?: number } | null;
using statement = db.prepare(sql);
const row = statement.get(...params) as { total?: number; count?: number } | null;
return row?.total ?? row?.count ?? 0;
}
@@ -344,9 +345,8 @@ function dataDirForDbPath(path: string): string | undefined {
function sourceCounts(db: Database): Record<string, number> {
const counts: Record<string, number> = {};
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");
packages/mnemopi/src/core/migrations/e6-triplestore-split.ts
+1 -1
View File
@@ -121,7 +121,7 @@ function kindCounts(rows: readonly TripleCandidateRow[]): Record<string, number>
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 (?, ?, ?, ?, ?, ?)
`);
packages/mnemopi/test/statement-lifetime.test.ts
+72
View File
@@ -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();
});
});