Implementation Plan: Postgres-backed Session Store (CacheDBInterface adapter)
Goal: store cognee session data (QA entries, agent traces, usage logs, and small string KV values) in Postgres via a new CacheDBInterface adapter, removing the Redis requirement for session memory. Default backend stays fs; Redis and Tapes keep working unchanged. The design keeps a clean seam for a future Turbopuffer backend.
Design stance: minimal-surface adapter that slots into the existing backend plug point (like FSCacheAdapter), follows the graph Postgres adapter's engine/schema patterns, and adds zero new dependencies (SQLAlchemy is core; asyncpg/psycopg2 ship in the existing postgres extra). Where the minimal approach was naive (TTL purging, multi-worker RMW, lock semantics, prune scope), the robust alternatives are grafted in below — each divergence from the minimal design is justified inline.
1. Overview & goals
- Add
CACHE_BACKEND=postgresas a fourth cache backend alongsideredis,fs,tapes. - Implement
PostgresCacheAdapter(CacheDBInterface)with full behavioral parity toRedisAdapter/FSCacheAdapter(TTL semantics, merge semantics, error contracts), improving on known Redis races where doing so is observably identical (atomic deletes,FOR UPDATEupdates). - Support the formal string KV methods on
CacheDBInterface(get_value/set_value/delete_value) for small cache values. The old graph-to-session sync contract is removed; new backends should not add anasync_redisshim for it. - Non-goals (v1): data migration tooling (cache is 7-day-TTL ephemeral), per-dataset DB isolation (sessions are per-user), distributed
SHARED_LADYBUG_LOCKsupport on Postgres (Phase 6), multi-workerimprove()mutex.
2. Current state
Cache interface — cognee/infrastructure/databases/cache/cache_db_interface.py: ABC with sync lock methods (acquire_lock/release_lock, concrete hold_lock() contextmanager), async QA CRUD (create_qa_entry, get_latest_qa_entries, get_all_qa_entries, update_qa_entry, delete_feedback, delete_qa_entry, delete_session), agent traces (append_agent_trace_step, get_agent_trace_session, get_agent_trace_feedback, get_agent_trace_count), prune, close, log_usage/get_usage_logs, plus concrete back-compat shims add_qa/get_latest_qa/get_all_qas (do not reimplement). Base __init__(host, port, lock_key="default_lock", log_key="usage_logs"). Payload models: SessionQAEntry/SessionAgentTraceEntry in cognee/infrastructure/databases/cache/models.py (validators: feedback_score 1–5, used_graph_element_ids only node_ids/edge_ids, memify_metadata str→bool, trace sanitization/truncation).
Backends — cache/redis/RedisAdapter.py (sync redis.Redis for locks + async redis.asyncio.Redis for data; one Redis LIST per key agent_sessions:{user_id}:{session_id} / agent_traces:{user_id}:{session_id} / {log_key}:{user_id}; JSON-string elements; EXPIRE-on-write sliding TTL; FLUSHDB prune); cache/fscache/FsCacheAdapter.py (diskcache, whole list as one JSON value under self.cache, cache.transact(), lock methods raise SharedLadybugLockRequiresRedisError); cache/tapes/TapesCacheAdapter.py (FS subclass mirroring QA creates over HTTP).
Factory — cache/get_cache_engine.py: @lru_cache'd create_cache_engine(...) branches on CacheConfig.cache_backend (Literal["redis","fs","tapes"], default "fs", in cache/config.py); returns None when caching and usage_logging are both off; close_cache_engine() closes + cache_clear()s. Verified gotcha: from ...RedisAdapter import RedisAdapter executes unconditionally inside the caching-on block, before the backend dispatch — it only works because the unused core dep fakeredis[lua] transitively installs redis.
Session manager — cognee/infrastructure/session/session_manager.py (built by get_session_manager.py): wraps every interface method, no-ops when cache_engine is None. It no longer stores graph snapshots in session prompts. For one release, delete_session still deletes the legacy graph_knowledge:{user_id}:{session_id} key via delete_value so old cache rows do not linger.
Locks — two unrelated systems: (1) cognee/infrastructure/locks/session_lock.py: pure in-process asyncio locks (per-(session_id, op) dict + improve-lock set), explicitly single-worker scope, no Redis dependency, no change required; (2) CacheDBInterface.acquire_lock/release_lock: sync methods used only by the Ladybug graph adapter (graph/ladybug/adapter.py ~line 187, via asyncio.to_thread) when SHARED_LADYBUG_LOCK=true — Redis-only today.
Relational sibling — cognee/modules/session_lifecycle/models.py (session_records, session_model_usage): alembic-managed lifecycle/metrics rows, already Postgres-compatible, untouched by this plan. No FK between cache rows and session_records (cache rows TTL-expire; lifecycle rows persist — by design).
Template — cognee/infrastructure/databases/graph/postgres/adapter.py + tables.py: own create_async_engine(uri, json_serializer=lambda obj: json.dumps(obj, cls=JSONEncoder), **pool_args) (verified, lines 40–58), async_sessionmaker(expire_on_commit=False), private MetaData() with create_all(checkfirst=True) in initialize() — not alembic.
3. Target architecture
SessionManager / usage_logger / forget / prune_system / memify tasks
│ (unchanged call sites)
▼
get_cache_engine() ──reads── CacheConfig (.env: CACHE_BACKEND, CACHE_DB_URL, ...)
│ @lru_cache create_cache_engine()
├── "redis" → RedisAdapter (unchanged)
├── "fs" → FSCacheAdapter (unchanged)
├── "tapes" → TapesCacheAdapter (unchanged)
└── "postgres" → PostgresCacheAdapter (NEW, lazy import)
│
├── own async engine (postgresql+asyncpg://...)
│ json_serializer=JSONEncoder (UUID/datetime-safe JSONB)
├── tables (private MetaData, create-on-init):
│ cache_qa_entries ← agent_sessions:{u}:{s} lists
│ cache_trace_entries← agent_traces:{u}:{s} lists
│ cache_usage_logs ← {log_key}:{u} lists
│ cache_kv ← generic small string KV values
└── get_value/set_value/delete_value over cache_kv
4. Postgres adapter design
4.1 Module layout
Mirror the existing backend layout:
cognee/infrastructure/databases/cache/postgres/
├── __init__.py # re-export PostgresCacheAdapter
├── tables.py # private MetaData + SQLAlchemy Core tables
└── PostgresCacheAdapter.py # class PostgresCacheAdapter(CacheDBInterface)
4.2 Class skeleton
class PostgresCacheAdapter(CacheDBInterface):
def __init__(
self,
connection_string: str, # any SQLAlchemy async URL (asyncpg in prod, aiosqlite in tests)
lock_key: str = "default_lock",
log_key: str = "usage_logs",
session_ttl_seconds: int | None = 604800,
agentic_lock_expire: int = 240, # stored now, used by Phase 6 advisory locks
agentic_lock_timeout: int = 300,
purge_interval_seconds: int = 900,
):
super().__init__(host="", port=0, lock_key=lock_key, log_key=log_key)
self.db_uri = connection_string
self.session_ttl_seconds = session_ttl_seconds
pool_args = dict(get_relational_config().pool_args or {})
self.engine = create_async_engine(
connection_string,
json_serializer=lambda obj: json.dumps(obj, cls=JSONEncoder),
**pool_args,
)
self.sessionmaker = async_sessionmaker(bind=self.engine, expire_on_commit=False)
self._initialized = False
self._init_lock = asyncio.Lock()
self._last_purge = 0.0
Notes:
- Call
super().__init__(Redis does; FS doesn't — calling it giveslock_key/log_key/lockattributes). - Single hashable
connection_stringarg keeps the@lru_cache'd factory happy. - Owning the engine (rather than borrowing
get_relational_engine()'s) keepsclose()safe:close_cache_engine()→adapter.close()→engine.dispose()must not kill the shared relational pool. - Engine-level
json_serializerwithcognee.modules.storage.utils.JSONEncoderis load-bearing: without it, asyncpg JSONB inserts containing UUIDs/datetimes fail (graph-adapter gotcha, replicated). - Connection validation is lazy (in
_ensure_initialized), not eager like Redis'sping()— the constructor must stay sync and lru_cache-friendly; the contract only requiresCacheConnectionErrorto surface, which it does on first use. Also raise a clearCacheConnectionError("CACHE_BACKEND=postgres requires cognee[postgres]")if the asyncpg import fails.
4.3 Table DDL (cache/postgres/tables.py)
Private MetaData() — not the relational declarative Base, not alembic-managed (graph-adapter precedent: graph_node/graph_edge are create-on-init while session_records is alembic-managed; private MetaData keeps alembic autogenerate from seeing these). Payload columns use JSONB().with_variant(JSON(), "sqlite") so unit tests run on aiosqlite. cache_-prefixed names avoid collision with session_records/session_model_usage.
-- QA entries: one row per entry; qa_id promoted to a column for direct UPDATE
-- (Redis buries it in JSON and does linear scan + LSET)
CREATE TABLE cache_qa_entries (
seq BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, -- insertion order
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
qa_id TEXT NOT NULL,
payload JSONB NOT NULL, -- full SessionQAEntry.model_dump() (incl. qa_id, time)
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NULL, -- NULL = no expiry
UNIQUE (user_id, session_id, qa_id)
);
CREATE INDEX idx_cache_qa_session ON cache_qa_entries (user_id, session_id, seq);
CREATE INDEX idx_cache_qa_expires ON cache_qa_entries (expires_at) WHERE expires_at IS NOT NULL;
-- Agent traces: append-only
CREATE TABLE cache_trace_entries (
seq BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
payload JSONB NOT NULL, -- SessionAgentTraceEntry.model_dump()
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NULL
);
CREATE INDEX idx_cache_trace_session ON cache_trace_entries (user_id, session_id, seq);
CREATE INDEX idx_cache_trace_expires ON cache_trace_entries (expires_at) WHERE expires_at IS NOT NULL;
-- Usage logs (Redis key {log_key}:{user_id})
CREATE TABLE cache_usage_logs (
seq BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
log_key TEXT NOT NULL, -- adapter's self.log_key
user_id TEXT NOT NULL,
payload JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
expires_at TIMESTAMPTZ NULL
);
CREATE INDEX idx_cache_usage ON cache_usage_logs (log_key, user_id, seq);
-- String KV for small cache values; legacy graph_knowledge:{u}:{s} keys may be deleted by SessionManager
CREATE TABLE cache_kv (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
expires_at TIMESTAMPTZ NULL
);
Notes:
session_idstays a plain string, no FK tosession_records(samesession_idexists under multiple users; cache rows expire while lifecycle rows persist). Keying is always(user_id, session_id)— the exact relational analogue of the Redis key prefixes, which also preserves cross-user reads (recall / sessions router resolve the owner'suser_idfromSessionRecord).- JSONB, not bytea/pickle: parity with
graph_node.propertiesand the Redis JSON-string encoding; queryable for debugging. - Hot-path:
get_latest_qa_entriesruns on every completion (last 10);idx_cache_qa_sessionmakes the tail read an index scan. Rows can be tens of KB (payload embedscontext) — a laterpayload - 'context'projection optimization is possible, but the contract returns full entries today; don't change behavior.
4.4 Method-by-method implementation
Every public async method: await self._ensure_initialized() first (once-flag under self._init_lock; runs conn.run_sync(cache_metadata.create_all, checkfirst=True), wraps first-connect failure in CacheConnectionError), then async with self.sessionmaker() as session, session.begin():. Error contract (pinned by test_redis_adapter_crud.py/test_fs_adapter_crud.py): backend/SQLAlchemy errors → wrap in CacheConnectionError; create-path validation failures → also CacheConnectionError (odd but tested); update-path validation → SessionQAEntryValidationError propagated unwrapped.
| Method | Implementation |
|---|---|
create_qa_entry(user_id, session_id, question, context, answer, qa_id=None, ...) |
qa_id: str | None = None with str(uuid4()) fallback — match the adapters, not the ABC's qa_id: str. Build SessionQAEntry (stamps time=datetime.utcnow().isoformat(), runs validators; failure → CacheConnectionError). INSERT payload=entry.model_dump(); session-wide TTL refresh (§4.5); scoped lazy purge. One transaction. |
get_latest_qa_entries(u, s, last_n=5) |
SELECT payload WHERE u/s AND not-expired ORDER BY seq DESC LIMIT :n, reverse in Python → chronological; SessionQAEntry.model_validate each. Return [] always on empty, including last_n==1 — the Redis None-on-last_n==1 quirk is not replicated; all callers go through SessionManager and treat falsy uniformly (FS already returns []). Pin in the unit test. |
get_all_qa_entries(u, s) |
Same, ORDER BY seq ASC, no LIMIT; [] if none. |
update_qa_entry(u, s, qa_id, ...) |
One transaction: SELECT payload ... WHERE user_id/session_id/qa_id ... FOR UPDATE (with_for_update(); no-op on the sqlite test variant, fine). No row → False. Merge in Python with exact existing semantics: None preserves every field; memify_metadata = {**existing, **new} (MERGE not replace — apply_feedback_weights/apply_frequency_weights idempotency flags depend on it); re-validate via SessionQAEntry.model_validate, letting SessionQAEntryValidationError propagate; UPDATE payload; TTL refresh; True. FOR UPDATE makes this RMW multi-worker-safe — strictly better than Redis's load/LSET race, observably identical. |
delete_feedback(u, s, qa_id) |
Same FOR UPDATE pattern; set feedback_text/feedback_score to None in payload; TTL refresh; bool. (This is the only way to clear feedback — update_qa_entry(feedback_score=None) preserves.) |
delete_qa_entry(u, s, qa_id) |
Single atomic DELETE ... WHERE user_id/session_id/qa_id (replaces Redis's non-atomic DEL+RPUSH-loop rewrite — crash-safe improvement); TTL refresh on remaining rows (Redis re-applies TTL after rewrite); return rowcount > 0. No "drop key when empty" step — empty session ≡ zero rows. |
delete_session(u, s) |
One transaction: DELETE FROM cache_qa_entries + DELETE FROM cache_trace_entries for (u, s); True if either rowcount > 0. (SessionManager.delete_session separately deletes the legacy graph_knowledge: key via delete_value — keep that division for one release.) |
append_agent_trace_step(u, s, trace_id, origin_function, status, ...) |
Build SessionAgentTraceEntry (method_params or {}; model validators do sanitization/truncation for free; blank trace_id/origin_function → CacheConnectionError); INSERT; trace-session TTL refresh. |
get_agent_trace_session(u, s, last_n=None) |
ORDER BY seq ASC; when last_n is not None, DESC LIMIT + reverse. Reads do NOT refresh TTL. |
get_agent_trace_feedback(u, s, last_n=None) |
[e.session_feedback for e in await self.get_agent_trace_session(...)] — exactly the delegation both adapters use. |
get_agent_trace_count(u, s) |
SELECT count(*) with expiry filter; 0 for missing session. |
prune() |
DELETE FROM / TRUNCATE the four cache tables only. Deliberate, documented divergence from Redis FLUSHDB (which nukes co-tenant keys and other users' data) — strictly safer; both proposals agree. |
close() |
await self.engine.dispose(close=True); dispose the sync lock engine if created (Phase 6); idempotent, swallow + debug-log errors (called by close_cache_engine and prune_data.py). Reset self._initialized = False so a reused instance re-inits. |
log_usage(user_id, log_entry, ttl=604800) |
INSERT into cache_usage_logs with log_key=self.log_key, expires_at = now()+ttl if ttl else NULL; also refresh expires_at on the user's existing rows under the same (log_key, user_id) — Redis EXPIREs the whole list, so per-row-only TTL would diverge. |
get_usage_logs(user_id, limit=100) |
SELECT payload WHERE log_key/user_id AND not-expired ORDER BY seq DESC LIMIT :limit — already most-recent-first (Redis does lrange + reverse). |
acquire_lock() / release_lock(lock=None) |
SYNC — do not make async (Ladybug calls them via asyncio.to_thread). v1: raise SharedLadybugLockRequiresRedisError, imported from cognee.infrastructure.databases.exceptions.exceptions directly (it is NOT re-exported in the package __init__) — exactly like FsCacheAdapter. Inherited hold_lock() then raises too — correct. Phase 6 upgrade in §7. |
| Inherited (do not reimplement) | hold_lock(), add_qa, get_latest_qa, get_all_qas. |
Concurrency posture: unlike the graph adapter, no global asyncio.Lock serializing writes — row-level FOR UPDATE + single-statement deletes give correct multi-worker behavior with better throughput; the graph adapter's lock exists for bulk upsert batches that don't occur here. Add the graph adapter's tenacity retry on asyncpg DeadlockDetectedError as belt-and-braces.
4.5 TTL strategy: expires_at + read-time filter + lazy purge + throttled sweep
Replicates Redis sliding whole-key TTL exactly:
- Write refresh: every mutating op runs, inside its transaction,
UPDATE <table> SET expires_at = now() + :ttl WHERE user_id=:u AND session_id=:s(skipped whensession_ttl_secondsisNone/<= 0—expires_atstays NULL, expiry disabled). Whole-session refresh matchesEXPIRE session_key. Sessions are small (tens–hundreds of rows); cheap. - Read filter: every SELECT adds
AND (expires_at IS NULL OR expires_at > now()). Reads never refresh TTL (tested Redis behavior). Correctness never depends on purging. - Purge ("purge on init only" would leak rows in long-lived processes): (a) scoped
DELETE ... WHERE user_id=:u AND session_id=:s AND expires_at <= now()opportunistically on writes to that session; (b) a throttled global sweepDELETE FROM <each table> WHERE expires_at <= now()at most once perpurge_interval_seconds(default 900) per process, guarded bypg_try_advisory_lockon Postgres (skip the guard on the sqlite test variant) so concurrent workers don't stampede. No external cron. Precedents:FSCacheAdapter.cache.expire()on init;session_lifecyclecomputing "abandoned" at read time. - KV behavior: string KV values are exact-key values with optional TTL support where the adapter exposes it. No graph-to-session checkpoint keys are written anymore.
4.6 String KV methods
The old hidden async_redis duck-type contract is gone with graph-to-session sync. Implement the formal CacheDBInterface string KV methods directly:
async def get_value(self, key: str) -> str | None: ...
async def set_value(self, key: str, value: str, ttl: int | None = None) -> None: ...
async def delete_value(self, key: str) -> None: ...
get_value returns str | None. set_value should upsert by key and set expires_at when a TTL is supplied. delete_value should be idempotent.
5. Config & provider selection changes
cognee/infrastructure/databases/cache/config.py:
cache_backend: Literal["redis", "fs", "tapes", "postgres"] = "fs"(default unchanged).- New fields:
cache_db_url: Optional[str] = None(envCACHE_DB_URL, e.g.postgresql+asyncpg://cognee:cognee@localhost:5432/cognee_db) andcache_purge_interval_seconds: int = 900. Single-URL chosen over five discreteCACHE_DB_*fields: one hashable kwarg through the lru_cache'd factory, one config field instead of five, and it matches the adapter'sconnection_stringconstructor. Do not reusecache_host/cache_port— they default to Redis'slocalhost:6379(silent misdirection risk). - Fallback resolution (helper
_resolve_cache_db_urlinget_cache_engine.py, mirroringget_graph_engine.py's GRAPH_DATABASE_* → DB_* fallback):CACHE_DB_URLif set; else, whenget_relational_config().db_provider == "postgres", buildpostgresql+asyncpg://DB_USERNAME:DB_PASSWORD@DB_HOST:DB_PORT/DB_NAMEwith a warning-level log; else raiseCacheConnectionError("CACHE_BACKEND=postgres requires CACHE_DB_URL or DB_PROVIDER=postgres"). - Extend
to_dict()and the class docstring (both pinned bycognee/tests/unit/infrastructure/databases/cache/test_cache_config.py::test_cache_config_defaults/test_cache_config_to_dict, which assert the exact full dict — update them).
cognee/infrastructure/databases/cache/get_cache_engine.py:
- Thread two new hashable params through both signatures (same pattern as the
tapes_*params):create_cache_engine(..., cache_db_url: str | None = None, cache_purge_interval_seconds: int = 900); pass fromget_cache_engine(). - New branch with lazy import (verified pattern:
TapesCacheAdapteris imported inside its branch):
elif config.cache_backend == "postgres":
from cognee.infrastructure.databases.cache.postgres.PostgresCacheAdapter import (
PostgresCacheAdapter,
)
return PostgresCacheAdapter(
connection_string=_resolve_cache_db_url(cache_db_url),
lock_key=lock_key,
log_key=log_key,
session_ttl_seconds=session_ttl_seconds,
agentic_lock_expire=agentic_lock_expire,
agentic_lock_timeout=agentic_lock_timeout,
purge_interval_seconds=cache_purge_interval_seconds,
)
- Update the
ValueErrormessage to"'redis', 'fs', 'tapes', 'postgres'". - Move the currently-unconditional
from ...RedisAdapter import RedisAdapter(verified: executes for ALL backends once caching is on) inside the== "redis"branch — 2-line cleanup that removes the accidental reliance onfakeredispulling inredis. Low-risk, do it in this PR. - lru_cache caveats respected:
cache_db_urlis a hashable string; constructor params frozen after first call untilcache_clear()(existing behavior); note the existing pool-per-lock_keyinstantiation pattern (Ladybug per-db lock_key) in the adapter docstring.
pyproject.toml: no changes — SQLAlchemy is core; asyncpg/psycopg2 are in the existing postgres extra. Document "CACHE_BACKEND=postgres requires cognee[postgres]".
6. Schema creation / migration approach
Graph-adapter style, not alembic:
- Tables on a private
MetaData()incache/postgres/tables.py; alembic'senv.py(target_metadata = Base.metadata) never sees them; no migration files. - Idempotent
create_all(checkfirst=True)inside_ensure_initialized()(once-flag +asyncio.Lock), re-run lazily afterclose_cache_engine()/prune_system'scache_clear()recreates the adapter — necessary because the cache factory has no_GraphEngineHandleequivalent. - Rationale: cache content is ephemeral/TTL'd, schema is additive, and the repo precedent is explicit (
graph_node/graph_edgecreate-on-init vs alembic-managedsession_records). - No per-dataset database handler — do not register anything in
supported_dataset_database_handlers.pyor copyPostgresGraphDatasetDatabaseHandler. Sessions are per-user, not per-dataset; isolation parity comes from(user_id, session_id)columns. Deployments wanting physical separation pointCACHE_DB_URLat a dedicated database.
7. Session lock handling
cognee/infrastructure/locks/session_lock.py: no change. It is pure in-process asyncio with no Redis dependency; its documented single-worker scope is unchanged. Note in the PR description that the adapter'sFOR UPDATEinupdate_qa_entry/delete_feedbackmakes those RMW paths multi-worker-safe at the storage layer anyway (closes the gap session_lock.py's own docstring flags).CacheDBInterface.acquire_lock/release_lock(shared Ladybug lock):- v1 (core PR): raise
SharedLadybugLockRequiresRedisError— established FS precedent;SHARED_LADYBUG_LOCK=true + CACHE_BACKEND=postgresstays unsupported with a clear error. - Phase 6 (separate PR): lazy-create a small sync engine (
postgresql+psycopg2,pool_size=2; psycopg2 is already in thepostgresextra) used only for locks.lock_id = int.from_bytes(sha256(self.lock_key.encode()).digest()[:8], "big", signed=True).acquire_lock(): check out a dedicated connection, loopSELECT pg_try_advisory_lock(:id)with 0.1 s sleeps until theagentic_lock_timeoutdeadline; on success return a handle object owning the connection; on timeout raiseRuntimeError(mirrors Redis).release_lock(lock=None):pg_advisory_unlockon the passed handle, notself.lock(explicitly tested Redis behavior); close/return the connection; swallow errors if not held. Handle-bound connections givethread_local=Falseparity (releasable from worker threads). Semantics note to document: advisory locks release on connection death (safer than, but different from, Redis's 240 sagentic_lock_expireauto-expiry, which has no direct equivalent).
- v1 (core PR): raise
- Multi-worker
improve()mutex viapg_try_advisory_lockbehindtry_acquire_improve_lock: explicit non-goal / future work.
8. Test plan
Unit CRUD — new cognee/tests/unit/infrastructure/databases/cache/test_postgres_adapter_crud.py, mirroring test_redis_adapter_crud.py's ~25 cases. No real Postgres in pytest (repo convention — Redis suites use a hand-rolled _InMemoryRedisList fake; here we get a real-engine equivalent for free): construct PostgresCacheAdapter("sqlite+aiosqlite:///<tmp>/cache.db") — the with_variant(JSON(), "sqlite") columns and identity-PK→autoincrement degradation make the same code paths run without a server. Cover:
- QA create/get-latest/get-all ordering; uuid4
qa_idfallback;[]on empty includinglast_n=1(pins the FS-style choice) - update: None-preserves every field;
memify_metadataMERGE;SessionQAEntryValidationErrorpropagates unwrapped;Falseon missing qa_id delete_feedbacknulls both fields;delete_qa_entryrowcount semantics + TTL survival;delete_sessionclears both tables- traces: append/get/count/feedback delegation; sanitization via model validators; blank
trace_id→CacheConnectionError - TTL:
expires_atset/refreshed on create/update/delete_feedback/after-delete; NOT refreshed on reads; disabled whensession_ttl_seconds in (0, None); expired rows invisible to reads (assert via direct SQL with backdatedexpires_at) prune()clears only the four cache tables;close()idempotent; legacyadd_qa/get_latest_qa/get_all_qasshimsacquire_lock/release_lockraiseSharedLadybugLockRequiresRedisError- KV
get_value/set_value/delete_valueround-trips, including one legacygraph_knowledge:{u}:{s}deletion path throughSessionManager.delete_session
Integration — add "postgres" to the params=["fs", "redis"] fixtures in cognee/tests/integration/infrastructure/session/test_session_sdk_integration.py, test_session_persistence_memify_integration.py, test_feedback_weights_memify_integration.py (aiosqlite URL, no server); new test_session_manager_postgres.py mirroring test_session_manager_redis.py (LLM mocked via AsyncMock on LLMGateway.acreate_structured_output).
Config/factory — update test_cache_config.py exact-dict assertions; new factory tests: CACHE_BACKEND=postgres returns PostgresCacheAdapter, unknown backend ValueError lists all four, URL fallback resolution (CACHE_DB_URL → DB_* → error).
CI e2e (real Postgres) — third twin job run_conversation_sessions_test_postgres in .github/workflows/e2e_tests.yml, copying run_conversation_sessions_test_redis minus the redis service (the pgvector/pg17 service is already provisioned there): env CACHING=true AUTO_FEEDBACK=true CACHE_BACKEND=postgres DB_PROVIDER=postgres (relying on the DB_* fallback, which also exercises it) or explicit CACHE_DB_URL; extras "postgres" only; runs cognee/tests/test_conversation_history.py unmodified. Phase 6 adds a SHARED_LADYBUG_LOCK=true CACHE_BACKEND=postgres variant of the concurrent-subprocess job running test_concurrent_subprocess_access.py, plus real-Postgres-gated advisory-lock unit tests (acquire/release of the passed handle, contention timeout → RuntimeError, cross-connection mutual exclusion).
Regression coverage — FS/Redis/Postgres tests proving generic string KV round-trips and the legacy graph_knowledge: cleanup path work consistently.
9. Rollout & backward compatibility
- No behavior change by default:
cache_backenddefault stays"fs"; redis/tapes branches untouched;get_cache_engine()still returnsNonewhen caching + usage_logging are off (every consumer already handlesNone). - No data migration tooling: session cache is 7-day-TTL ephemeral; switching backends starts fresh — same story as redis↔fs today. One-liner in docs.
SessionRecordrows in the relational DB are unaffected. forget(everything=True)/prune_system/prune_datapaths work unchanged (prune(),close_cache_engine(),cache_clear()); document that Postgresprune()is scoped to cognee cache tables (an improvement overFLUSHDB).- Failure modes: misconfigured backend raises
CacheConnectionError(503) at first use;SHARED_LADYBUG_LOCK=true+ postgres raisesSharedLadybugLockRequiresRedisErrorwith a clear message until Phase 6. - Docs:
.env.template"Session cache settings" block (~line 375): add# CACHE_BACKEND=postgres+# CACHE_DB_URL=postgresql+asyncpg://...;config.pydocstring backend list; factoryValueErrormessage;CLAUDE.mdextras note ("postgres — also enables the Postgres session-cache backend"); file a task for the external docs.cognee.ai "sessions-and-caching" page.docker-compose.ymlneeds nothing (postgres service exists; redis stays behind its profile).
10. Phased work breakdown
| Phase | Scope | Size |
|---|---|---|
| P1 — Adapter core | cache/postgres/tables.py + PostgresCacheAdapter.py: engine, _ensure_initialized, QA CRUD (FOR UPDATE updates, atomic deletes), TTL refresh + read filter + scoped/throttled purge, prune, close |
L (~1.5–2 days, ~450 LOC) |
| P2 — Traces, usage logs, locks-raise, string KV | append_agent_trace_step/trace reads, log_usage/get_usage_logs (whole-list TTL refresh), lock methods raising SharedLadybugLockRequiresRedisError, get_value/set_value/delete_value |
M (~0.5–1 day, ~150 LOC) |
| P3 — Wiring | config.py Literal + cache_db_url + cache_purge_interval_seconds + to_dict + docstring; get_cache_engine.py branch + _resolve_cache_db_url fallback + error msg + RedisAdapter import moved into its branch; config-test fixes |
S (~0.5 day) |
| P4 — Tests | Unit CRUD suite (aiosqlite), test_session_manager_postgres.py, "postgres" params in 3 integration files, factory/config tests |
M (~1 day, ~500 LOC tests) |
| P5 — CI + docs | run_conversation_sessions_test_postgres e2e twin job, .env.template, CLAUDE.md, docs.cognee.ai task |
S (~0.5 day) |
| P6 — Follow-up PR: advisory locks | psycopg2 pg_try_advisory_lock sync lock impl lifting the SHARED_LADYBUG_LOCK restriction + concurrent-subprocess CI variant |
M (~1 day) |
Core (P1–P5): ~4 days, one reviewable PR (or two: adapter+tests, then wiring+CI). P6 ships separately.
11. Future: Turbopuffer adapter sketch (do not build now — verify the seam)
Turbopuffer is a namespaced object/vector store with upsert-by-id, delete-by-id, and attribute-filtered queries. The same CacheDBInterface maps because every cognee cache access is by exact composite key, append-or-point-update, with no cross-key scans:
- Namespaces: today's key strings verbatim (
agent_sessions:{user_id}:{session_id},agent_traces:{...},usage_logs:{user_id}) — key-prefix tenancy carries over unchanged. - Rows: id =
qa_id/ trace uuid; attributes = entry fields plus a client-stamped monotonicseq/created_atfor ordering (no serial column); vector optional (zero/1-dim placeholder if mandatory — or embedquestion+answerfor free semantic recall later, a genuine upside). - Operations: create/append → upsert;
get_latest/get_all→ seq-ordered query (tail = desc + limit + reverse);update_qa_entry/delete_feedback→ read-merge-upsert by id (back to Redis-grade RMW races — noFOR UPDATE; acceptable for a cache tier, flag in docs);delete_session→ namespace delete;prune()→ enumerate + deletecognee-*namespaces by prefix; KV → acache_kvnamespace with key-as-id rows; TTL →expires_atattribute + query filter + periodic GC (no native TTL — the Postgres TTL design transfers directly); locks → raiseSharedLadybugLockRequiresRedisError(precedent). - Seam requirements this plan satisfies: (1) all storage details stay behind
CacheDBInterface; no caller touches adapter internals; (2) entries cross the boundary only as pydanticmodel_dump()payloads; (3) ordering is adapter-internal (seqcolumn vsseqattribute) — nothing leaks; do not let SQL row ids escape into return values during P2; (4) wiring is mechanical:cache_backendLiteral +="turbopuffer", lazy-import elif,TURBOPUFFER_API_KEY/TURBOPUFFER_REGIONthreaded through the lru_cache'd factory.
12. Open questions
- Default fallback to the relational DB: when
CACHE_DB_URLis unset andDB_PROVIDER=postgres, the cache silently shares the relational database (distinctcache_*tables, warning logged). Is co-tenancy acceptable as the default, or shouldCACHE_DB_URLbe mandatory? - Normalize the Redis
None-on-last_n==1quirk inRedisAdapteritself (return[]like FS/Postgres), or leave Redis as-is and only document Postgres's[]choice? (Some tests pin the Redis behavior.) - Hot-path payload size: should
get_latest_qa_entrieseventually projectpayload - 'context'for history reads (formatted history usesinclude_context=Falsebut still fetches full entries)? Behavior-preserving today; revisit with real latency data. - Throttled global purge tuning: is 900 s / per-process advisory-lock-guarded sweep enough for high-volume deployments, or is a documented external cron (
DELETE ... WHERE expires_at <= now()) preferable at scale? fakeredis[lua]core dependency: verified unused in-repo and only load-bearing because it transitively installsredisfor the unconditional import this plan removes — candidate for deletion in a separate cleanup PR (needs a check that no downstream consumers rely on it).- Lock auto-expiry parity (Phase 6): Redis locks auto-expire after
agentic_lock_expire=240s; pg advisory locks hold until connection death. Is connection-scoped release acceptable, or do we need a watchdog that closes the lock connection after 240 s?
Source: SESSION_POSTGRES_CACHE_PLAN.md