diff --git a/packages/mnemopi/CHANGELOG.md b/packages/mnemopi/CHANGELOG.md index 21170cfd9..938ea7844 100644 --- a/packages/mnemopi/CHANGELOG.md +++ b/packages/mnemopi/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed working-memory TTL trim silently deleting restored or imported durable rows: rows keeping `consolidated_at = NULL` with an old `timestamp` are no longer trimmed when flagged `IMPORTED`, `importFromDict` stamps imported rows as consolidated, and every working-memory delete path (trim, `forgetWorking`, force-import overwrite) now cascades linked annotations, embeddings, facts, memoria projections, gists, and graph edges instead of leaving orphans. ([#4819](https://github.com/can1357/oh-my-pi/issues/4819)) + ## [16.3.9] - 2026-07-06 ### Fixed diff --git a/packages/mnemopi/src/core/beam/store.ts b/packages/mnemopi/src/core/beam/store.ts index 5d0d2f9dd..e541d0ddb 100644 --- a/packages/mnemopi/src/core/beam/store.ts +++ b/packages/mnemopi/src/core/beam/store.ts @@ -163,27 +163,106 @@ function findDuplicate(beam: BeamMemoryState, content: string): 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 + ); +} + +/** Tables whose rows point back to a `working_memory` id via `source_memory_id`. */ +const MEMORIA_SOURCE_TABLES = [ + "memoria_facts", + "memoria_instructions", + "memoria_kg", + "memoria_preferences", + "memoria_timelines", +] as const; + +/** + * Remove every artifact linked to the given `working_memory` ids so no deletion + * path leaves orphans behind. Covers annotations, embeddings, extracted facts + * (`facts.source_msg_id`), memoria projections (`*.source_memory_id`), episodic + * gists, and the graph edges tied to those memory / gist / fact node ids. + * + * Idempotent and schema-tolerant: `gists` / `graph_edges` only exist once an + * `EpisodicGraph` has initialised, so they are guarded. Callers own the + * transaction and the base `working_memory` delete. + */ +function purgeWorkingMemoryArtifacts(db: BeamMemoryState["db"], ids: readonly string[]): void { + if (ids.length === 0) return; + const placeholders = ids.map(() => "?").join(", "); + + 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 { + 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.prepare(`DELETE FROM annotations WHERE memory_id IN (${placeholders})`).run(...ids); + db.prepare(`DELETE FROM memory_embeddings WHERE memory_id IN (${placeholders})`).run(...ids); + for (const table of MEMORIA_SOURCE_TABLES) { + db.prepare(`DELETE FROM ${table} WHERE source_memory_id IN (${placeholders})`).run(...ids); + } + + if (tableExists(db, "gists")) { + db.prepare(`DELETE FROM gists WHERE memory_id IN (${placeholders})`).run(...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( + ...refs, + ...refs, + ); + } +} + +/** + * TTL / overflow trim for transient working memory. Only genuine scratch is + * eligible: `consolidated_at IS NULL` no longer suffices on its own, since + * restored or imported durable rows legitimately carry a NULL consolidation + * marker with an old event timestamp (issue #4819). Rows flagged `IMPORTED` + * are treated as durable and never trimmed, and trimmed rows cascade all linked + * artifacts via `purgeWorkingMemoryArtifacts`. + */ function trimWorkingMemory(beam: BeamMemoryState): void { const limit = beam.config.workingMemoryLimit; if (!Number.isFinite(limit) || limit <= 0) return; const ttlHours = beam.config.workingMemoryTtlHours; const cutoff = toUtcIso(new Date(Date.now() - ttlHours * 3_600_000)); - beam.db - .prepare(` - DELETE FROM working_memory - WHERE session_id = ? - AND consolidated_at IS NULL - AND ( - timestamp < ? OR - id NOT IN ( - SELECT id FROM working_memory - WHERE session_id = ? AND consolidated_at IS NULL - ORDER BY timestamp DESC - LIMIT ? - ) - ) - `) - .run(beam.sessionId, cutoff, beam.sessionId, limit); + const ids = ( + 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' + ORDER BY timestamp DESC + LIMIT ? + ) + ) + `) + .all(beam.sessionId, cutoff, beam.sessionId, limit) as { id: string }[] + ).map(row => row.id); + if (ids.length === 0) return; + const placeholders = ids.map(() => "?").join(", "); + transaction(beam.db, () => { + beam.db + .prepare(`DELETE FROM working_memory WHERE id IN (${placeholders}) AND session_id = ?`) + .run(...ids, beam.sessionId); + purgeWorkingMemoryArtifacts(beam.db, ids); + }); } function addTemporalAnnotations(beam: BeamMemoryState, memoryId: string, timestamp: string, source: string): void { @@ -676,7 +755,7 @@ export function forgetWorking(beam: BeamMemoryState, memoryId: string): boolean .run(memoryId, beam.sessionId); deleted = result.changes; if (deleted > 0) { - beam.db.prepare("DELETE FROM annotations WHERE memory_id = ?").run(memoryId); + purgeWorkingMemoryArtifacts(beam.db, [memoryId]); } }); if (deleted > 0) invalidateCaches(beam); @@ -772,6 +851,10 @@ export function importFromDict(beam: BeamMemoryState, data: Record(); transaction(db, () => { @@ -786,6 +869,7 @@ export function importFromDict(beam: BeamMemoryState, data: Record { expect(dest.db.prepare("SELECT COUNT(*) AS count FROM scratchpad").get()).toEqual({ count: 1 }); expect(scratchpadRead(dest).map(row => row.content)).toEqual([]); }); + + it("keeps restored durable rows and cascades linked artifacts on trim, force-import, and forget (issue #4819)", () => { + const beam = makeState("trim-4819"); + // EpisodicGraph owns the `gists` / `graph_edges` schema; init it so the + // cascade can be exercised end to end on the shared connection. + new EpisodicGraph({ db: beam.db, dbPath: ":memory:" }); + const oldTimestamp = new Date(Date.now() - 1000 * 3_600_000).toISOString(); + const countOf = (sql: string, ...params: (string | number | null)[]): number => { + const row = beam.db.prepare(sql).get(...params) as { count: number }; + return row.count; + }; + const seedArtifacts = (memoryId: string): void => { + beam.db + .prepare("INSERT INTO annotations (memory_id, kind, value) VALUES (?, 'mentions', 'Alice')") + .run(memoryId); + beam.db + .prepare("INSERT INTO memory_embeddings (memory_id, embedding_json, model) VALUES (?, '[0.1]', 't')") + .run(memoryId); + beam.db + .prepare( + "INSERT INTO facts (fact_id, session_id, subject, predicate, object, source_msg_id) VALUES (?, 'trim-4819', 'Alice', 'is', 'User', ?)", + ) + .run(`fact-${memoryId}`, memoryId); + beam.db + .prepare( + "INSERT INTO memoria_facts (session_id, fact_type, key, value, source_memory_id) VALUES ('trim-4819', 'name', 'name', 'Alice', ?)", + ) + .run(memoryId); + beam.db + .prepare("INSERT INTO gists (id, text, memory_id) VALUES (?, 'g', ?)") + .run(`gist_${memoryId}`, memoryId); + beam.db + .prepare("INSERT INTO graph_edges (source, target, edge_type) VALUES (?, ?, 'ctx')") + .run(memoryId, `gist_${memoryId}`); + beam.db + .prepare("INSERT INTO graph_edges (source, target, edge_type) VALUES (?, ?, 'rel')") + .run(`gist_${memoryId}`, `fact-${memoryId}`); + }; + const artifactCount = (memoryId: string): number => + countOf("SELECT COUNT(*) AS count FROM annotations WHERE memory_id = ?", memoryId) + + countOf("SELECT COUNT(*) AS count FROM memory_embeddings WHERE memory_id = ?", memoryId) + + countOf("SELECT COUNT(*) AS count FROM facts WHERE source_msg_id = ?", memoryId) + + countOf("SELECT COUNT(*) AS count FROM memoria_facts WHERE source_memory_id = ?", memoryId) + + countOf("SELECT COUNT(*) AS count FROM gists WHERE memory_id = ?", memoryId) + + countOf( + "SELECT COUNT(*) AS count FROM graph_edges WHERE source = ? OR target = ? OR source = ? OR target = ?", + memoryId, + memoryId, + `gist_${memoryId}`, + `gist_${memoryId}`, + ); + + // (1) Restored durable row: old timestamp, consolidated_at NULL, IMPORTED tier. + const durableId = "restored-durable"; + beam.db + .prepare( + "INSERT INTO working_memory (id, content, source, timestamp, session_id, importance, trust_tier, consolidated_at) VALUES (?, 'canonical fact', 'backup', ?, 'trim-4819', 0.9, 'IMPORTED', NULL)", + ) + .run(durableId, oldTimestamp); + seedArtifacts(durableId); + + // (2) Transient scratch row: old timestamp, consolidated_at NULL, STATED tier. + const transientId = "transient-scratch"; + beam.db + .prepare( + "INSERT INTO working_memory (id, content, source, timestamp, session_id, importance, trust_tier, consolidated_at) VALUES (?, 'idle chatter', 'conversation', ?, 'trim-4819', 0.2, 'STATED', NULL)", + ) + .run(transientId, oldTimestamp); + seedArtifacts(transientId); + + // A normal write triggers the automatic trim. + remember(beam, "a fresh conversational note", { source: "conversation" }); + + // Durable row survives with all artifacts intact. + expect(get(beam, durableId)?.content).toBe("canonical fact"); + expect(artifactCount(durableId)).toBe(7); + + // Transient old row is trimmed and every linked artifact cascades. + expect(get(beam, durableId) === null).toBe(false); + expect(countOf("SELECT COUNT(*) AS count FROM working_memory WHERE id = ?", transientId)).toBe(0); + expect(artifactCount(transientId)).toBe(0); + + // (3) forgetWorking cascades every linked artifact, not just annotations. + expect(forgetWorking(beam, durableId)).toBe(true); + expect(artifactCount(durableId)).toBe(0); + }); + + it("marks imported working memory as consolidated so restored banks survive trim (issue #4819)", () => { + const dest = makeState("import-4819"); + const oldTimestamp = new Date(Date.now() - 1000 * 3_600_000).toISOString(); + importFromDict( + dest, + { + working_memory: [ + { + id: "restored-import", + content: "durable restored fact", + timestamp: oldTimestamp, + session_id: "import-4819", + trust_tier: "STATED", + consolidated_at: null, + }, + ], + }, + true, + ); + const importedRow = dest.db + .prepare("SELECT consolidated_at FROM working_memory WHERE id = 'restored-import'") + .get() as { consolidated_at: string | null }; + expect(importedRow.consolidated_at).not.toBeNull(); + + remember(dest, "a fresh note", { source: "conversation" }); + expect(get(dest, "restored-import")?.content).toBe("durable restored fact"); + }); + + it("force-import overwrite cleans stale linked artifacts of the replaced row (issue #4819)", () => { + const dest = makeState("import-overwrite-4819"); + const id = "overwrite-me"; + dest.db + .prepare( + "INSERT INTO working_memory (id, content, source, timestamp, session_id, importance, trust_tier) VALUES (?, 'stale', 'backup', '2020-01-01T00:00:00.000Z', 'import-overwrite-4819', 0.5, 'IMPORTED')", + ) + .run(id); + dest.db.prepare("INSERT INTO annotations (memory_id, kind, value) VALUES (?, 'mentions', 'Stale')").run(id); + dest.db + .prepare("INSERT INTO memory_embeddings (memory_id, embedding_json, model) VALUES (?, '[0.9]', 'old')") + .run(id); + dest.db + .prepare( + "INSERT INTO facts (fact_id, session_id, subject, predicate, object, source_msg_id) VALUES ('stale-fact', 'import-overwrite-4819', 'a', 'is', 'b', ?)", + ) + .run(id); + + importFromDict( + dest, + { + working_memory: [ + { id, content: "fresh replacement", session_id: "import-overwrite-4819", trust_tier: "IMPORTED" }, + ], + }, + true, + ); + + expect(get(dest, id)?.content).toBe("fresh replacement"); + const staleArtifacts = + ( + dest.db.prepare("SELECT COUNT(*) AS count FROM annotations WHERE memory_id = ?").get(id) as { + count: number; + } + ).count + + ( + dest.db.prepare("SELECT COUNT(*) AS count FROM memory_embeddings WHERE memory_id = ?").get(id) as { + count: number; + } + ).count + + ( + dest.db.prepare("SELECT COUNT(*) AS count FROM facts WHERE source_msg_id = ?").get(id) as { + count: number; + } + ).count; + expect(staleArtifacts).toBe(0); + }); });