feat(memories): per-bank store capabilities

hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py

@@ -284,7 +284,7 @@ async def _dedup_reconcile_create(
     # twin's existing embedding (the merged text is >= threshold similar, so it stays
     # representative and avoids a re-embed + a dialect-specific vector UPDATE).
     store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
         # Oracle-safe: _native_search_vector_update emits the to_tsvector clause only for a native
         # PG tsvector column, "" otherwise (see #3021 — the raw ::regconfig cast breaks Oracle).
         search_vector_clause = _native_search_vector_update(config, "$1")
@@ -349,7 +349,7 @@ async def _dedup_reconcile_update(
     # guarantees twin and updated share scope, so dropping the updated row's tags loses no
     # visibility. Temporal fields follow the surviving twin (minimal scope; matches create).
     store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
         # Oracle-safe search_vector clause (#3021): "" unless a native PG tsvector column.
         search_vector_clause = _native_search_vector_update(config, "$1")
         await conn.execute(
@@ -510,7 +510,7 @@ async def _filter_live_source_memories(
     if not source_memory_ids:
         return []
     store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
         rows = await conn.fetch(
             f"SELECT id FROM {fq_table('memory_units')} WHERE id = ANY($1::uuid[]) AND bank_id = $2 FOR SHARE",
             source_memory_ids,
@@ -632,7 +632,7 @@ async def _count_observations_for_scope(
     Observations with no tags are not counted (the limit does not apply to them).
     """
     store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
         return await conn.fetchval(
             f"SELECT COUNT(*) FROM {fq_table('memory_units')} "
             f"WHERE bank_id = $1 AND fact_type = 'observation' AND tags @> $2::varchar[]",
@@ -2092,7 +2092,7 @@ async def _execute_update_action(
 
     t0 = time.time()
     store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
         await conn.execute(
             f"""
             UPDATE {fq_table("memory_units")}
@@ -2227,7 +2227,7 @@ async def _execute_delete_action(
 ) -> None:
     """Delete a superseded or contradicted observation."""
     store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
         await conn.execute(
             f"DELETE FROM {fq_table('memory_units')} WHERE id = $1 AND bank_id = $2 AND fact_type = 'observation'",
             uuid.UUID(observation_id),
@@ -2597,7 +2597,7 @@ async def _create_observation_directly(
     # search_vector the configured backend needs); a store that owns its rows takes it through
     # upsert_observation as a normal Observation-type memory carrying all of its own state.
     store = get_memories()
-    if store.writes_memory_rows_in_sql:
+    if store.writes_memory_rows_in_sql_for(bank_id):
         # Query varies based on text search backend.
         from ..schema import _is_oracle  # noqa: PLC0415
 

hindsight-api-slim/hindsight_api/engine/maintenance.py

@@ -132,7 +132,12 @@ def _any_job_enabled() -> bool:
     @staticmethod
     def _cross_store_recovery_enabled() -> bool:
         """True when the memories store keeps memories outside SQL and therefore has
-        cross-store write-group txns a crashed writer could leave undecided."""
+        cross-store write-group txns a crashed writer could leave undecided.
+
+        Deliberately reads the PROCESS-LEVEL class attribute, not the per-bank
+        ``writes_memory_rows_in_sql_for(bank_id)`` — this only decides whether the recovery LOOP
+        needs to run at all. A store that routes some banks outside SQL keeps the class attribute
+        False so the loop runs, then ``recover_pending_txns`` is bank-scoped inside it."""
         try:
             from .memories import get_memories
 

hindsight-api-slim/hindsight_api/engine/memories/base.py

@@ -394,6 +394,20 @@ def name(self) -> str:
     #: the inline SQL. Cold, never-searched, key-based — see docs/documents-chunks.md.
     owns_document_store: bool = False
 
+    def writes_memory_rows_in_sql_for(self, bank_id: str) -> bool:
+        """Per-bank form of :attr:`writes_memory_rows_in_sql`. Defaults to the class attribute, so a
+        single-store extension needs no override. A store that routes different banks to different
+        backends (some in SQL, some not) overrides this to answer PER BANK; every *bank-scoped* call
+        site consults this instead of the class attribute, so mixed banks each take the correct path.
+        (The few process-level gates — e.g. "is cross-store txn recovery relevant at all" — keep
+        reading the class attribute.)"""
+        return self.writes_memory_rows_in_sql
+
+    def owns_document_store_for(self, bank_id: str) -> bool:
+        """Per-bank form of :attr:`owns_document_store`. Defaults to the class attribute; a
+        bank-routing store overrides it. See :meth:`writes_memory_rows_in_sql_for`."""
+        return self.owns_document_store
+
     # ------------------------------------------------------------------ lifecycle
 
     async def initialize(self) -> None:

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -5296,7 +5296,7 @@ def to_tuple_format(results):
                 from .memories import get_memories
 
                 _obs_store = get_memories()
-                if observation_ids_ordered and not _obs_store.writes_memory_rows_in_sql:
+                if observation_ids_ordered and not _obs_store.writes_memory_rows_in_sql_for(bank_id):
                     # A store that keeps memories outside SQL: fetch each observation, then its
                     # source memories, for their chunk_ids — the join the SQL branch does, walked
                     # in observation-rank order so per-observation grouping is preserved.
@@ -5378,7 +5378,7 @@ def to_tuple_format(results):
                     # row, so it selects one fewer column and keeps the asyncpg Records as-is — no
                     # per-chunk ``dict`` allocation for an overlay it never runs.
                     _chunk_store = get_memories()
-                    _owns_docs = _chunk_store.owns_document_store
+                    _owns_docs = _chunk_store.owns_document_store_for(bank_id)
                     if _owns_docs:
                         _chunk_cols = "chunk_id, chunk_text, chunk_index, document_id"
                     else:
@@ -5575,7 +5575,7 @@ def _source_fact_dict(
                         # Resolve each observation's sources. This is a recall hot path, so the SQL
                         # store reads only the two columns it needs rather than a full memory row; a
                         # store that owns its rows answers from its own objects via one addressed read.
-                        if store.writes_memory_rows_in_sql:
+                        if store.writes_memory_rows_in_sql_for(bank_id):
                             obs_rows = [
                                 {"id": str(r["id"]), "source_memory_ids": r["source_memory_ids"]}
                                 for r in await sf_conn.fetch(
@@ -5612,7 +5612,7 @@ def _source_fact_dict(
                         # needed, so the SQL store selects those (bank-scoped) instead of the full
                         # 17-column memory row — the difference is measurable on this hot path.
                         if source_ids_ordered:
-                            if store.writes_memory_rows_in_sql:
+                            if store.writes_memory_rows_in_sql_for(bank_id):
                                 source_row_by_id = {
                                     str(r["id"]): _source_fact_dict(
                                         uid=str(r["id"]),
@@ -5985,7 +5985,7 @@ async def get_document(
             from .memories import get_memories
 
             _store = get_memories()
-            if _store.writes_memory_rows_in_sql:
+            if _store.writes_memory_rows_in_sql_for(bank_id):
                 # Use a subquery for counts to avoid GROUP BY on CLOB columns
                 # (Oracle cannot use CLOB types as comparison keys in GROUP BY).
                 doc = await conn.fetchrow(
@@ -6047,7 +6047,7 @@ async def get_document(
                     # A store that owns the document store keeps the extracted text in
                     # its own store, not in documents.original_text (which is NULL here). Overlay
                     # it from the store so get_document still returns the body.
-                    if _store.owns_document_store:
+                    if _store.owns_document_store_for(bank_id):
                         _rec = await _store.get_document_record(
                             bank_id=bank_id, document_id=document_id, include_text=True
                         )
@@ -6124,7 +6124,7 @@ async def delete_document(
                 from .memories import get_memories
 
                 _store = get_memories()
-                if _store.writes_memory_rows_in_sql:
+                if _store.writes_memory_rows_in_sql_for(bank_id):
                     unit_rows = await conn.fetch(
                         f"SELECT id FROM {fq_table('memory_units')} WHERE document_id = $1 AND fact_type IN ('experience', 'world')",
                         document_id,
@@ -6170,7 +6170,7 @@ async def delete_document(
                 # cascade to its memories (they are not SQL rows) — drop them through the store,
                 # tagged with a write-group so the store tombstone commits atomically with the
                 # Postgres document delete (a rolled-back delete must not orphan the memories).
-                if deleted and not _store.writes_memory_rows_in_sql:
+                if deleted and not _store.writes_memory_rows_in_sql_for(bank_id):
                     _del_txn = await _store.begin_txn(conn=conn, fq_table=fq_table, bank_id=bank_id, mutating=True)
                     await _store.delete_document(
                         conn=conn, fq_table=fq_table, bank_id=bank_id, document_id=document_id, txn=_del_txn
@@ -6179,7 +6179,7 @@ async def delete_document(
                     # extracted text + chunk bodies; the orphan sweep reclaims the blobs), under the
                     # same write-group so it commits atomically with the Postgres document delete.
                     # This is the EXPLICIT deletion — distinct from the re-ingest facts-delete above.
-                    if _store.owns_document_store:
+                    if _store.owns_document_store_for(bank_id):
                         await _store.delete_document_record(bank_id=bank_id, document_id=document_id, txn=_del_txn)
 
                 # Invalidate observations referencing these (now-deleted) memories
@@ -6290,7 +6290,7 @@ async def update_document(
                     from .memories import MemoryPatch, get_memories
 
                     _store = get_memories()
-                if tags is not None and not _store.writes_memory_rows_in_sql:
+                if tags is not None and not _store.writes_memory_rows_in_sql_for(bank_id):
                     # A store that keeps memories outside SQL: retag the document's memories, then
                     # invalidate the observations built on them and requeue their sources so the
                     # next consolidation rebuilds them under the new tags (the cascade the SQL
@@ -6464,7 +6464,7 @@ async def delete_memory_unit(
                 from .memories import get_memories
 
                 _store = get_memories()
-                if _store.writes_memory_rows_in_sql:
+                if _store.writes_memory_rows_in_sql_for(bank_id):
                     row = await conn.fetchrow(
                         f"SELECT bank_id, fact_type FROM {fq_table('memory_units')} WHERE id = $1",
                         str(unit_uuid),
@@ -6493,7 +6493,7 @@ async def delete_memory_unit(
                 # observations inserted concurrently by consolidation (otherwise a
                 # racing insert committed between the sweep and the delete would
                 # leave an orphan referencing this just-deleted source memory).
-                if _store.writes_memory_rows_in_sql:
+                if _store.writes_memory_rows_in_sql_for(bank_id):
                     deleted = await conn.fetchval(
                         f"DELETE FROM {fq_table('memory_units')} WHERE id = $1 RETURNING id", unit_id
                     )
@@ -6805,7 +6805,7 @@ async def delete_bank(
                             from .memories import get_memories as _get_memories_for_scope
 
                             _scope_store = _get_memories_for_scope()
-                            if _scope_store.writes_memory_rows_in_sql:
+                            if _scope_store.writes_memory_rows_in_sql_for(bank_id):
                                 unit_id_rows = await conn.fetch(
                                     f"SELECT id FROM {fq_table('memory_units')} WHERE bank_id = $1 AND fact_type = $2",
                                     bank_id,
@@ -6947,7 +6947,7 @@ async def delete_bank(
         from .memories import DeletePredicate, get_memories
 
         store = get_memories()
-        if not store.writes_memory_rows_in_sql:
+        if not store.writes_memory_rows_in_sql_for(bank_id):
             if fact_type:
                 await store.delete_where(bank_id, DeletePredicate(fact_types=[fact_type]))
             else:
@@ -6997,7 +6997,7 @@ async def clear_observations(
         backend = await self._get_backend()
         async with acquire_with_retry(backend) as conn:
             async with conn.transaction():
-                if store.writes_memory_rows_in_sql:
+                if store.writes_memory_rows_in_sql_for(bank_id):
                     # Count observations before deletion
                     count = await conn.fetchval(
                         f"SELECT COUNT(*) FROM {fq_table('memory_units')} WHERE bank_id = $1 AND fact_type = 'observation'",
@@ -7130,7 +7130,7 @@ async def retry_failed_consolidation(
         store = get_memories()
         backend = await self._get_backend()
         async with acquire_with_retry(backend) as conn:
-            if store.writes_memory_rows_in_sql:
+            if store.writes_memory_rows_in_sql_for(bank_id):
                 count = await conn.fetchval(
                     f"""
                     SELECT COUNT(*) FROM {fq_table("memory_units")}
@@ -7210,7 +7210,7 @@ async def clear_observations_for_memory(
                     from .memories import get_memories
 
                     _store = get_memories()
-                    if _store.writes_memory_rows_in_sql:
+                    if _store.writes_memory_rows_in_sql_for(bank_id):
                         await conn.execute(
                             f"""
                             UPDATE {fq_table("memory_units")}
@@ -8507,7 +8507,7 @@ async def get_chunk(
             from .memories import get_memories
 
             _store = get_memories()
-            if _store.owns_document_store:
+            if _store.owns_document_store_for(bank_id):
                 _t = await _store.get_chunk_text(
                     bank_id=chunk["bank_id"],
                     document_id=chunk["document_id"],
@@ -8598,7 +8598,7 @@ async def list_document_chunks(
             from .memories import get_memories
 
             _store = get_memories()
-            if _store.owns_document_store:
+            if _store.owns_document_store_for(bank_id):
                 _texts = await _store.list_chunk_texts(bank_id=bank_id, document_id=document_id)
                 if _texts is not None:
                     _texts_by_index = dict(enumerate(_texts))
@@ -10694,7 +10694,7 @@ async def _compute_bank_stats(self, bank_id: str) -> dict[str, Any]:
             from .memories import get_memories
 
             _store = get_memories()
-            if _store.writes_memory_rows_in_sql:
+            if _store.writes_memory_rows_in_sql_for(bank_id):
                 consolidation_row = await conn.fetchrow(
                     f"""
                     SELECT

hindsight-api-slim/hindsight_api/engine/reflect/tools.py

@@ -349,7 +349,7 @@ async def tool_expand(
     from ..memories import get_memories
 
     _store = get_memories()
-    if _store.writes_memory_rows_in_sql:
+    if _store.writes_memory_rows_in_sql_for(bank_id):
         memories = await conn.fetch(
             f"""
             SELECT id, text, chunk_id, document_id, fact_type, context

hindsight-api-slim/hindsight_api/engine/retain/bank_utils.py

@@ -464,7 +464,7 @@ async def list_banks(pool) -> list:
             last_doc = row["last_document_at"]
 
             fact_count = row["fact_count"]
-            if not _store.writes_memory_rows_in_sql:
+            if not _store.writes_memory_rows_in_sql_for(bank_id):
                 fact_count = sum(
                     (await _store.count_memories(conn=conn, fq_table=fq_table, bank_id=row["bank_id"])).values()
                 )

hindsight-api-slim/hindsight_api/engine/retain/chunk_storage.py

@@ -76,7 +76,7 @@ async def delete_chunks_by_ids(conn, chunk_ids: list[str], bank_id: str | None =
     from ..memories import META_CHUNK_ID, DeletePredicate, get_memories
 
     _store = get_memories()
-    if bank_id and not _store.writes_memory_rows_in_sql:
+    if bank_id and not _store.writes_memory_rows_in_sql_for(bank_id):
         for _cid in chunk_ids:
             await _store.delete_where(bank_id, DeletePredicate(metadata_equals={META_CHUNK_ID: _cid}), txn=txn)
 
@@ -168,7 +168,7 @@ async def store_chunks_batch(
     # same shape as store_document_text=False, and idempotency is unaffected (content_hash stays).
     from ..memories import get_memories
 
-    if get_memories().owns_document_store:
+    if get_memories().owns_document_store_for(bank_id):
         store_text = False
 
     # Prepare chunk data for batch insert

hindsight-api-slim/hindsight_api/engine/retain/fact_storage.py

@@ -371,7 +371,7 @@ async def _upsert_document_row(
     # the bulky body is written to the store up front (orchestrator._store_document_bodies).
     from ..memories import get_memories
 
-    if get_memories().owns_document_store:
+    if get_memories().owns_document_store_for(bank_id):
         original_text = None
     await conn.execute(
         f"""
@@ -411,7 +411,7 @@ async def update_memory_units_tags(
     from ..memories import MemoryPatch, get_memories
 
     store = get_memories()
-    if not store.writes_memory_rows_in_sql:
+    if not store.writes_memory_rows_in_sql_for(bank_id):
         # A store that keeps memories outside SQL: page the document's memories and patch each
         # one's tags through the store — the UPDATE below is a no-op on its empty memory_units.
         page = await store.scan_memories(

hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py

@@ -1275,7 +1275,7 @@ async def _store_document_bodies(
     from ..memories import get_memories
 
     store = get_memories()
-    if not store.owns_document_store:
+    if not store.owns_document_store_for(bank_id):
         return
     # The record's content_hash must equal what the SQL documents row stores, so a read is
     # consistent whichever it comes from: sanitize + sha256 the same combined_content. The