diff --git a/hindsight-api-slim/hindsight_api/engine/memories/base.py b/hindsight-api-slim/hindsight_api/engine/memories/base.py index 2be8580410..7250b94f9a 100644 --- a/hindsight-api-slim/hindsight_api/engine/memories/base.py +++ b/hindsight-api-slim/hindsight_api/engine/memories/base.py @@ -92,6 +92,30 @@ 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`` sees the change. Those consumers are +#: incremental export, cache invalidation, the mental-model staleness check +#: (:meth:`any_memory_updated_since`) and its delta refresh — and recall's own +#: ``created_after`` / ``created_before`` window, which despite the name filters on this +#: column, so what stamps it also decides what a date-bounded recall returns. +#: +#: The consolidation *scheduler* is the one deliberate exception: when a pass records +#: that it folded a fact (or requeues one whose observation went away) it writes only +#: ``consolidated_at`` / ``consolidation_failed_at``, which are scheduler state rather +#: than the memory. Stamping there would make every 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` and the requeue sites that clear +#: the markers inline therefore leave the column alone. +#: +#: The exemption is that *situation*, not the two columns: a write that clears the markers +#: as part of a real change to the memory still stamps — :meth:`restore_memory` brings an +#: archived memory back and resets it for re-consolidation in one statement, and that is an +#: edit. 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 +906,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 @@ -1078,6 +1105,9 @@ async def restore_memory(self, *, conn, fq_table, bank_id: str, unit_id: str, tx Returns the restored memory (so the caller can recompute its embedding — the archive need not keep one), or ``None`` if it was not archived. + + Bringing a memory back is an edit, so this stamps ``updated_at`` even though + it also resets the consolidation markers (see :data:`META_UPDATED_AT`). """ @abstractmethod @@ -1088,6 +1118,10 @@ async def set_memory_embedding(self, *, conn, fq_table, bank_id: str, unit_id: s the store whose write is the row itself — reverting or editing a memory has to put a freshly computed vector back on it, so this is a real write for both. ``embedding`` is a float list or the pgvector literal. + + The vector is part of the memory, so this stamps ``updated_at`` itself rather + than leaning on the edit statement its in-tree callers happen to pair it with + (see :data:`META_UPDATED_AT`). """ async def clear_unit_entities(self, *, conn, fq_table, bank_id: str, unit_id: str) -> None: diff --git a/hindsight-api-slim/hindsight_api/engine/memories/pg/reads.py b/hindsight-api-slim/hindsight_api/engine/memories/pg/reads.py index 764c591b99..88ea885e9f 100644 --- a/hindsight-api-slim/hindsight_api/engine/memories/pg/reads.py +++ b/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. """ diff --git a/hindsight-api-slim/hindsight_api/engine/memories/pg/writes.py b/hindsight-api-slim/hindsight_api/engine/memories/pg/writes.py index 67e742d071..2c62a97209 100644 --- a/hindsight-api-slim/hindsight_api/engine/memories/pg/writes.py +++ b/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, diff --git a/hindsight-api-slim/hindsight_api/engine/memory_engine.py b/hindsight-api-slim/hindsight_api/engine/memory_engine.py index faff95f494..dc0e0936b1 100644 --- a/hindsight-api-slim/hindsight_api/engine/memory_engine.py +++ b/hindsight-api-slim/hindsight_api/engine/memory_engine.py @@ -7406,7 +7406,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, @@ -7461,6 +7462,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")} @@ -8125,7 +8129,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, @@ -8255,6 +8260,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")} @@ -8320,7 +8326,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 diff --git a/hindsight-api-slim/hindsight_api/engine/transfer/importer.py b/hindsight-api-slim/hindsight_api/engine/transfer/importer.py index 1d667a66d0..82c974f769 100644 --- a/hindsight-api-slim/hindsight_api/engine/transfer/importer.py +++ b/hindsight-api-slim/hindsight_api/engine/transfer/importer.py @@ -865,6 +865,11 @@ async def _restore_fact_lifecycle( when present (mirroring the document-row handling); ``consolidated_at`` / ``consolidation_failed_at`` are set verbatim — a source-``NULL`` (unconsolidated) fact stays eligible, which is correct. + + No ``updated_at`` stamp (see :data:`~..memories.base.META_UPDATED_AT`): this fixup + runs in the same transaction as the insert that created the row, so the column + already carries this transaction's timestamp. The same holds for the observation + fixups below. """ rows: list[tuple[uuid.UUID, datetime | None, datetime | None, datetime | None]] = [] for original_index, fact in enumerate(facts): diff --git a/hindsight-api-slim/tests/test_memory_units_updated_at.py b/hindsight-api-slim/tests/test_memory_units_updated_at.py new file mode 100644 index 0000000000..7e0e67dc5a --- /dev/null +++ b/hindsight-api-slim/tests/test_memory_units_updated_at.py @@ -0,0 +1,252 @@ +"""``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, the mental-model staleness check and recall's own +``created_after`` / ``created_before`` window, so a write path that changes a +memory without stamping it makes the chase silently skip that change. Two did: +the document tag propagation and ``set_invalidation_reason``. + +The deliberate exception is the consolidation scheduler's own 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. That exemption is +the *situation*, not the two columns — ``restore_memory`` clears the same markers +as part of a real change and does stamp. All three halves are pinned here. + +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 contextlib import asynccontextmanager +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) + + +@asynccontextmanager +async def _bank(memory: MemoryEngine, slug: str, request_context: RequestContext): + """Provision a throwaway bank and drop it even when an assertion fails.""" + bank_id = f"test-mu-updated-{slug}-{uuid.uuid4().hex[:8]}" + await memory.get_bank_profile(bank_id=bank_id, request_context=request_context) + try: + yield bank_id + finally: + await memory.delete_bank(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.""" + async with _bank(memory, "tags", request_context) as bank_id: + 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 + + @pytest.mark.asyncio + async def test_embedding_write_stamps_updated_at(self, memory: MemoryEngine, request_context: RequestContext): + """The stored vector is part of the memory, so the store method stamps on its own. + + Its in-tree callers (curation edit, revert) already stamp through ``apply_edit`` / + ``restore_memory`` in the same transaction, so this pins the store's own contract + rather than a reachable gap — a store extension may call it standalone. + """ + async with _bank(memory, "embed", request_context) as bank_id: + 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 + + @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.""" + async with _bank(memory, "reason", request_context) as bank_id: + 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 + + @pytest.mark.asyncio + async def test_revert_stamps_updated_at_though_it_clears_the_markers( + self, memory: MemoryEngine, request_context: RequestContext + ): + """The exemption is the scheduler's pass, not the two columns. + + Reverting an archived memory runs ``restore_memory``, which clears + ``consolidated_at`` / ``consolidation_failed_at`` in the same statement that brings + the row back — and bringing it back is an edit, so it stamps. + """ + async with _bank(memory, "revert", request_context) as bank_id: + 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) + + with ( + patch.object(memory, "submit_async_consolidation", new=AsyncMock()), + patch.object(memory, "submit_async_graph_maintenance", new=AsyncMock()), + ): + await memory.update_memory_unit( + bank_id, str(mem_id), state="invalidated", request_context=request_context + ) + async with pool.acquire() as conn: + # The archive row is copied from the live one, so it inherits the baseline. + assert await _updated_at(conn, mem_id, "invalidated_memory_units") == _BASELINE + + await memory.update_memory_unit(bank_id, str(mem_id), state="valid", request_context=request_context) + + async with pool.acquire() as conn: + 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 + + +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.""" + async with _bank(memory, "mark", request_context) as bank_id: + 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 + + @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.""" + async with _bank(memory, "requeue", request_context) as bank_id: + 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