fix(engine): stamp memory_units.updated_at on the writes that change a memory (#3490)

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

@@ -92,6 +92,22 @@ class StoreWriteUnavailable(RuntimeError):
 META_OBSERVATION_SCOPES = "observation_scopes"
 META_TEXT_SIGNALS = "text_signals"
 META_CREATED_AT = "created_at"
+#: When the memory last changed, and the contract every write path owes it (#3490):
+#: a write that changes what the memory *is* — text, context, dates, fact_type, tags,
+#: metadata, embedding, an observation's sources — stamps ``updated_at``, so a consumer
+#: chasing ``WHERE updated_at > watermark`` (incremental export, cache invalidation, the
+#: mental-model staleness check and its delta refresh) sees the change.
+#:
+#: Consolidation bookkeeping is the one deliberate exception: ``consolidated_at`` and
+#: ``consolidation_failed_at`` are scheduler state rather than the memory, and stamping
+#: them would make every consolidation pass look like an edit to every fact it folded —
+#: re-flagging mental models stale and re-feeding unchanged facts to a delta refresh.
+#: :meth:`MemoriesExtension.mark_consolidated` leaves the column alone, and so do the
+#: requeue sites that clear the markers inline. A store that owns memories itself is
+#: expected to keep the same contract.
+#:
+#: No timestamp can report a hard delete; a consumer that must catch those needs a
+#: content fingerprint, not a watermark.
 META_UPDATED_AT = "updated_at"
 # Observation bookkeeping. `source_memory_ids` is a JSON list: an implementation
 # with no edge relation carries an observation's sources denormalised.
@@ -882,6 +898,9 @@ async def mark_consolidated(
 
         ``failed`` stamps the failure marker instead, so a memory the LLM could
         not consolidate is not retried forever.
+
+        This is scheduler state, not an edit: it must leave the memory's
+        ``updated_at`` alone (see :data:`META_UPDATED_AT`).
         """
 
     @abstractmethod

hindsight-api-slim/hindsight_api/engine/memories/pg/reads.py

@@ -461,8 +461,9 @@ async def mark_consolidated(
     observations are never themselves consolidated, so nothing about them should
     be reset by a requeue.
 
-    ``updated_at`` is deliberately left alone, matching the consolidator's own
-    statements: consolidation bookkeeping is not an edit to the memory, and
+    ``updated_at`` is deliberately left alone — the one exception to the contract
+    documented on ``META_UPDATED_AT`` (``memories.base``) that every other write
+    path owes the column. Consolidation bookkeeping is not an edit to the memory, and
     bumping it would make every consolidation pass look like a write to the
     staleness check below.
     """

hindsight-api-slim/hindsight_api/engine/memories/pg/writes.py

@@ -293,6 +293,8 @@ async def delete_stale_observations(
     )
 
     if remaining_source_ids:
+        # Requeue: consolidation bookkeeping, so `updated_at` is deliberately not
+        # stamped (see META_UPDATED_AT in ..base) — nothing about these facts changed.
         await conn.execute(
             f"""
             UPDATE {fq_table("memory_units")}
@@ -420,7 +422,8 @@ async def invalidate_memory(*, conn, fq_table, bank_id: str, unit_id: str, reaso
 
 async def set_invalidation_reason(*, conn, fq_table, bank_id: str, unit_id: str, reason: str | None) -> None:
     await conn.execute(
-        f"UPDATE {fq_table('invalidated_memory_units')} SET invalidation_reason = $3 WHERE id = $1 AND bank_id = $2",
+        f"UPDATE {fq_table('invalidated_memory_units')} SET invalidation_reason = $3, updated_at = now() "
+        f"WHERE id = $1 AND bank_id = $2",
         str(unit_id),
         bank_id,
         reason,
@@ -498,7 +501,8 @@ async def restore_memory(*, conn, fq_table, bank_id: str, unit_id: str) -> Store
 
 async def set_memory_embedding(*, conn, fq_table, bank_id: str, unit_id: str, embedding) -> None:
     await conn.execute(
-        f"UPDATE {fq_table('memory_units')} SET embedding = $3::vector WHERE id = $1 AND bank_id = $2",
+        f"UPDATE {fq_table('memory_units')} SET embedding = $3::vector, updated_at = now() "
+        f"WHERE id = $1 AND bank_id = $2",
         str(unit_id),
         bank_id,
         embedding,

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

@@ -7412,7 +7412,8 @@ async def update_document(
                     unit_ids = [str(row["id"]) for row in unit_rows]
 
                     await conn.execute(
-                        f"UPDATE {fq_table('memory_units')} SET tags = $1 WHERE document_id = $2 AND bank_id = $3",
+                        f"UPDATE {fq_table('memory_units')} SET tags = $1, updated_at = now() "
+                        f"WHERE document_id = $2 AND bank_id = $3",
                         tags,
                         document_id,
                         bank_id,
@@ -7467,6 +7468,9 @@ async def update_document(
                                 f"DELETE FROM {fq_table('memory_units')} WHERE id = ANY($1::uuid[])",
                                 obs_ids,
                             )
+                            # Requeue the sources: bookkeeping only, so `updated_at`
+                            # stays put (see META_UPDATED_AT). The tag change above is
+                            # what stamped these rows.
                             await conn.execute(
                                 f"""
                                 UPDATE {fq_table("memory_units")}
@@ -8131,7 +8135,8 @@ async def clear_observations(
                         bank_id,
                     )
 
-                    # Reset consolidated_at on source memories so they get re-consolidated
+                    # Reset consolidated_at on source memories so they get re-consolidated.
+                    # Bookkeeping only: `updated_at` stays put (see META_UPDATED_AT).
                     await conn.execute(
                         f"UPDATE {fq_table('memory_units')} SET consolidated_at = NULL WHERE bank_id = $1 AND fact_type IN ('experience', 'world')",
                         bank_id,
@@ -8261,6 +8266,7 @@ async def retry_failed_consolidation(
                     """,
                     bank_id,
                 )
+                # Bookkeeping only: `updated_at` stays put (see META_UPDATED_AT).
                 await conn.execute(
                     f"""
                     UPDATE {fq_table("memory_units")}
@@ -8326,7 +8332,8 @@ async def clear_observations_for_memory(
                 deleted_count = await self._delete_stale_observations_for_memories(conn, bank_id, [memory_id])
 
                 # Also reset this memory's own consolidated_at so it gets re-consolidated
-                # (the memory was a source for the deleted observations, so it needs new ones)
+                # (the memory was a source for the deleted observations, so it needs new ones).
+                # Bookkeeping only: `updated_at` stays put (see META_UPDATED_AT).
                 if deleted_count > 0:
                     from .memories import get_memories
 

hindsight-api-slim/hindsight_api/engine/transfer/importer.py

@@ -886,7 +886,8 @@ async def _restore_fact_lifecycle(
         return
     await conn.executemany(
         f"UPDATE {fq_table('memory_units')} "
-        f"SET created_at = COALESCE($2, created_at), consolidated_at = $3, consolidation_failed_at = $4 "
+        f"SET created_at = COALESCE($2, created_at), consolidated_at = $3, consolidation_failed_at = $4, "
+        f"updated_at = now() "
         f"WHERE id = $1 AND bank_id = $5",
         [
             (unit_id, created_at, consolidated_at, failed_at, bank_id)
@@ -967,7 +968,8 @@ async def _import_observations(
                     # insert_facts_batch derives event_date for normal writes;
                     # transfer restores the source value carried by the archive.
                     await conn.execute(
-                        f"UPDATE {fq_table('memory_units')} SET event_date = $1 WHERE id = $2 AND bank_id = $3",
+                        f"UPDATE {fq_table('memory_units')} SET event_date = $1, updated_at = now() "
+                        f"WHERE id = $2 AND bank_id = $3",
                         obs.event_date,
                         observation_uuid,
                         bank_id,
@@ -1012,14 +1014,15 @@ async def _link_observation_sources(
             [(observation_id, sid) for sid in dict.fromkeys(source_ids)],
         )
         await conn.execute(
-            f"UPDATE {fq_table('memory_units')} SET proof_count = $1 WHERE id = $2 AND bank_id = $3",
+            f"UPDATE {fq_table('memory_units')} SET proof_count = $1, updated_at = now() "
+            f"WHERE id = $2 AND bank_id = $3",
             proof_count,
             observation_id,
             bank_id,
         )
     else:
         await conn.execute(
-            f"UPDATE {fq_table('memory_units')} SET source_memory_ids = $1, proof_count = $2 "
+            f"UPDATE {fq_table('memory_units')} SET source_memory_ids = $1, proof_count = $2, updated_at = now() "
             f"WHERE id = $3 AND bank_id = $4",
             source_ids,
             proof_count,

hindsight-api-slim/tests/test_memory_units_updated_at.py

@@ -0,0 +1,221 @@
+"""``memory_units.updated_at`` must move whenever a memory actually changes (#3490).
+
+Consumers chase the column (``WHERE updated_at > watermark``) for incremental
+export, cache invalidation and the mental-model staleness check, so a write path
+that changes a memory without stamping it makes the chase silently skip that
+change. Several did: the document tag propagation, ``set_memory_embedding``,
+``set_invalidation_reason`` and the transfer importer's fixups.
+
+The one deliberate exception is consolidation bookkeeping (``consolidated_at`` /
+``consolidation_failed_at``): stamping it would make every consolidation pass
+look like an edit to every fact it folded, re-flagging mental models stale and
+re-feeding unchanged facts to a delta refresh. Both halves are pinned here —
+the paths that must stamp, and the ones that must not.
+
+The assertions read ``updated_at`` with raw SQL because it is not part of
+``StoredMemory``; these are the SQL store's statements, which is where the fix
+lives.
+"""
+
+import uuid
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from hindsight_api import RequestContext
+from hindsight_api.engine.memories import get_memories
+from hindsight_api.engine.memory_engine import MemoryEngine, fq_table
+
+# A backdated baseline every case starts from, so "did this statement stamp the
+# column" is a comparison against a value no ``now()`` can collide with.
+_BASELINE = datetime(2020, 1, 1, tzinfo=timezone.utc)
+
+
+async def _seed_memory(
+    memory: MemoryEngine, conn, bank_id: str, text: str, *, document_id: str | None = None
+) -> uuid.UUID:
+    """Insert one fact through the store, bypassing the LLM retain pipeline."""
+    store = get_memories()
+    fact = SimpleNamespace(
+        fact_text=text,
+        embedding=memory.embeddings.encode([text])[0],
+        fact_type="experience",
+        tags=[],
+        context=None,
+        document_id=document_id,
+        chunk_id=None,
+        metadata=None,
+        observation_scopes=None,
+        entities=[],
+        causal_relations=[],
+        occurred_start=None,
+        occurred_end=None,
+        mentioned_at=None,
+    )
+    unit_ids = await store.insert_facts(
+        conn=conn, ops=memory._backend.ops, bank_id=bank_id, facts=[fact], document_id=document_id
+    )
+    return uuid.UUID(unit_ids[0])
+
+
+async def _backdate(conn, memory_id: uuid.UUID, table: str = "memory_units") -> None:
+    """Park ``updated_at`` in the past so a later stamp is unambiguous."""
+    await conn.execute(f"UPDATE {table} SET updated_at = $1 WHERE id = $2", _BASELINE, memory_id)
+
+
+async def _updated_at(conn, memory_id: uuid.UUID, table: str = "memory_units") -> datetime:
+    return await conn.fetchval(f"SELECT updated_at FROM {table} WHERE id = $1", memory_id)
+
+
+async def _ensure_bank(memory: MemoryEngine, bank_id: str, request_context: RequestContext) -> None:
+    await memory.get_bank_profile(bank_id=bank_id, request_context=request_context)
+
+
+class TestWritesThatMustStamp:
+    @pytest.mark.asyncio
+    async def test_document_tag_propagation_stamps_updated_at(
+        self, memory: MemoryEngine, request_context: RequestContext
+    ):
+        """Retagging a document changes its memories' tags — and their updated_at."""
+        bank_id = f"test-mu-updated-tags-{uuid.uuid4().hex[:8]}"
+        await _ensure_bank(memory, bank_id, request_context)
+        doc_id = f"doc-{uuid.uuid4().hex[:8]}"
+
+        pool = await memory._get_pool()
+        async with pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO documents (id, bank_id, original_text, content_hash, created_at, updated_at) "
+                "VALUES ($1, $2, 'some doc', 'hash123', NOW(), NOW())",
+                doc_id,
+                bank_id,
+            )
+            mem_id = await _seed_memory(memory, conn, bank_id, "Alice loves hiking.", document_id=doc_id)
+            await _backdate(conn, mem_id)
+
+        with patch.object(memory, "submit_async_consolidation", new=AsyncMock()):
+            assert await memory.update_document(doc_id, bank_id, tags=["new-tag"], request_context=request_context)
+
+        async with pool.acquire() as conn:
+            assert await _updated_at(conn, mem_id) > _BASELINE
+
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+    @pytest.mark.asyncio
+    async def test_embedding_write_stamps_updated_at(self, memory: MemoryEngine, request_context: RequestContext):
+        """A re-embed (curation edit, revert) rewrites the stored vector — that is a change."""
+        bank_id = f"test-mu-updated-embed-{uuid.uuid4().hex[:8]}"
+        await _ensure_bank(memory, bank_id, request_context)
+
+        pool = await memory._get_pool()
+        async with pool.acquire() as conn:
+            mem_id = await _seed_memory(memory, conn, bank_id, "Alice loves hiking.")
+            await _backdate(conn, mem_id)
+
+            embedding = str(list(map(float, memory.embeddings.encode(["Alice loves climbing."])[0])))
+            await get_memories().set_memory_embedding(
+                conn=conn, fq_table=fq_table, bank_id=bank_id, unit_id=str(mem_id), embedding=embedding
+            )
+
+            assert await _updated_at(conn, mem_id) > _BASELINE
+
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+    @pytest.mark.asyncio
+    async def test_invalidation_reason_change_stamps_updated_at(
+        self, memory: MemoryEngine, request_context: RequestContext
+    ):
+        """Editing an archived memory's reason is an edit to the archive row."""
+        bank_id = f"test-mu-updated-reason-{uuid.uuid4().hex[:8]}"
+        await _ensure_bank(memory, bank_id, request_context)
+
+        pool = await memory._get_pool()
+        async with pool.acquire() as conn:
+            store = get_memories()
+            mem_id = await _seed_memory(memory, conn, bank_id, "Alice loves hiking.")
+            await _backdate(conn, mem_id)
+            # The archive row is copied from the live one, so it inherits the baseline.
+            await store.invalidate_memory(
+                conn=conn, fq_table=fq_table, bank_id=bank_id, unit_id=str(mem_id), reason="wrong"
+            )
+            assert await _updated_at(conn, mem_id, "invalidated_memory_units") == _BASELINE
+
+            await store.set_invalidation_reason(
+                conn=conn, fq_table=fq_table, bank_id=bank_id, unit_id=str(mem_id), reason="outdated"
+            )
+
+            assert await _updated_at(conn, mem_id, "invalidated_memory_units") > _BASELINE
+
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+class TestConsolidationBookkeepingIsExempt:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "when,failed",
+        [
+            (datetime.now(timezone.utc), False),  # folded into an observation
+            (None, False),  # requeued after that observation was invalidated
+            (datetime.now(timezone.utc), True),  # the LLM could not consolidate it
+        ],
+        ids=["consolidated", "requeued", "failed"],
+    )
+    async def test_mark_consolidated_leaves_updated_at_alone(
+        self, memory: MemoryEngine, request_context: RequestContext, when: datetime | None, failed: bool
+    ):
+        """Scheduler state is not an edit: stamping it would re-flag every mental model stale."""
+        bank_id = f"test-mu-updated-mark-{uuid.uuid4().hex[:8]}"
+        await _ensure_bank(memory, bank_id, request_context)
+
+        pool = await memory._get_pool()
+        async with pool.acquire() as conn:
+            mem_id = await _seed_memory(memory, conn, bank_id, "Alice loves hiking.")
+            await _backdate(conn, mem_id)
+
+            await get_memories().mark_consolidated(
+                conn=conn, fq_table=fq_table, bank_id=bank_id, unit_ids=[str(mem_id)], when=when, failed=failed
+            )
+
+            assert await _updated_at(conn, mem_id) == _BASELINE
+
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+    @pytest.mark.asyncio
+    async def test_requeue_after_observation_cleanup_leaves_updated_at_alone(
+        self, memory: MemoryEngine, request_context: RequestContext
+    ):
+        """The engine's own inline requeue (clear_observations_for_memory) is bookkeeping too."""
+        bank_id = f"test-mu-updated-requeue-{uuid.uuid4().hex[:8]}"
+        await _ensure_bank(memory, bank_id, request_context)
+
+        pool = await memory._get_pool()
+        async with pool.acquire() as conn:
+            mem_id = await _seed_memory(memory, conn, bank_id, "Alice loves hiking.")
+            await get_memories().mark_consolidated(
+                conn=conn,
+                fq_table=fq_table,
+                bank_id=bank_id,
+                unit_ids=[str(mem_id)],
+                when=datetime.now(timezone.utc),
+            )
+            await conn.execute(
+                "INSERT INTO memory_units "
+                "(id, bank_id, text, fact_type, event_date, source_memory_ids, proof_count, created_at, updated_at) "
+                "VALUES ($1, $2, 'Alice is a hiker.', 'observation', NOW(), $3, 1, NOW(), NOW())",
+                uuid.uuid4(),
+                bank_id,
+                [mem_id],
+            )
+            await _backdate(conn, mem_id)
+
+        with patch.object(memory, "submit_async_consolidation", new=AsyncMock()):
+            result = await memory.clear_observations_for_memory(bank_id, str(mem_id), request_context=request_context)
+        assert result["deleted_count"] == 1
+
+        async with pool.acquire() as conn:
+            # The requeue happened (consolidated_at cleared) but the memory did not change.
+            assert await conn.fetchval("SELECT consolidated_at FROM memory_units WHERE id = $1", mem_id) is None
+            assert await _updated_at(conn, mem_id) == _BASELINE
+
+        await memory.delete_bank(bank_id, request_context=request_context)