LongTermMemory

Purpose

SQLite-backed persistent store for MemoryEntry objects, with multi-tenant partitioning, importance-based retrieval, and time-based decay. This is the durable tier of the memory stack — unlike ShortTermMemory or WorkingMemory, its contents survive process restarts.

Constructor

def __init__(self, db_path: str = "./memory_storage/ltm.db", *, tenant_id: str = "default") -> None

On construction: creates the memories table (and three indexes: idx_tenant, idx_importance, idx_timestamp) if they don't already exist, via CREATE TABLE IF NOT EXISTS. A fresh threading.Lock is created per instance.

Schema

CREATE TABLE IF NOT EXISTS memories (
    id          TEXT PRIMARY KEY,
    tenant_id   TEXT NOT NULL,
    content     TEXT NOT NULL,      -- JSON-serialized
    timestamp   REAL NOT NULL,
    memory_type TEXT NOT NULL,      -- MemoryType.value
    importance  REAL NOT NULL DEFAULT 0.5,
    decay_factor REAL NOT NULL DEFAULT 1.0,
    access_count INTEGER NOT NULL DEFAULT 0,
    last_access  REAL NOT NULL,
    tags        TEXT NOT NULL DEFAULT '[]',    -- JSON-serialized list
    metadata    TEXT NOT NULL DEFAULT '{}'      -- JSON-serialized dict
);

Note the primary key is id alone — not (id, tenant_id). If you ever share one LongTermMemory instance/database file across multiple tenant_ids by constructing more than one LongTermMemory pointed at the same db_path, an id collision between two different tenants' entries (unlikely given MD5-based IDs, but possible if two identical-content entries are created with an identical timestamp) would silently overwrite across tenants on store()'s ON CONFLICT(id) DO UPDATE. In practice, AIMemoryManager gives each tenant its own separate .db file, so this is only a risk if you construct LongTermMemory instances yourself and point multiple tenant_ids at the same db_path.

Public methods

store(entry: MemoryEntry) -> None

Upserts (INSERT ... ON CONFLICT(id) DO UPDATE). On conflict, only importance, decay_factor, access_count, last_access, tags, metadata are updated — content, timestamp, and memory_type are not updated on conflict (so re-storing an entry with the same ID but different content will silently keep the original content).

store_many(entries: List[MemoryEntry]) -> None

Uses INSERT OR REPLACE (not ON CONFLICT DO UPDATE) via executemany — this does fully replace all columns on conflict, unlike store()'s partial-update semantics. store() and store_many() therefore have subtly different upsert behavior for the same underlying table.

delete(entry_id: str) -> None

Deletes the row matching id and the instance's tenant_id — will not delete another tenant's entry even if you pass its ID.

delete_below_importance(threshold: float) -> int

Deletes all rows for this tenant with importance < threshold, returns the number of rows deleted (cursor.rowcount). Filters on the raw stored importance column, not importance * decay_factor — if decay hasn't been recently applied via apply_global_decay, an entry's true "effective" importance may be lower than what's stored, and this method won't catch it.

get(entry_id: str) -> Optional[MemoryEntry]

Fetches by id + tenant_id. If found, calls entry.access() (bumping access_count/importance/last_access on the in-memory object) and then persists that change back via _update_access() — so every get() call has a write side effect, not just a read.

get_top_by_importance(limit: int = 20, min_importance: float = 0.0) -> List[MemoryEntry]

SELECT * ... WHERE tenant_id=? AND importance >= ? ORDER BY importance DESC LIMIT ?. Does not call .access() on returned entries (unlike get()), so bulk retrieval doesn't inflate importance/access stats.

get_recent(limit: int = 20) -> List[MemoryEntry]

Same pattern, ordered by timestamp DESC. No .access() side effect.

count() -> int

SELECT COUNT(*) ... WHERE tenant_id=?.

apply_global_decay(decay_rate: float = 0.1) -> int

entries = self.get_top_by_importance(limit=100_000)
for e in entries:
    days = (now - e.last_access) / 86400
    new_decay = max(0.1, 1.0 - decay_rate * days)
    new_imp = max(0.05, e.importance * new_decay)
    # UPDATE ... SET importance=new_imp, decay_factor=new_decay ...

