store-owned: finish Protocol-B removal (PG-free zero-batch tracking)

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

@@ -1395,8 +1395,19 @@ async def _process_one_llm_batch(llm_batch_local: list[dict[str, Any]], batch_nu
             # they are durable-but-invisible in the external store while this batch runs its LLM work. The
             # witness row + decide happen in ONE short transaction at the end (below) — we must not
             # hold a Postgres transaction across the LLM calls in the sub-batch loop.
+            #
+            # A store that owns the whole retain (store_owned_retain) keeps ALL of this batch's memory
+            # writes — observation upserts/deletes and the mark_consolidated stamps — in ITS store, not
+            # Postgres, so there is nothing to make atomic with a Postgres witness. Skip the write-group
+            # entirely (``_batch_txn = None`` → the writes below are plain, immediately-visible writes).
+            # This is also why consolidation was the source of the undecided write-group txns that stall
+            # the store's indexer: mint-early / witness-late meant a crash or a sibling-cancel between
+            # mint and decide left a pending txn with no witness. With no txn there is nothing to leave
+            # undecided. The mental-model refresh-tag bookkeeping below becomes a plain Postgres write
+            # (best-effort rather than atomic-with-the-batch — a missed tag only defers a refresh).
             _txn_provider = get_memories()
-            _batch_txn = await _txn_provider.mint_txn(bank_id=bank_id, mutating=True)
+            _store_owned = _txn_provider.store_owned_retain_for(bank_id)
+            _batch_txn = None if _store_owned else await _txn_provider.mint_txn(bank_id=bank_id, mutating=True)
 
             try:
                 pending: list[list[dict[str, Any]]] = [llm_batch_local]
@@ -1511,11 +1522,13 @@ async def _process_one_llm_batch(llm_batch_local: list[dict[str, Any]], batch_nu
                             txn=_batch_txn,
                         )
                     async with conn.transaction():
-                        await _txn_provider.write_txn_witness(_batch_txn, conn=conn, fq_table=fq_table)
+                        if _batch_txn is not None:
+                            await _txn_provider.write_txn_witness(_batch_txn, conn=conn, fq_table=fq_table)
                         # Persist this batch's mental-model refresh tags atomically with the
                         # witness, so they share the batch's fate: durable iff the batch is
                         # (#3411). Only the succeeded source facts — the ones just marked
-                        # consolidated — contribute a tag.
+                        # consolidated — contribute a tag. (Store-owned: no witness, so this is a
+                        # plain best-effort write; a missed tag only defers a mental-model refresh.)
                         if operation_id and succeeded_ids:
                             succeeded_set = {str(mem_id) for mem_id in succeeded_ids}
                             batch_tags = sorted(
@@ -1535,18 +1548,23 @@ async def _process_one_llm_batch(llm_batch_local: list[dict[str, Any]], batch_nu
                 # task mid-batch instead of letting it run to completion. Kept OUTSIDE the
                 # decide(commit=True) below on purpose: once the witness has committed, the
                 # batch's fate is decided and an abort here would discard durable writes.
-                try:
-                    await _txn_provider.decide_txn(_batch_txn, commit=False)
-                except Exception:
-                    logger.warning(
-                        f"[CONSOLIDATION] bank={bank_id} failed to abort write-group for"
-                        f" llm_batch #{batch_num_local}; recovery sweep will resolve it",
-                        exc_info=True,
-                    )
+                # Store-owned batches hold no write-group (writes were plain and are already
+                # durable/visible); there is nothing to abort — consolidation is idempotent on retry.
+                if _batch_txn is not None:
+                    try:
+                        await _txn_provider.decide_txn(_batch_txn, commit=False)
+                    except Exception:
+                        logger.warning(
+                            f"[CONSOLIDATION] bank={bank_id} failed to abort write-group for"
+                            f" llm_batch #{batch_num_local}; recovery sweep will resolve it",
+                            exc_info=True,
+                        )
                 raise
             # Postgres committed the witness: publish the batch's write-group. On a crash before
-            # here the writes stay invisible and the recovery sweep resolves them (spec §5).
-            await _txn_provider.decide_txn(_batch_txn, commit=True)
+            # here the writes stay invisible and the recovery sweep resolves them (spec §5). No-op for
+            # a store-owned batch (no write-group; its writes were already visible).
+            if _batch_txn is not None:
+                await _txn_provider.decide_txn(_batch_txn, commit=True)
 
             cancelled_local = False
             if operation_id and not await memory_engine._check_op_alive(operation_id):

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

@@ -1246,6 +1246,7 @@ async def record_unit_entity_postings(
         self,
         unit_entity_pairs: list[tuple[str, str]] | list[tuple[str, str, datetime | None]],
         bank_id: str | None = None,
+        store_write: bool = True,
     ):
         """Store-owned variant of :meth:`link_units_to_entities_batch` that touches NO
         Postgres connection.
@@ -1257,6 +1258,13 @@ async def record_unit_entity_postings(
         its connection-free store phase and never hold the data-plane connection across the
         object-store write. NOT for the Postgres store, whose posting is a real ``unit_entities``
         INSERT that requires the connection.
+
+        ``store_write=False`` skips the store-side posting and does ONLY the co-occurrence
+        accumulation. The caller uses this when it has already attached entity ids to the memories
+        as part of the same write (a single deferred write with entities inline, instead of
+        write-then-reattach) — so the store row is already correct and a second store write would
+        be redundant. Co-occurrence still runs: it references only ``entities`` and is needed by the
+        entity-graph endpoint and resolution's disambiguation signal regardless of who wrote the row.
         """
         if not unit_entity_pairs:
             return
@@ -1265,10 +1273,16 @@ async def record_unit_entity_postings(
             (t[0], t[1], t[2] if len(t) >= 3 else None)  # type: ignore[misc]
             for t in unit_entity_pairs
         ]
-        return await self._link_units_to_entities_batch_impl(None, normalized, bank_id)
+        return await self._link_units_to_entities_batch_impl(
+            None, normalized, bank_id, store_write=store_write
+        )
 
     async def _link_units_to_entities_batch_impl(
-        self, conn, unit_entity_pairs: list[tuple[str, str, datetime | None]], bank_id: str | None = None
+        self,
+        conn,
+        unit_entity_pairs: list[tuple[str, str, datetime | None]],
+        bank_id: str | None = None,
+        store_write: bool = True,
     ):
         # Sorted bulk insert to prevent deadlocks from inconsistent lock ordering
         # across concurrent transactions on the unit_entities unique index.
@@ -1280,16 +1294,19 @@ async def _link_units_to_entities_batch_impl(
         # memories store records it. Co-occurrence below is separate and unaffected:
         # it references only `entities`, which stays in Postgres either way, and is
         # read by the entity-graph endpoint and by resolution's disambiguation signal.
+        # `store_write=False` means the caller already wrote the postings inline with the
+        # memories, so we skip the (redundant) second store write and keep only co-occurrence.
         from .memories import get_memories
 
-        await get_memories().record_unit_entities(
-            conn=conn,
-            ops=self._ops,
-            fq_table=fq_table,
-            bank_id=bank_id,
-            unit_ids=unit_ids,
-            entity_ids=entity_ids,
-        )
+        if store_write:
+            await get_memories().record_unit_entities(
+                conn=conn,
+                ops=self._ops,
+                fq_table=fq_table,
+                bank_id=bank_id,
+                unit_ids=unit_ids,
+                entity_ids=entity_ids,
+            )
 
         # Build maps keyed by unit_id:
         #   unit_to_entities: entity set per unit (for the co-occurrence cross-product)

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

@@ -453,6 +453,40 @@ def name(self) -> str:
     #: the inline SQL. Cold, never-searched, key-based — see docs/documents-chunks.md.
     owns_document_store: bool = False
 
+    #: Whether this store commits the ENTIRE retain — entity resolution/minting, the memory upserts,
+    #: and the document replace — in ONE atomic server-side call (its ``retain`` method). Default
+    #: False: the host runs its normal retain (Phase-1 entity resolution in SQL, then the write). A
+    #: store that sets this True resolves entity NAMES itself and needs no Postgres connection phase,
+    #: so the orchestrator skips SQL entity resolution + the documents/chunks/entities rows + the
+    #: commit witness and issues one ``retain`` instead. Bank-scoped via
+    #: :meth:`store_owned_retain_for`.
+    store_owned_retain: bool = False
+
+    def store_owned_retain_for(self, bank_id: str) -> bool:
+        """Per-bank form of :attr:`store_owned_retain`. Defaults to the class attribute; a router
+        that keeps some banks in a store-owned backend and others in SQL overrides it to answer PER
+        BANK. See :meth:`owns_document_store_for`."""
+        return self.store_owned_retain
+
+    async def retain(
+        self,
+        bank_id: str,
+        unit_ids: list[str],
+        facts: list,
+        *,
+        document_id: str | None = None,
+        unit_entity_names: dict[str, list[str]] | None = None,
+        replace_document_id: str = "",
+        resolve_threshold: float = 0.0,
+    ):
+        """Commit an entire retain in one server-side call — resolve/mint the ``unit_entity_names``
+        against the store's own registry, write the memories with the resulting entity ids, and
+        (when ``replace_document_id`` is set) tombstone the document's prior version — all atomically.
+        Only a store advertising :attr:`store_owned_retain` implements this; the orchestrator calls it
+        exactly when :meth:`store_owned_retain_for` is true, so the default never runs. It exists on
+        the interface so a routing extension delegates it automatically (see RoutingMemories)."""
+        raise NotImplementedError("this store does not support a store-owned retain")
+
     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 keeps different banks in different
@@ -661,6 +695,34 @@ async def get_document_record(self, *, bank_id: str, document_id: str, include_t
         """A document's metadata (and, if asked, its extracted ``original_text``), or ``None``."""
         raise NotImplementedError
 
+    async def list_documents(
+        self,
+        *,
+        bank_id: str,
+        search_query: "str | None" = None,
+        limit: int = 100,
+        offset: int = 0,
+    ) -> dict:
+        """Page this bank's documents from the store's OWN registry — the ``{items, total, limit,
+        offset}`` shape the documents browser expects. Only a store that owns its document metadata
+        overrides this (a Postgres-backed store lists from the SQL ``documents`` table instead, so
+        the engine only calls this for an ``owns_document_store`` store). Default raises so a
+        mis-routed call is loud rather than silently empty."""
+        raise NotImplementedError
+
+    async def count_documents(self, *, bank_id: str) -> int:
+        """This bank's document count, from the store's own registry — the bank-stats document
+        total. Only an ``owns_document_store`` store overrides this (a Postgres store counts the
+        SQL ``documents`` table instead); the engine only calls it for a store that owns its docs."""
+        raise NotImplementedError
+
+    async def get_entity_graph(self, *, bank_id: str, limit: int = 1000, min_count: int = 1) -> dict:
+        """The entity co-occurrence graph (``{nodes, edges, ...}``) from the store's OWN aggregate.
+        Only a store that owns its entities overrides this (a Postgres store reads its
+        ``entity_cooccurrences`` table); the engine calls it only for a store-owned bank, whose SQL
+        table is empty."""
+        raise NotImplementedError
+
     async def get_chunk_text(self, *, bank_id: str, document_id: str, chunk_index: int) -> "str | None":
         """One chunk's text by position, or ``None`` if the document/index does not exist."""
         raise NotImplementedError
@@ -1112,6 +1174,7 @@ async def apply_edit(
         event_date,
         mentioned_at,
         entity_ids: list[str] | None,
+        entity_names: list[str] | None = None,
         txn=None,
     ) -> None:
         """Apply a curation field edit to a live memory.
@@ -1122,10 +1185,20 @@ async def apply_edit(
         embedding is *not* written here — the caller re-embeds from the new fields
         and calls :meth:`set_memory_embedding` after.
 
-        ``entity_ids`` is the resolved entity set the memory should now carry; a
-        store that keeps them on the memory writes them here, one that keeps them
-        in a join table has already re-linked them and ignores this. ``None`` means
-        the entity set was not part of this edit.
+        The new entity set for the memory is supplied one of two ways, and a store
+        uses whichever fits how it keeps its registry:
+
+        * ``entity_names`` — the raw names the edit resolved to. A store that owns
+          its entity registry resolves + mints these against its OWN registry
+          (exactly as its :meth:`retain` does) and rewrites the memory's entity
+          ids from the result, so a brand-new entity created by an edit lands in
+          that registry. When it is not ``None`` it is the authoritative set and
+          ``entity_ids`` is ignored.
+        * ``entity_ids`` — the already-resolved set, for a store whose registry is
+          the host's SQL (the host minted them and, for a join-table store, has
+          already re-linked them, so it ignores this).
+
+        Both ``None`` means the entity set was not part of this edit.
         """
         raise NotImplementedError
 

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

@@ -535,6 +535,7 @@ async def apply_edit(
         event_date,
         mentioned_at,
         entity_ids: list[str] | None,
+        entity_names: list[str] | None = None,  # noqa: ARG002 — this store's registry is SQL; the host already minted+linked, so entity_ids is authoritative.
         txn=None,
     ) -> None:
         await writes.apply_edit(