feat(api): attachments on recall and read surfaces for store-owned banks (#4321)

* feat(attachments): per-fact attachments for store-owned banks, from the rows the store returns

Recall, GET /memories/list and GET /memories/{id} now report per-fact
attachments for banks whose memories store owns its rows. The ids travel
on what the store already returns, so resolving them adds no store
round-trip and never reads memory_units for such a bank (that read plans
across every per-bank vector index).

- memories/base.py: META_ATTACHMENT_IDS; FactRecord.attachment_ids,
  written to metadata_bag as a deduplicated JSON list (omitted when
  empty); build_fact_records fills it from the processed fact;
  StoredMemory.attachment_ids; list/get items carry "attachment_ids".
- search/types.py: RetrievalResult.attachment_ids (None = not carried),
  passed through ScoredResult.to_dict.
- response_models.py: MemoryFact.attachment_ids, internal (excluded
  from serialisation), set from fused results or by a store-answered
  recall.
- memory_engine.py: attachments_for_memories(..., carried=) resolves
  the carried (document_id, ids) per unit for a store-owned bank and
  still returns before any Postgres access when nothing is carried; the
  per-document resolution is shared with the unchanged SQL path.
- api/http.py: recall passes the carried ids; list/get take
  "attachment_ids" off store-rendered items so the payload has the same
  shape on either backend.

Filenames stay null for store-owned banks: document_attachments has an FK
to documents, and the store-owned retain writes no SQL documents row.

* fix(attachments): document and chunk attachments for store-owned banks

GET /documents/{id}, GET /documents/{id}/chunks and GET /chunks/{id}
returned no attachments for a bank whose memories store owns its
documents: attachments_for_chunks read the SQL chunks table and
attachments_for_documents read document_attachments, and neither holds
anything for such a bank (it keeps chunks in the store, and has no SQL
documents row for the edge FK to reference).

Same principle as the per-fact read: derive the ids from the text the
route already read from the store, and resolve them against the
attachments table.

- attachments_for_chunks(..., carried_texts=): chunk_id ->
  (document_id, chunk_text). For a store-owned bank the ids come from
  the carried text only; the chunks table is never read, and a page with
  no placeholder returns before any Postgres access.
- attachments_for_documents(..., carried_texts=): document_id -> text.
  For a store-owned bank the ids come from that text; a document with no
  carried text (the retain-ingress revisit) is read from the store, and
  its chunk texts stand in when the full text is not kept.
- http: list-chunks, get-chunk and get-document pass the texts they hold.

Filenames stay null for these banks: they live on the document edge.
The Postgres path is unchanged.

hindsight-api-slim/hindsight_api/api/http.py

@@ -956,11 +956,27 @@ async def _attach_to_memories(
     LLM call attributes the diagram to the paragraph that never mentioned it.
 
     One lookup for the whole page, not one per memory.
+
+    A store that owns its rows renders them itself and puts each memory's ids on
+    the item as ``attachment_ids``. Those are taken off here — the key is an
+    internal carrier, and leaving it would make the payload differ by backend —
+    and handed to the engine, so the lookup resolves them instead of reading them
+    back from a table the store never wrote.
     """
-    unit_ids = [item.get("id") for item in items if isinstance(item, dict) and item.get("id")]
+    unit_ids: list[str] = []
+    carried: dict[str, tuple[str | None, list[str]]] = {}
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        ids = item.pop("attachment_ids", None)
+        if not item.get("id"):
+            continue
+        unit_ids.append(item["id"])
+        if ids is not None:
+            carried[str(item["id"])] = (item.get("document_id"), list(ids))
     if not unit_ids:
         return
-    by_unit = await memory_app.attachments_for_memories(bank_id, unit_ids, request_context)
+    by_unit = await memory_app.attachments_for_memories(bank_id, unit_ids, request_context, carried=carried)
     if not by_unit:
         return
     for item in items:
@@ -974,6 +990,7 @@ async def _attach_to_recall_results(
     bank_id: str,
     results: "list[RecallResult]",
     request_context: RequestContext,
+    carried: "dict[str, tuple[str | None, list[str]]] | None" = None,
 ) -> None:
     """Add ``attachments`` to recall results — the same per-fact edge as :func:`_attach_to_memories`.
 
@@ -986,11 +1003,15 @@ async def _attach_to_recall_results(
     One lookup for the whole page. For a bank that has retained no attachments it
     is a single indexed read of the ids column that returns nothing to resolve,
     which is why this is unconditional rather than another `include` flag.
+
+    ``carried`` is unit id -> ``(document_id, attachment_ids)`` for results whose
+    ids the memories store returned on the row. For a store-owned bank that is the
+    only source: the engine resolves them and never reads ``memory_units``.
     """
     unit_ids = [result.id for result in results if result.id]
     if not unit_ids:
         return
-    by_unit = await memory_app.attachments_for_memories(bank_id, unit_ids, request_context)
+    by_unit = await memory_app.attachments_for_memories(bank_id, unit_ids, request_context, carried=carried)
     if not by_unit:
         return
     for result in results:
@@ -5769,7 +5790,17 @@ def _register_routes(app: FastAPI):
                 )
 
             recall_results = [_fact_to_result(fact) for fact in core_result.results]
-            await _attach_to_recall_results(app.state.memory, bank_id, recall_results, request_context)
+            await _attach_to_recall_results(
+                app.state.memory,
+                bank_id,
+                recall_results,
+                request_context,
+                carried={
+                    fact.id: (fact.document_id, fact.attachment_ids)
+                    for fact in core_result.results
+                    if fact.attachment_ids is not None
+                },
+            )
 
             # Convert chunks from engine to HTTP API format
             chunks_response = None
@@ -7276,7 +7307,13 @@ def _register_routes(app: FastAPI):
                 raise HTTPException(status_code=404, detail="Document not found")
             items = result.get("items") or []
             by_chunk = await app.state.memory.attachments_for_chunks(
-                bank_id, [c["chunk_id"] for c in items if c.get("chunk_id")], request_context
+                bank_id,
+                [c["chunk_id"] for c in items if c.get("chunk_id")],
+                request_context,
+                # The page already holds each chunk's text; a store-owned bank resolves from it.
+                carried_texts={
+                    c["chunk_id"]: (c.get("document_id"), c.get("chunk_text")) for c in items if c.get("chunk_id")
+                },
             )
             for chunk in items:
                 records = by_chunk.get(chunk.get("chunk_id"))
@@ -7355,7 +7392,14 @@ def _register_routes(app: FastAPI):
             document = await app.state.memory.get_document(document_id, bank_id, request_context=request_context)
             if not document:
                 raise HTTPException(status_code=404, detail="Document not found")
-            by_document = await app.state.memory.attachments_for_documents(bank_id, [document_id], request_context)
+            by_document = await app.state.memory.attachments_for_documents(
+                bank_id,
+                [document_id],
+                request_context,
+                # Used only for a store-owned bank, which has no document edge to read; a null
+                # text (full text not kept) makes the engine fall back to the chunk texts.
+                carried_texts={document_id: document.get("original_text")},
+            )
             if by_document.get(document_id):
                 document["attachments"] = [_attachment_payload(bank_id, record) for record in by_document[document_id]]
             return document
@@ -7456,7 +7500,12 @@ def _register_routes(app: FastAPI):
             # it belongs to, and attachments_for_chunks authorizes against it.
             chunk_bank = chunk.get("bank_id")
             if chunk_bank:
-                by_chunk = await app.state.memory.attachments_for_chunks(chunk_bank, [chunk_id], request_context)
+                by_chunk = await app.state.memory.attachments_for_chunks(
+                    chunk_bank,
+                    [chunk_id],
+                    request_context,
+                    carried_texts={chunk_id: (chunk.get("document_id"), chunk.get("chunk_text"))},
+                )
                 if by_chunk.get(chunk_id):
                     chunk["attachments"] = [_attachment_payload(chunk_bank, record) for record in by_chunk[chunk_id]]
             return chunk

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

@@ -12,6 +12,7 @@ from __future__ import annotations
 import logging
 
 from .base import (
+    META_ATTACHMENT_IDS,
     META_CHUNK_ID,
     CausalEdgeRecord,
     DeletePredicate,
@@ -75,6 +76,7 @@ def set_memories(memories: MemoriesExtension | None) -> None:
 
 
 __all__ = [
+    "META_ATTACHMENT_IDS",
     "META_CHUNK_ID",
     "CausalEdgeRecord",
     "DeletePredicate",

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

@@ -127,6 +127,10 @@ META_CONSOLIDATED_AT = "consolidated_at"
 # query is "not yet consolidated", so it needs a value to match on: every memory is
 # written with "0" and flipped to "1" once folded into an observation.
 META_CONSOLIDATED_FLAG = "consolidated"
+#: The attachments a fact was drawn from, as a JSON list of short ids — the per-fact
+#: provenance the extractor records. Carried on the memory so read surfaces can resolve
+#: it from the rows the store already returned, without a second lookup.
+META_ATTACHMENT_IDS = "attachment_ids"
 CONSOLIDATED_NO = "0"
 CONSOLIDATED_YES = "1"
 
@@ -193,6 +197,9 @@ class StoredMemory:
     # outside SQL has no `memory_links` table to reconstruct these from, so without them
     # on the read model an export of such a bank silently loses every causal relation.
     causal_edges: list[CausalEdgeRecord] = field(default_factory=list)
+    # Short ids of the attachments this fact was drawn from (see META_ATTACHMENT_IDS).
+    # Carried so the list and detail views resolve them from this read alone.
+    attachment_ids: list[str] = field(default_factory=list)
 
 
 @dataclass
@@ -377,6 +384,9 @@ class FactRecord:
     source_memory_ids: list[str] = field(default_factory=list)
     # When this memory was folded into an observation (sources only).
     consolidated_at: datetime | None = None
+    # Short ids of the attachments this fact was drawn from — what Postgres keeps in
+    # `memory_units.attachment_ids`. Empty for a fact stated in plain text.
+    attachment_ids: list[str] = field(default_factory=list)
 
     def metadata_bag(self) -> dict[str, str]:
         """Render the non-modelled columns as an opaque str→str bag."""
@@ -412,6 +422,9 @@ class FactRecord:
         # Observations are not themselves consolidated, so only sources carry the flag.
         if self.fact_type != "observation":
             bag[META_CONSOLIDATED_FLAG] = CONSOLIDATED_YES if self.consolidated_at else CONSOLIDATED_NO
+        if self.attachment_ids:
+            # Deduplicated in first-seen order, the same normalisation the SQL write applies.
+            bag[META_ATTACHMENT_IDS] = json.dumps(list(dict.fromkeys(self.attachment_ids)))
         return bag
 
 
@@ -495,6 +508,7 @@ def build_fact_records(
                 created_at=now,
                 entity_ids=entity_ids,
                 causal_edges=causal_edges,
+                attachment_ids=list(getattr(fact, "attachment_ids", None) or []),
             )
         )
     return records
@@ -1685,11 +1699,20 @@ class MemoriesExtension(Extension, ABC):
 
         ``total`` is the count matching the filters, not the page size, because
         the UI pages on it.
+
+        A store that owns its rows puts each memory's attachment ids on its item as
+        ``"attachment_ids": list[str]`` (see :data:`META_ATTACHMENT_IDS`). The HTTP
+        layer takes the key off and resolves the ids; it cannot read them back from
+        ``memory_units``, which holds none of a store-owned bank's memories. An item
+        without the key shows no attachments.
         """
 
     @abstractmethod
     async def get_memory_unit(self, *, conn, ops, fq_table, bank_id: str, unit_id: str) -> dict[str, Any] | None:
-        """One memory rendered for the curation detail view, or ``None``."""
+        """One memory rendered for the curation detail view, or ``None``.
+
+        Carries ``"attachment_ids"`` on the same terms as :meth:`list_memory_units`.
+        """
 
     # ------------------------------------------------------------------ curation archive
     #
@@ -2020,6 +2043,7 @@ class MemoriesExtension(Extension, ABC):
 __all__ = [
     "CONSOLIDATED_NO",
     "CONSOLIDATED_YES",
+    "META_ATTACHMENT_IDS",
     "META_CHUNK_ID",
     "META_CONSOLIDATED_AT",
     "META_CONSOLIDATED_FLAG",

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

@@ -22,7 +22,7 @@ import random
 import sys
 import time
 import uuid
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Iterator, Sequence
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Iterator, Mapping, Sequence
 from contextlib import AsyncExitStack, asynccontextmanager, contextmanager
 from dataclasses import dataclass, field, replace
 from datetime import UTC, datetime, timedelta, timezone
@@ -2017,6 +2017,36 @@ def _attachment_ids_of(value: "Any") -> list[str]:
     return [str(v) for v in value]
 
 
+async def _resolve_memory_attachments(
+    conn,
+    bank_id: str,
+    refs: "Mapping[str, tuple[str | None, Sequence[str]]]",
+) -> "dict[str, list[StoredAttachment]]":
+    """Resolve each memory's attachment ids, keyed by unit id; ``refs`` is unit id -> (document_id, ids).
+
+    Where the ids came from — `memory_units` or the store's own rows — is the caller's
+    concern; this only reads the SQL ``attachments`` / ``document_attachments`` tables.
+    Resolved per document because the filename lives on the document edge, and a page
+    of memories usually spans very few documents. A memory whose ids all fail to
+    resolve (the blob was reclaimed) is omitted rather than mapped to an empty list.
+    """
+    from .retain.attachment_store import load_bank_attachments
+
+    by_document: dict[str | None, dict[str, StoredAttachment]] = {}
+    for document_id, ids in refs.values():
+        cached = by_document.setdefault(document_id, {})
+        missing = [i for i in dict.fromkeys(ids) if i not in cached]
+        if missing:
+            cached.update(await load_bank_attachments(conn, bank_id, missing, document_id=document_id))
+
+    resolved: dict[str, list[StoredAttachment]] = {}
+    for unit_id, (document_id, ids) in refs.items():
+        records = [by_document[document_id][i] for i in ids if i in by_document[document_id]]
+        if records:
+            resolved[unit_id] = records
+    return resolved
+
+
 def _provider_default_base_url(provider: str | None) -> str:
     """The base URL a provider needs when the caller did not supply one.
 
@@ -6559,17 +6589,33 @@ class MemoryEngine(MemoryEngineInterface):
         bank_id: str,
         document_ids: "Sequence[str]",
         request_context: "RequestContext",
+        *,
+        carried_texts: "Mapping[str, str | None] | None" = None,
     ) -> "dict[str, list[StoredAttachment]]":
         """The attachments each document references, keyed by document_id.
 
         Read from ``document_attachments`` rather than by re-parsing the document
         body: that table is derived from the same text on every write, and joining
         it avoids pulling whole documents back just to scan them for placeholders.
+
+        A store-owned bank has no SQL ``documents`` row, so no ``document_attachments``
+        row can exist for it (the edge's FK needs the document row). There the ids are
+        derived from the document's text instead: ``carried_texts`` (document_id -> the
+        text a caller already read from the store) when given, else the store's own
+        record, falling back to its chunk texts when the full text is not kept.
+        Filenames live only on that edge, so they come back ``None`` for such a bank.
         """
         from .retain.attachment_store import StoredAttachment
 
         if not document_ids:
             return {}
+        from .memories import get_memories
+
+        store = get_memories()
+        if store.store_owned_for(bank_id):
+            return await self._attachments_for_store_owned_documents(
+                store, bank_id, list(dict.fromkeys(document_ids)), request_context, carried_texts or {}
+            )
         profile = await self.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
         if profile is None:
             return {}
@@ -6603,11 +6649,57 @@ class MemoryEngine(MemoryEngineInterface):
             )
         return grouped
 
+    async def _attachments_for_store_owned_documents(
+        self,
+        store,
+        bank_id: str,
+        document_ids: list[str],
+        request_context: "RequestContext",
+        carried_texts: "Mapping[str, str | None]",
+    ) -> "dict[str, list[StoredAttachment]]":
+        """:meth:`attachments_for_documents` for a store-owned bank: ids derived from the text.
+
+        A carried text costs nothing to scan. A document without one is read from the store —
+        the retain-ingress revisit has no text in hand, and it only asks when the caller wrote
+        something placeholder-shaped. The record's ``original_text`` is null when a deployment
+        does not keep full text; its chunk texts still carry every placeholder, so they stand in.
+        """
+        from .retain.attachment_content import iter_placeholder_ids
+
+        texts = {d: carried_texts.get(d) for d in document_ids}
+        if all(texts[d] is not None for d in document_ids) and not any(
+            any(True for _ in iter_placeholder_ids(texts[d] or "")) for d in document_ids
+        ):
+            return {}
+        profile = await self.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
+        if profile is None:
+            return {}
+        for document_id in document_ids:
+            if texts[document_id] is not None:
+                continue
+            record = await store.get_document_record(bank_id=bank_id, document_id=document_id, include_text=True)
+            if record is None:
+                continue
+            text = record.get("original_text")
+            if text is None:
+                text = "\n".join(
+                    t or "" for t in (await store.list_chunk_texts(bank_id=bank_id, document_id=document_id) or [])
+                )
+            texts[document_id] = text
+        refs = {d: (d, ids) for d in document_ids if (ids := list(dict.fromkeys(iter_placeholder_ids(texts[d] or ""))))}
+        if not refs:
+            return {}
+        backend = await self._get_backend()
+        async with backend.acquire() as conn:
+            return await _resolve_memory_attachments(conn, bank_id, refs)
+
     async def attachments_for_chunks(
         self,
         bank_id: str,
         chunk_ids: "Sequence[str]",
         request_context: "RequestContext",
+        *,
+        carried_texts: "Mapping[str, tuple[str | None, str | None]] | None" = None,
     ) -> "dict[str, list[StoredAttachment]]":
         """The attachments each chunk references, keyed by chunk_id.
 
@@ -6615,12 +6707,34 @@ class MemoryEngine(MemoryEngineInterface):
         and one chunk usually yields several facts of which only some were read
         off the screenshot. Per-fact provenance comes from
         ``attachments_for_memories`` instead.
+
+        ``carried_texts`` is chunk_id -> ``(document_id, chunk_text)`` for chunks whose
+        text the caller already read from the memories store. For a store-owned bank it
+        is the only source: such a bank keeps no SQL ``chunks`` rows, so reading them
+        could only come back empty.
         """
         from .retain.attachment_content import iter_placeholder_ids
         from .retain.attachment_store import load_bank_attachments
 
         if not chunk_ids:
             return {}
+        from .memories import get_memories
+
+        if get_memories().store_owned_for(bank_id):
+            wanted = set(chunk_ids)
+            refs = {
+                chunk_id: (document_id, ids)
+                for chunk_id, (document_id, text) in (carried_texts or {}).items()
+                if chunk_id in wanted and (ids := list(dict.fromkeys(iter_placeholder_ids(text or ""))))
+            }
+            if not refs:
+                return {}
+            profile = await self.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
+            if profile is None:
+                return {}
+            backend = await self._get_backend()
+            async with backend.acquire() as conn:
+                return await _resolve_memory_attachments(conn, bank_id, refs)
         profile = await self.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
         if profile is None:
             return {}
@@ -6663,6 +6777,8 @@ class MemoryEngine(MemoryEngineInterface):
         bank_id: str,
         unit_ids: "Sequence[str]",
         request_context: "RequestContext",
+        *,
+        carried: "Mapping[str, tuple[str | None, Sequence[str]]] | None" = None,
     ) -> "dict[str, list[StoredAttachment]]":
         """The attachments each memory was actually drawn from, keyed by unit id.
 
@@ -6672,25 +6788,41 @@ class MemoryEngine(MemoryEngineInterface):
         mentioned it. A fact stated in the text has no ids and correctly shows
         nothing.
 
-        The ids live on ``memory_units.attachment_ids`` — an attribute of the
-        memory, like its tags — so this reads the column and resolves the ids,
-        rather than joining a junction table that only a Postgres-backed memory
-        store would ever have written.
+        The ids are an attribute of the memory, like its tags. For a Postgres-backed
+        bank they live on ``memory_units.attachment_ids`` and this reads the column.
+        For a store-owned bank they come back on the rows the store already returned
+        — recall results and list/detail items — and the caller hands them in as
+        ``carried``: unit id -> ``(document_id, attachment_ids)``. Either way only the
+        ids are resolved here, against the SQL ``attachments`` / ``document_attachments``
+        tables, which every bank writes and which carry no vector indexes.
         """
-        from .retain.attachment_store import load_bank_attachments
-
         if not unit_ids:
             return {}
-        # A store-owned bank keeps its memories outside SQL, so `memory_units` holds none of
-        # them and this read can only come back empty. It is not a cheap empty read either:
-        # the table carries partial vector indexes per bank, and the planner opens and locks
-        # every one of them to plan any statement against it. In a tenant with a few thousand
-        # banks that is ~15k locks and ~450ms of planning to return nothing -- on every recall,
-        # which is where this is called from.
         from .memories import get_memories
 
         if get_memories().store_owned_for(bank_id):
-            return {}
+            # Never `memory_units` for a store-owned bank. It holds none of the bank's memories,
+            # so the read can only come back empty, and it is not a cheap empty read: the table
+            # carries partial vector indexes per bank, and the planner opens and locks every one
+            # of them to plan any statement against it. In a tenant with a few thousand banks
+            # that is ~15k locks and ~450ms of planning to return nothing -- on every recall.
+            # The ids the store returned on its rows are the whole answer, so a page that
+            # carried none returns before touching Postgres at all.
+            wanted = {str(u) for u in unit_ids}
+            refs = {
+                unit_id: (document_id, list(ids))
+                for unit_id, (document_id, ids) in (carried or {}).items()
+                if unit_id in wanted and ids
+            }
+            if not refs:
+                return {}
+            profile = await self.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
+            if profile is None:
+                return {}
+            backend = await self._get_backend()
+            async with backend.acquire() as conn:
+                return await _resolve_memory_attachments(conn, bank_id, refs)
+
         profile = await self.get_bank_profile(bank_id, request_context=request_context, create_if_missing=False)
         if profile is None:
             return {}
@@ -6709,31 +6841,12 @@ class MemoryEngine(MemoryEngineInterface):
             )
             if not rows:
                 return {}
-            ids_by_unit = {row["id"]: ids for row in rows if (ids := _attachment_ids_of(row["attachment_ids"]))}
-            document_by_unit = {row["id"]: row["document_id"] for row in rows}
-            # The filename lives on the document edge, so resolve per document.
-            # A page of memories usually spans very few documents, and the common
-            # case is one.
-            by_document: dict[str | None, dict[str, StoredAttachment]] = {}
-            for unit_id, ids in ids_by_unit.items():
-                document_id = document_by_unit.get(unit_id)
-                if document_id not in by_document:
-                    by_document[document_id] = {}
-                missing = [i for i in ids if i not in by_document[document_id]]
-                if missing:
-                    by_document[document_id].update(
-                        await load_bank_attachments(conn, bank_id, missing, document_id=document_id)
-                    )
-
-        return {
-            unit_id: [
-                by_document[document_by_unit.get(unit_id)][i]
-                for i in ids
-                if i in by_document[document_by_unit.get(unit_id)]
-            ]
-            for unit_id, ids in ids_by_unit.items()
-            if any(i in by_document[document_by_unit.get(unit_id)] for i in ids)
-        }
+            refs = {
+                row["id"]: (row["document_id"], ids)
+                for row in rows
+                if (ids := _attachment_ids_of(row["attachment_ids"]))
+            }
+            return await _resolve_memory_attachments(conn, bank_id, refs)
 
     async def retrieve_bank_attachment(
         self,
@@ -9004,6 +9117,7 @@ class MemoryEngine(MemoryEngineInterface):
                         tags=result_dict.get("tags"),
                         source_fact_ids=source_fact_ids_by_obs.get(result_id) if include_source_facts else None,
                         scores=scores_by_id.get(result_id),
+                        attachment_ids=result_dict.get("attachment_ids"),
                     )
                 )
 

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

@@ -406,6 +406,11 @@ class MemoryFact(BaseModel):
         None,
         description="Recall scores from each pipeline stage (final/reranker/semantic/keyword). Not returned for source facts.",
     )
+    # Internal, never serialised: the short ids of the attachments this fact was drawn
+    # from, when the memories store returned them on the row. ``None`` means "not
+    # carried", and the HTTP layer then reads them from `memory_units` instead; a list
+    # (possibly empty) is resolved as-is, so a store that owns its rows is never asked twice.
+    attachment_ids: list[str] | None = Field(None, exclude=True)
 
 
 class ChunkInfo(BaseModel):

hindsight-api-slim/hindsight_api/engine/search/types.py

@@ -92,6 +92,13 @@ class RetrievalResult:
     # re-fetch, so the two paths stay interchangeable rather than one being an approximation.
     source_memory_ids: list[str] | None = None
 
+    # Short ids of the attachments this fact was drawn from, if the backend carried them.
+    # ``None`` means "not carried" (the default store, which reads them back from
+    # ``memory_units.attachment_ids`` when the response is rendered); a list — possibly
+    # empty — means the backend returned them on the row, so the read surface resolves
+    # them without asking the store again.
+    attachment_ids: list[str] | None = None
+
     # Retrieval-specific scores (only one will be set depending on retrieval method)
     similarity: float | None = None  # Semantic retrieval
     bm25_score: float | None = None  # BM25 retrieval
@@ -216,6 +223,7 @@ class ScoredResult:
             "chunk_id": self.retrieval.chunk_id,
             "tags": self.retrieval.tags,
             "metadata": self.retrieval.metadata,
+            "attachment_ids": self.retrieval.attachment_ids,
             "semantic_similarity": self.retrieval.similarity,
             "bm25_score": self.retrieval.bm25_score,
         }

hindsight-api-slim/tests/test_attachments_store_owned.py

@@ -1,23 +1,105 @@
-"""A store-owned bank's attachment lookup must not touch Postgres.
+"""A store-owned bank's per-fact attachments come off the rows the store returned.
 
-Recall resolves the attachments behind each fact through ``attachments_for_memories``, which
-reads ``memory_units.attachment_ids``. For a bank whose memories store owns its rows, that table
-holds none of them, so the read can only come back empty -- and it is not a cheap empty read. The
-table carries partial vector indexes per bank, and the planner opens and locks every index on a
-table to plan any statement against it: in a tenant with a few thousand banks, ~15k locks and
-hundreds of milliseconds of planning, on every recall.
+Recall, list-memories and get-memory report the attachments each fact was drawn from. For a
+Postgres-backed bank the ids are read from ``memory_units.attachment_ids``. For a bank whose
+memories store owns its rows that table holds none of them, and reading it is not a cheap empty
+read: the table carries partial vector indexes per bank, and the planner opens and locks every
+index on a table to plan any statement against it -- in a tenant with a few thousand banks, ~15k
+locks and hundreds of milliseconds of planning, on every recall.
 
-So the property asserted is that the lookup returns before it asks for the bank profile or a
-connection, not merely that it returns ``{}``: an empty result is what the expensive read
-produced too.
+So the store carries the ids on what it already returns (recall rows, list/detail items), and the
+read surfaces resolve them from there. The properties pinned here:
+
+* the ids survive the write model (``FactRecord.metadata_bag``) and the retain pipeline's record
+  builder;
+* the lookup for a store-owned bank never plans a statement against ``memory_units``, and with
+  nothing carried it returns before asking for the bank profile or a connection at all;
+* every read surface actually hands back the ``attachments`` for such a bank -- asserted on the
+  HTTP payload, over a bank whose ``memory_units`` holds nothing, so the only way to produce them
+  is from the carried ids.
 """
 
+import json
+import uuid
+from contextlib import asynccontextmanager
+from datetime import datetime, timezone
 from types import SimpleNamespace
 
 import pytest
 
 import hindsight_api.engine.memories as memories_module
+from hindsight_api.engine.chunk_ids import build_chunk_id
+from hindsight_api.engine.memories import set_memories
+from hindsight_api.engine.memories.base import META_ATTACHMENT_IDS, FactRecord, StoredMemory, build_fact_records
 from hindsight_api.engine.memory_engine import MemoryEngine
+from hindsight_api.engine.response_models import MemoryFact, RecallResult
+from hindsight_api.engine.retain.attachment_content import (
+    attachment_placeholder,
+    compute_attachment_hash,
+    short_attachment_id,
+)
+from hindsight_api.engine.retain.attachment_store import StoredAttachment, _record_attachments
+from tests.test_memories_extension import InMemoryMemories
+
+UNIT_A = "00000000-0000-0000-0000-00000000000a"
+UNIT_B = "00000000-0000-0000-0000-00000000000b"
+UNIT_PLAIN = "00000000-0000-0000-0000-0000000000cc"
+# A real hash/short-id pair, so a placeholder written into document or chunk text resolves.
+SHOT_HASH = compute_attachment_hash(b"the vpn reset screenshot")
+SHOT = short_attachment_id(SHOT_HASH)
+DIAGRAM = "0f1e2d3c4b5a"
+DOCUMENT_ID = "vpn-article"
+
+
+# -- the write model ---------------------------------------------------------
+
+
+def test_metadata_bag_carries_attachment_ids_deduplicated_in_order():
+    record = FactRecord(
+        unit_id=UNIT_A, text="t", embedding=[0.0], fact_type="world", attachment_ids=[SHOT, DIAGRAM, SHOT]
+    )
+
+    bag = record.metadata_bag()
+
+    assert json.loads(bag[META_ATTACHMENT_IDS]) == [SHOT, DIAGRAM]
+
+
+def test_metadata_bag_omits_the_key_for_a_fact_with_no_attachments():
+    """A plain-text fact is the overwhelmingly common case: no key, not an empty list."""
+    record = FactRecord(unit_id=UNIT_A, text="t", embedding=[0.0], fact_type="world")
+
+    assert META_ATTACHMENT_IDS not in record.metadata_bag()
+
+
+def test_build_fact_records_carries_each_facts_attachment_ids():
+    """The builder the store-owned retain paths use must not drop the extractor's attribution."""
+
+    def _fact(ids):
+        return SimpleNamespace(
+            fact_text="t",
+            embedding=[0.0],
+            fact_type="world",
+            tags=[],
+            context=None,
+            document_id="doc",
+            chunk_id=None,
+            metadata=None,
+            observation_scopes=None,
+            entities=[],
+            causal_relations=[],
+            occurred_start=None,
+            occurred_end=None,
+            mentioned_at=None,
+            attachment_ids=ids,
+        )
+
+    records = build_fact_records([UNIT_A, UNIT_PLAIN], [_fact([SHOT]), _fact([])])
+
+    assert [r.attachment_ids for r in records] == [[SHOT], []]
+    assert json.loads(records[0].metadata_bag()[META_ATTACHMENT_IDS]) == [SHOT]
+
+
+# -- the engine lookup -------------------------------------------------------
 
 
 class _NoPostgres:
@@ -30,27 +112,366 @@ class _NoPostgres:
         raise AssertionError("a store-owned bank's attachment lookup took a connection")
 
 
+class _AttachmentsOnlyConn:
+    """Answers the attachment-table read; fails on any statement that names ``memory_units``."""
+
+    def __init__(self):
+        self.statements: list[str] = []
+
+    async def fetch(self, sql, bank_id, ids, document_id=None):
+        self.statements.append(sql)
+        assert "memory_units" not in sql, "a store-owned bank's attachment lookup planned against memory_units"
+        known = {
+            SHOT: ("h" * 64, "image/png", 10, "image"),
+            DIAGRAM: ("d" * 64, "image/svg+xml", 20, "image"),
+        }
+        return [
+            {
+                "attachment_hash": known[i][0],
+                "short_id": i,
+                "media_type": known[i][1],
+                "byte_size": known[i][2],
+                "storage_key": f"k/{i}",
+                "kind": known[i][3],
+                "filename": None,
+            }
+            for i in ids
+            if i in known
+        ]
+
+
+class _EngineWithConn:
+    def __init__(self, conn):
+        self.conn = conn
+
+    async def get_bank_profile(self, *a, **k):
+        return {"bank_id": "bank-1"}
+
+    async def _get_backend(self):
+        conn = self.conn
+
+        class _Backend:
+            @asynccontextmanager
+            async def acquire(self):
+                yield conn
+
+        return _Backend()
+
+
 def _memories(store_owned: bool):
     return SimpleNamespace(store_owned_for=lambda bank_id: store_owned)
 
 
 @pytest.mark.asyncio
 async def test_a_store_owned_bank_resolves_no_attachments_without_touching_postgres(monkeypatch):
+    """With nothing carried there is nothing to resolve, so not even the bank profile is read.
+
+    The property is that the lookup returns before it asks for the bank profile or a connection,
+    not merely that it returns ``{}``: an empty result is what the expensive read produced too.
+    """
     monkeypatch.setattr(memories_module, "get_memories", lambda: _memories(store_owned=True))
 
-    result = await MemoryEngine.attachments_for_memories(
-        _NoPostgres(), "bank-1", ["00000000-0000-0000-0000-000000000001"], request_context=None
-    )
+    result = await MemoryEngine.attachments_for_memories(_NoPostgres(), "bank-1", [UNIT_A], request_context=None)
 
     assert result == {}
 
 
+@pytest.mark.asyncio
+async def test_a_store_owned_bank_resolves_the_carried_ids_without_memory_units(monkeypatch):
+    monkeypatch.setattr(memories_module, "get_memories", lambda: _memories(store_owned=True))
+    conn = _AttachmentsOnlyConn()
+
+    result = await MemoryEngine.attachments_for_memories(
+        _EngineWithConn(conn),
+        "bank-1",
+        [UNIT_A, UNIT_B, UNIT_PLAIN],
+        request_context=None,
+        carried={
+            UNIT_A: ("doc-1", [SHOT, DIAGRAM]),
+            UNIT_B: ("doc-2", [DIAGRAM]),
+            UNIT_PLAIN: ("doc-1", []),
+            # Carried for a unit this page is not rendering: not resolved, not returned.
+            "00000000-0000-0000-0000-0000000000ff": ("doc-1", [SHOT]),
+        },
+    )
+
+    assert {unit: [r.short_id for r in records] for unit, records in result.items()} == {
+        UNIT_A: [SHOT, DIAGRAM],
+        UNIT_B: [DIAGRAM],
+    }
+    assert result[UNIT_A][0].media_type == "image/png"
+    # One read per document: the filename lives on the document edge.
+    assert len(conn.statements) == 2
+
+
 @pytest.mark.asyncio
 async def test_a_bank_whose_rows_live_in_sql_still_reads_them(monkeypatch):
     """The guard must not swallow the Postgres-backed case: there the read is the feature."""
     monkeypatch.setattr(memories_module, "get_memories", lambda: _memories(store_owned=False))
 
     with pytest.raises(AssertionError, match="read the bank profile"):
-        await MemoryEngine.attachments_for_memories(
-            _NoPostgres(), "bank-1", ["00000000-0000-0000-0000-000000000001"], request_context=None
+        await MemoryEngine.attachments_for_memories(_NoPostgres(), "bank-1", [UNIT_A], request_context=None)
+
+
+# -- every read surface, end to end -----------------------------------------
+
+
+class _CarryingStore(InMemoryMemories):
+    """A store-owned store that returns each memory's attachment ids on its rows.
+
+    ``answers_full_recall`` picks which recall path is exercised: the store answering the whole
+    recall (its rows become ``MemoryFact`` directly), or declining it so the engine fuses the arm
+    results (its rows become ``RetrievalResult``). Both must surface the same attachments.
+    """
+
+    def __init__(self, answers_full_recall: bool):
+        super().__init__({})
+        self.answers_full_recall = answers_full_recall
+
+    async def full_recall(self, request):
+        if not self.answers_full_recall:
+            return None
+        return RecallResult(
+            results=[
+                MemoryFact(
+                    id=row.unit_id,
+                    text=row.text,
+                    fact_type=row.fact_type,
+                    document_id=row.document_id,
+                    attachment_ids=list(row.attachment_ids),
+                )
+                for row in self.rows.values()
+            ]
         )
+
+    async def search(self, *, conn, bank_id, fact_types, query_embedding, query_text, limit, **kwargs):
+        out = await super().search(
+            conn=conn,
+            bank_id=bank_id,
+            fact_types=fact_types,
+            query_embedding=query_embedding,
+            query_text=query_text,
+            limit=limit,
+            **kwargs,
+        )
+        for arms in out.values():
+            for result in [*arms.semantic, *arms.bm25]:
+                result.attachment_ids = list(self.rows[result.id].attachment_ids)
+        return out
+
+    async def get_document_record(self, *, bank_id, document_id, include_text=False):
+        # A real store's record carries its write stamps (epoch ms), which the document route
+        # requires; the base stub leaves them out because none of its own tests render one.
+        record = await super().get_document_record(bank_id=bank_id, document_id=document_id, include_text=include_text)
+        if record is not None:
+            stamp = int(datetime.now(timezone.utc).timestamp() * 1000)
+            record.update(created_at=stamp, updated_at=stamp)
+        return record
+
+    def _render(self, row: StoredMemory) -> dict:
+        return {
+            "id": row.unit_id,
+            "text": row.text,
+            "fact_type": row.fact_type,
+            "document_id": row.document_id,
+            "attachment_ids": list(row.attachment_ids),
+        }
+
+    async def list_memory_units(self, *, conn, ops, fq_table, bank_id, limit=100, offset=0, **kwargs):
+        items = [self._render(row) for row in self.rows.values()]
+        return {"items": items[offset : offset + limit], "total": len(items), "limit": limit, "offset": offset}
+
+    async def get_memory_unit(self, *, conn, ops, fq_table, bank_id, unit_id):
+        row = self.rows.get(str(unit_id))
+        return None if row is None else self._render(row)
+
+
+@pytest.fixture
+def restore_default_store():
+    yield
+    set_memories(None)
+
+
+async def _store_owned_bank(memory, request_context, answers_full_recall: bool) -> tuple[str, _CarryingStore]:
+    bank_id = f"so-attach-{uuid.uuid4().hex[:8]}"
+    store = _CarryingStore(answers_full_recall)
+    set_memories(store)
+    # The store owns the facts, never the bank row, so create it the way a retain would.
+    await memory.get_bank_profile(bank_id, request_context=request_context)
+    backend = await memory._get_backend()
+    async with backend.acquire() as conn:
+        await _record_attachments(
+            conn,
+            bank_id,
+            [
+                StoredAttachment(
+                    attachment_hash=SHOT_HASH,
+                    short_id=SHOT,
+                    media_type="image/png",
+                    byte_size=68,
+                    storage_key=f"attachments/{bank_id}/{SHOT}",
+                    kind="image",
+                )
+            ],
+        )
+    now = datetime.now(timezone.utc)
+    store.rows[UNIT_A] = StoredMemory(
+        unit_id=UNIT_A,
+        text="To reset the VPN, click the reset button.",
+        fact_type="world",
+        document_id="vpn-article",
+        created_at=now,
+        attachment_ids=[SHOT],
+    )
+    store.rows[UNIT_PLAIN] = StoredMemory(
+        unit_id=UNIT_PLAIN,
+        text="The VPN client is called Sentinel.",
+        fact_type="world",
+        document_id="vpn-article",
+        created_at=now,
+    )
+    return bank_id, store
+
+
+def _assert_shot(attachments, bank_id: str) -> None:
+    assert attachments, "no attachments returned"
+    assert [a["id"] for a in attachments] == [SHOT]
+    assert attachments[0]["media_type"] == "image/png"
+    assert attachments[0]["url"] == f"/v1/default/banks/{bank_id}/attachments/{SHOT}"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("answers_full_recall", [True, False], ids=["store-answered", "engine-fused"])
+async def test_recall_returns_the_attachments_the_store_carried(
+    api_client, memory, request_context, restore_default_store, answers_full_recall
+):
+    bank_id, _ = await _store_owned_bank(memory, request_context, answers_full_recall)
+
+    response = await api_client.post(
+        f"/v1/default/banks/{bank_id}/memories/recall",
+        json={"query": "how do I reset the VPN", "types": ["world"], "limit": 10},
+    )
+
+    assert response.status_code == 200, response.text
+    by_id = {r["id"]: r for r in response.json()["results"]}
+    assert set(by_id) == {UNIT_A, UNIT_PLAIN}
+    _assert_shot(by_id[UNIT_A].get("attachments"), bank_id)
+    assert by_id[UNIT_PLAIN].get("attachments") is None
+    # The carrier stays internal: the payload is the same shape on either backend.
+    assert all("attachment_ids" not in r for r in by_id.values())
+
+
+@pytest.mark.asyncio
+async def test_list_and_get_return_the_attachments_the_store_carried(
+    api_client, memory, request_context, restore_default_store
+):
+    bank_id, _ = await _store_owned_bank(memory, request_context, answers_full_recall=True)
+
+    listed = await api_client.get(f"/v1/default/banks/{bank_id}/memories/list")
+    detail = await api_client.get(f"/v1/default/banks/{bank_id}/memories/{UNIT_A}")
+
+    assert listed.status_code == 200, listed.text
+    items = {m["id"]: m for m in listed.json()["items"]}
+    _assert_shot(items[UNIT_A].get("attachments"), bank_id)
+    assert items[UNIT_PLAIN].get("attachments") is None
+    assert all("attachment_ids" not in m for m in items.values())
+
+    assert detail.status_code == 200, detail.text
+    _assert_shot(detail.json().get("attachments"), bank_id)
+    assert "attachment_ids" not in detail.json()
+
+
+# -- documents and chunks ----------------------------------------------------
+
+
+class _NoChunkTableConn(_AttachmentsOnlyConn):
+    """Also fails on any read of the SQL ``chunks`` table, which a store-owned bank never fills."""
+
+    async def fetch(self, sql, bank_id, ids, document_id=None):
+        assert " chunks" not in sql and '"chunks"' not in sql and ".chunks" not in sql, (
+            "a store-owned bank's chunk attachment lookup read the SQL chunks table"
+        )
+        return await super().fetch(sql, bank_id, ids, document_id)
+
+
+@pytest.mark.asyncio
+async def test_a_store_owned_chunk_with_no_placeholder_touches_no_postgres(monkeypatch):
+    monkeypatch.setattr(memories_module, "get_memories", lambda: _memories(store_owned=True))
+
+    result = await MemoryEngine.attachments_for_chunks(
+        _NoPostgres(), "bank-1", ["c0"], request_context=None, carried_texts={"c0": ("doc-1", "plain prose")}
+    )
+
+    assert result == {}
+
+
+@pytest.mark.asyncio
+async def test_a_store_owned_chunk_resolves_from_the_carried_text(monkeypatch):
+    monkeypatch.setattr(memories_module, "get_memories", lambda: _memories(store_owned=True))
+    conn = _NoChunkTableConn()
+
+    result = await MemoryEngine.attachments_for_chunks(
+        _EngineWithConn(conn),
+        "bank-1",
+        ["c0", "c1"],
+        request_context=None,
+        carried_texts={
+            "c0": ("doc-1", f"click {attachment_placeholder(SHOT_HASH)} then reconnect"),
+            "c1": ("doc-1", "no image here"),
+        },
+    )
+
+    assert {chunk: [r.short_id for r in records] for chunk, records in result.items()} == {"c0": [SHOT]}
+
+
+async def _seed_document(store: _CarryingStore, bank_id: str, *, keep_text: bool) -> str:
+    text = f"To reset the VPN, click the button: {attachment_placeholder(SHOT_HASH)}\nThen reconnect."
+    await store.put_document(
+        bank_id=bank_id,
+        document_id=DOCUMENT_ID,
+        content_hash="h",
+        original_text=text if keep_text else None,
+        chunk_texts=[text, "Then reconnect."],
+    )
+    return text
+
+
+@pytest.mark.asyncio
+async def test_document_and_chunk_reads_return_the_attachments_in_the_stored_text(
+    api_client, memory, request_context, restore_default_store
+):
+    bank_id, store = await _store_owned_bank(memory, request_context, answers_full_recall=True)
+    await _seed_document(store, bank_id, keep_text=True)
+
+    document = await api_client.get(f"/v1/default/banks/{bank_id}/documents/{DOCUMENT_ID}")
+    chunks = await api_client.get(f"/v1/default/banks/{bank_id}/documents/{DOCUMENT_ID}/chunks")
+    chunk = await api_client.get(f"/v1/default/chunks/{build_chunk_id(bank_id, DOCUMENT_ID, 0)}")
+
+    assert document.status_code == 200, document.text
+    _assert_shot(document.json().get("attachments"), bank_id)
+    # The filename lives on the document edge, which a store-owned bank cannot have.
+    assert document.json()["attachments"][0].get("filename") is None
+
+    assert chunks.status_code == 200, chunks.text
+    items = chunks.json()["items"]
+    _assert_shot(items[0].get("attachments"), bank_id)
+    assert items[1].get("attachments") is None
+
+    assert chunk.status_code == 200, chunk.text
+    _assert_shot(chunk.json().get("attachments"), bank_id)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("keep_text", [True, False], ids=["full-text", "chunks-only"])
+async def test_the_retain_revisit_lookup_reads_the_stored_text(
+    memory, request_context, restore_default_store, keep_text
+):
+    """The retain ingress asks which attachments a document already references, with no text in
+    hand, so that an edit re-sending its placeholders keeps them. For a store-owned bank the answer
+    comes from the stored text -- and from the chunk texts when the full text is not kept."""
+    bank_id, store = await _store_owned_bank(memory, request_context, answers_full_recall=True)
+    await _seed_document(store, bank_id, keep_text=keep_text)
+
+    existing = await memory.attachments_for_documents(bank_id, [DOCUMENT_ID, "never-retained"], request_context)
+
+    assert {d: [r.short_id for r in records] for d, records in existing.items()} == {DOCUMENT_ID: [SHOT]}