Persistence

By default the cache is in-memory. Pass a CognitionStore and units (with their understanding, retained raw, and provenance) persist — and the invalidation indexes rebuild on startup, so source-change invalidation keeps working after a restart.

Durable SQLite

Stdlib only — no server. Pass a file path:

from coalent import SemanticCache, SQLiteCognitionStore

store = SQLiteCognitionStore("coalent.db")
cache = SemanticCache(retriever, synthesizer, store=store)

cache.get("our leave policy")   # built once, written through to disk

Restart your process, point a new cache at the same file, and the unit is a hit — and source_changed still invalidates it:

store = SQLiteCognitionStore("coalent.db")
cache = SemanticCache(retriever, synthesizer, store=store)

cache.get("our leave policy").cache_hit          # True — restored from disk
cache.source_changed("confluence:98231", text=…) # still dirties the right unit

Durable Redis

For a cache shared across processes or hosts, use RedisCognitionStore. Bring your own redis client (so nothing is pinned to your lockfile), or build one from a URL with the coalent[redis] extra:

import redis
from coalent import SemanticCache, RedisCognitionStore

store = RedisCognitionStore(redis.Redis.from_url("redis://localhost:6379/0"))
# or, with the extra installed:
store = RedisCognitionStore.from_url("redis://localhost:6379/0")

cache = SemanticCache(retriever, synthesizer, store=store)

Same restart-safe contract as SQLite — units are stored as JSON under a key prefix and the invalidation graph rebuilds on startup. Pass a prefix= to isolate tenants or environments.

In-memory store

InMemoryCognitionStore() is the explicit, ephemeral default — handy when you want the store API without durability.

Bring your own store

CognitionStore is a small protocol — wrap Postgres, DynamoDB, or any KV (SQLite and Redis ship built-in):

class CognitionStore(Protocol):
    def get(self, unit_id: str) -> Cognition | None: ...
    def put(self, unit: Cognition) -> None: ...
    def delete(self, unit_id: str) -> None: ...
    def all(self) -> list[Cognition]: ...   # used to rebuild indexes on load
    def __len__(self) -> int: ...

Use coalent.semantic.serde (cognition_to_json / cognition_from_json) to serialize, exactly as the SQLite store does.

Next

• Provenance & freshness — what survives the restart and keeps working.
• End-to-end example — persistence in a full app.