Fetches up to 100,000 entries (with min_importance=0.0 implicitly, since it's not passed), recomputes a linear decay factor from days since last_access, and multiplies it onto the entry's current, already-possibly-decayed importance (not the original importance). Returns the number of rows updated (always equal to the input entry count — every fetched row gets an UPDATE, even if the new values are unchanged).

[!DANGER] Compounding decay on repeated calls Because new_imp = e.importance * new_decay multiplies onto whatever importance is already stored (not original_importance), calling apply_global_decay() twice in quick succession — before last_access changes — computes the same new_decay factor both times but applies it on top of the already-reduced importance the second time, over-decaying the entry relative to a single, correctly-computed decay for that elapsed time window. AIMemoryManager.run_maintenance() calls this method every time it runs, so calling run_maintenance() frequently without intervening access activity will decay long-term memories faster than the decay_rate alone would suggest. See Known Issues.

clear() -> None

Deletes all rows for this tenant (DELETE FROM memories WHERE tenant_id=?).

stats() -> Dict[str, Any]

Returns type, db_path, tenant_id, total_entries (via count()).

Raises

Internal workflow — _connect()

@contextmanager
def _connect(self):
    conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

A brand-new SQLite connection is opened and closed on every single call to any read/write method — there is no persistent connection pool. This is simple and safe (side-steps SQLite's connection/thread affinity rules via check_same_thread=False plus opening fresh each time) but means every operation pays SQLite's connection-open overhead; for very high-frequency access patterns this may be a measurable performance cost.

Thread safety

All write methods (store, store_many, delete, delete_below_importance, apply_global_decay, clear, _update_access) acquire self._lock (a threading.Lock) for the duration of the connection + operation. Read methods (get, get_top_by_importance, get_recent, count) do not acquire the lock — they open their own independent connection, so concurrent reads are safe (SQLite handles concurrent readers), but a read can observe a partially-committed state relative to a concurrent write only in the sense that SQLite's own transaction isolation applies (each connection/transaction is independent here since each call opens and commits its own connection).

Complexity

• store/delete: O(log n) for the underlying B-tree index operations (SQLite primary key + secondary indexes).
• get_top_by_importance/get_recent: O(log n + k) using the idx_importance/idx_timestamp indexes, where k is limit.
• apply_global_decay: O(n) — fetches and updates up to 100,000 rows per call.

Examples

import time
from zerda.memory import LongTermMemory, MemoryEntry, MemoryType

ltm = LongTermMemory(db_path="./data/acme_ltm.db", tenant_id="acme-corp")

# 1. Basic store + retrieve
entry = MemoryEntry(
    content={"input": "Our fiscal year starts in April", "output": "Noted."},
    timestamp=time.time(),
    memory_type=MemoryType.LONG_TERM,
    importance=0.8,
)
ltm.store(entry)
fetched = ltm.get(entry.id)
print(fetched.access_count)  # 1 — get() bumps it

# 2. Intermediate — top entries by importance
top = ltm.get_top_by_importance(limit=5, min_importance=0.5)

# 3. Advanced — periodic maintenance
updated = ltm.apply_global_decay(decay_rate=0.1)
deleted = ltm.delete_below_importance(threshold=0.1)
print(f"{updated} decayed, {deleted} removed")

# 4. Error handling — invalid db_path (e.g. read-only filesystem)
try:
    LongTermMemory(db_path="/root/no-permission/ltm.db")
except OSError as e:
    print("Cannot create persistence directory:", e)

# 5. Production — batch import
entries = [MemoryEntry(content={"input": f"fact {i}", "output": "ok"},
                        timestamp=time.time(), memory_type=MemoryType.LONG_TERM)
           for i in range(100)]
ltm.store_many(entries)
print(ltm.count())

Best practices

• Call apply_global_decay + delete_below_importance on a schedule (e.g. via AIMemoryManager.run_maintenance()), not on every request.
• Don't call apply_global_decay more often than your last_access granularity meaningfully changes — see the compounding-decay caveat.
• Use store_many for bulk imports rather than looping store() — it's a single transaction instead of N separate connections.

Common mistakes

• Sharing one db_path across multiple tenant_ids by constructing several LongTermMemory instances pointed at the same file — the schema doesn't scope the primary key by tenant, only application-level filtering does.
• Calling run_maintenance() (which calls apply_global_decay) very frequently, expecting decay to scale linearly with elapsed time — repeated calls compound.

Related APIs

• MemoryEntry
• ForgettingMechanism — an alternative, more configurable decay/deletion policy, applied to in-memory collections rather than directly to SQLite rows.
• AIMemoryManager — orchestrates one LongTermMemory per tenant.