fix(consolidation): guard observation_history append on 0-row observation UPDATE

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

@@ -2246,7 +2246,7 @@ async def _execute_update_action(
 
             t0 = time.time()
             if store.writes_memory_rows_in_sql:
-                await conn.execute(
+                updated_rows = await conn.execute_rows_affected(
                     f"""
                     UPDATE {fq_table("memory_units")}
                     SET text = $1,
@@ -2270,6 +2270,19 @@ async def _execute_update_action(
                     source_mentioned_at,
                     merged_tags,
                 )
+                # The source-liveness checks above guard the *source* memories; the
+                # observation row itself (WHERE id = $5) can still be invalidated/deleted
+                # concurrently, matching 0 rows. Bail out BEFORE the observation_history
+                # INSERT below — that INSERT carries an observation_id FK onto memory_units,
+                # so appending history for a now-missing row raises ForeignKeyViolationError,
+                # a non-retryable integrity failure that would fail the whole consolidation
+                # op for a row that simply no longer exists.
+                if updated_rows == 0:
+                    logger.debug(
+                        f"Update skipped: observation {observation_id} no longer exists "
+                        "(deleted/invalidated concurrently); not appending history"
+                    )
+                    return None
             else:
                 # Upsert overwrites the whole observation, so start from its current state (fetched
                 # from the store) and apply the same merge the SQL does — LEAST/GREATEST on the times

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

@@ -112,6 +112,23 @@ async def execute(self, query: str, *args: Any, timeout: float | None = None) ->
         """
         ...
 
+    async def execute_rows_affected(self, query: str, *args: Any, timeout: float | None = None) -> int:
+        """Execute a DML statement and return the number of rows it affected.
+
+        Normalizes the dialect-specific execute result into a plain int so callers
+        never hand-parse an ``"UPDATE <n>"`` / ``"DELETE <n>"`` command tag in
+        business logic (mirrors ``parse_json`` above, which normalizes the other
+        dialect-divergent result shape). asyncpg returns the tag directly; the
+        Oracle connection reshapes ``cursor.rowcount`` into the same trailing-count
+        form, so parsing the last token is dialect-safe. Returns 0 when the status
+        has no trailing count (e.g. a non-DML statement).
+        """
+        status = await self.execute(query, *args, timeout=timeout)
+        if not isinstance(status, str):
+            return 0
+        parts = status.split()
+        return int(parts[-1]) if parts and parts[-1].isdigit() else 0
+
     @abstractmethod
     async def executemany(self, query: str, args: list[tuple[Any, ...]], *, timeout: float | None = None) -> None:
         """Execute a query for each set of arguments.

hindsight-api-slim/tests/test_integrity_violation_not_retried.py

@@ -14,7 +14,9 @@
 
 import json
 import uuid
-from unittest.mock import AsyncMock, patch
+from contextlib import ExitStack
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import asyncpg
 import pytest
@@ -166,6 +168,176 @@ def test_invalid_embedding_dimension_error_is_non_retryable(message):
     assert _is_non_retryable_task_error(RuntimeError(message)) is True
 
 
+class _AsyncNullCtx:
+    """Async context manager that yields a preset value — stands in for
+    ``acquire_with_retry(pool)`` (yields a conn) and ``conn.transaction()``
+    (yields None) without touching a database. Re-entrant across sequential
+    ``async with`` blocks because __aenter__/__aexit__ hold no state."""
+
+    def __init__(self, value=None):
+        self._value = value
+
+    async def __aenter__(self):
+        return self._value
+
+    async def __aexit__(self, *exc):
+        return False
+
+
+def _fake_config() -> SimpleNamespace:
+    """Minimal config for _execute_update_action: ``text_search_extension='none'``
+    so no native tsvector clause is emitted, and observation-history enabled so
+    the 0-row guard is the only thing preventing the (FK-violating) history INSERT."""
+    return SimpleNamespace(
+        text_search_extension="none",
+        text_search_extension_native_language="english",
+        enable_observation_history=True,
+        observation_history_max_entries=10,
+    )
+
+
+def _observation_fact(observation_id: str):
+    from hindsight_api.engine.response_models import MemoryFact
+
+    return MemoryFact(
+        id=observation_id,
+        text="old observation text",
+        fact_type="observation",
+        tags=["scope_a"],
+        source_fact_ids=[],
+    )
+
+
+def _patch_update_action_deps(consolidator, conn, source_ids, append_mock) -> ExitStack:
+    """Enter the common patch set for the two _execute_update_action guard tests
+    and return the live ExitStack (use as ``with _patch_update_action_deps(...):``).
+
+    Stubs the pool/transaction acquisition, the (slow) embedder, the source
+    liveness checks (both preflight and the FOR SHARE recheck), the store
+    capability flag, config, and the history append — leaving the UPDATE
+    rowcount as the single variable under test.
+    """
+    store = SimpleNamespace(writes_memory_rows_in_sql=True)
+    stack = ExitStack()
+    stack.enter_context(patch("hindsight_api.config.get_config", _fake_config))
+    stack.enter_context(patch.object(consolidator, "acquire_with_retry", MagicMock(return_value=_AsyncNullCtx(conn))))
+    stack.enter_context(patch.object(consolidator, "get_memories", MagicMock(return_value=store)))
+    stack.enter_context(patch.object(consolidator, "_any_live_source_memory", AsyncMock(return_value=True)))
+    stack.enter_context(patch.object(consolidator, "_filter_live_source_memories", AsyncMock(return_value=source_ids)))
+    stack.enter_context(
+        patch.object(
+            consolidator.embedding_utils,
+            "generate_embeddings_batch",
+            AsyncMock(return_value=[[0.1, 0.2, 0.3]]),
+        )
+    )
+    stack.enter_context(patch.object(consolidator, "_append_observation_history", append_mock))
+    return stack
+
+
+@pytest.mark.asyncio
+async def test_update_action_bails_when_observation_row_missing():
+    """
+    Regression: the source-liveness checks in ``_execute_update_action`` guard the
+    *source* memories, but the observation row itself (``UPDATE ... WHERE id = $5``)
+    can be concurrently invalidated/deleted, matching 0 rows. The prior code ignored
+    the rowcount and still called ``_append_observation_history``, whose INSERT carries
+    an ``observation_id`` FK onto memory_units — raising ForeignKeyViolationError, a
+    non-retryable integrity failure that fails the whole consolidation op for a row
+    that no longer exists.
+
+    The fix returns None on rowcount==0 BEFORE the history append. Assert the guard
+    fires and no history is written.
+    """
+    from hindsight_api.engine.consolidation import consolidator
+
+    observation_id = str(uuid.uuid4())
+    source_ids = [uuid.uuid4()]
+
+    conn = AsyncMock()
+    conn.execute_rows_affected = AsyncMock(return_value=0)  # observation row gone → 0 rows matched
+    conn.transaction = MagicMock(return_value=_AsyncNullCtx(None))
+
+    append_mock = AsyncMock()
+    with _patch_update_action_deps(consolidator, conn, source_ids, append_mock):
+        result = await consolidator._execute_update_action(
+            pool=MagicMock(),
+            memory_engine=MagicMock(),
+            bank_id="bank-x",
+            source_memory_ids=source_ids,
+            observation_id=observation_id,
+            new_text="new observation text",
+            observations=[_observation_fact(observation_id)],
+            source_fact_tags=["scope_b"],
+        )
+
+    assert result is None, "Expected None when the observation row no longer exists"
+    append_mock.assert_not_called()  # the FK-violating INSERT must be skipped
+
+
+@pytest.mark.asyncio
+async def test_update_action_writes_history_when_row_present():
+    """Positive control: when the UPDATE matches a row (rowcount==1), the guard must
+    NOT interfere — observation_history is appended and the embedding is returned."""
+    from hindsight_api.engine.consolidation import consolidator
+
+    observation_id = str(uuid.uuid4())
+    source_ids = [uuid.uuid4()]
+
+    conn = AsyncMock()
+    conn.execute_rows_affected = AsyncMock(return_value=1)  # observation row present → 1 row matched
+    conn.transaction = MagicMock(return_value=_AsyncNullCtx(None))
+
+    memory_engine = MagicMock()
+    memory_engine._backend.ops.uses_observation_sources_table = False
+
+    append_mock = AsyncMock()
+    with _patch_update_action_deps(consolidator, conn, source_ids, append_mock):
+        result = await consolidator._execute_update_action(
+            pool=MagicMock(),
+            memory_engine=memory_engine,
+            bank_id="bank-x",
+            source_memory_ids=source_ids,
+            observation_id=observation_id,
+            new_text="new observation text",
+            observations=[_observation_fact(observation_id)],
+            source_fact_tags=["scope_b"],
+        )
+
+    assert result is not None, "Expected the embedding string back on a successful update"
+    append_mock.assert_called_once()
+
+
+@pytest.mark.parametrize(
+    "status, expected",
+    [
+        ("UPDATE 0", 0),
+        ("UPDATE 3", 3),
+        ("DELETE 5", 5),
+        ("INSERT 0 1", 1),  # PG insert tags are "INSERT <oid> <count>" — count is the last token
+        ("SELECT 7", 7),
+        ("OK", 0),  # non-DML / no trailing count
+        (None, 0),  # defensive: non-string status
+    ],
+)
+@pytest.mark.asyncio
+async def test_execute_rows_affected_parses_command_tag(status, expected):
+    """``DatabaseConnection.execute_rows_affected`` centralizes command-tag parsing
+    so callers get a plain int. Verify the trailing-count parse across the tag
+    shapes both dialects produce (asyncpg directly; Oracle reshapes cursor.rowcount
+    into the same form)."""
+    from hindsight_api.engine.db.base import DatabaseConnection
+
+    class _FakeConn:
+        async def execute(self, *args, **kwargs):
+            return status
+
+    # Call the real method with a minimal stub self — avoids implementing every
+    # abstract method just to exercise the concrete parser.
+    rows = await DatabaseConnection.execute_rows_affected(_FakeConn(), "UPDATE t SET x=1")
+    assert rows == expected
+
+
 @pytest.mark.asyncio
 async def test_non_integrity_error_still_retried(memory):
     """