From a30ade4f75cca5d13396e49c58355c652960fce4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Mon, 3 Aug 2026 18:16:25 +0200 Subject: [PATCH 1/2] fix(consolidation): skip observation_history when UPDATE matches 0 rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 code then fell through to _append_observation_history, whose INSERT carries an observation_id FK onto memory_units — raising ForeignKeyViolationError, a (correctly) non-retryable integrity failure that marked the whole consolidation op failed for a row that simply no longer exists. Capture the UPDATE status in the SQL branch and bail out (return None) before the history append when 0 rows matched. The store/upsert branch cannot hit the 0-row case, so it needs no guard. The Oracle wrapper reshapes rowcount into the same "UPDATE " form, so the parse is dialect-safe (mirrors config_resolver). Adds mock-level regression tests covering both the 0-row bail and the positive (rowcount==1) control. --- .../engine/consolidation/consolidator.py | 21 ++- .../test_integrity_violation_not_retried.py | 144 +++++++++++++++++- 2 files changed, 163 insertions(+), 2 deletions(-) diff --git a/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py b/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py index 17dc6d741a..e663ed7fa1 100644 --- a/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py +++ b/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( + update_status = await conn.execute( f""" UPDATE {fq_table("memory_units")} SET text = $1, @@ -2270,6 +2270,25 @@ 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. (The Oracle wrapper reshapes + # rowcount into the same "UPDATE " form, so this parse is dialect-safe.) + updated_rows = ( + int(update_status.split()[-1]) + if isinstance(update_status, str) and update_status.startswith("UPDATE") + else 0 + ) + 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 diff --git a/hindsight-api-slim/tests/test_integrity_violation_not_retried.py b/hindsight-api-slim/tests/test_integrity_violation_not_retried.py index d0c31712a7..0b0eb6e24b 100644 --- a/hindsight-api-slim/tests/test_integrity_violation_not_retried.py +++ b/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,146 @@ 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 = AsyncMock(return_value="UPDATE 0") # 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 = AsyncMock(return_value="UPDATE 1") # 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.asyncio async def test_non_integrity_error_still_retried(memory): """ From e51e0811388ecd89d70c26df7bf5fdcdeac251ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Tue, 4 Aug 2026 15:44:04 +0200 Subject: [PATCH 2/2] refactor(db): add execute_rows_affected primitive; use it for the 0-row guard Move the command-tag rowcount parse out of consolidation business logic and into the pg/oracle connection layer. DatabaseConnection.execute_rows_affected runs a DML statement and returns a plain int, normalizing the dialect-divergent result shape the same way parse_json normalizes JSON columns: 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. _execute_update_action now calls conn.execute_rows_affected(...) and checks the int directly instead of hand-parsing an "UPDATE " string. Adds a parser unit test covering the tag shapes both dialects emit. --- .../engine/consolidation/consolidator.py | 10 ++---- .../hindsight_api/engine/db/base.py | 17 ++++++++++ .../test_integrity_violation_not_retried.py | 34 +++++++++++++++++-- 3 files changed, 51 insertions(+), 10 deletions(-) diff --git a/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py b/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py index e663ed7fa1..fc0906d693 100644 --- a/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py +++ b/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: - update_status = await conn.execute( + updated_rows = await conn.execute_rows_affected( f""" UPDATE {fq_table("memory_units")} SET text = $1, @@ -2276,13 +2276,7 @@ async def _execute_update_action( # 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. (The Oracle wrapper reshapes - # rowcount into the same "UPDATE " form, so this parse is dialect-safe.) - updated_rows = ( - int(update_status.split()[-1]) - if isinstance(update_status, str) and update_status.startswith("UPDATE") - else 0 - ) + # op for a row that simply no longer exists. if updated_rows == 0: logger.debug( f"Update skipped: observation {observation_id} no longer exists " diff --git a/hindsight-api-slim/hindsight_api/engine/db/base.py b/hindsight-api-slim/hindsight_api/engine/db/base.py index 24fb03b52c..ab1b553945 100644 --- a/hindsight-api-slim/hindsight_api/engine/db/base.py +++ b/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 "`` / ``"DELETE "`` 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. diff --git a/hindsight-api-slim/tests/test_integrity_violation_not_retried.py b/hindsight-api-slim/tests/test_integrity_violation_not_retried.py index 0b0eb6e24b..752a8a600a 100644 --- a/hindsight-api-slim/tests/test_integrity_violation_not_retried.py +++ b/hindsight-api-slim/tests/test_integrity_violation_not_retried.py @@ -255,7 +255,7 @@ async def test_update_action_bails_when_observation_row_missing(): source_ids = [uuid.uuid4()] conn = AsyncMock() - conn.execute = AsyncMock(return_value="UPDATE 0") # 0 rows matched + conn.execute_rows_affected = AsyncMock(return_value=0) # observation row gone → 0 rows matched conn.transaction = MagicMock(return_value=_AsyncNullCtx(None)) append_mock = AsyncMock() @@ -285,7 +285,7 @@ async def test_update_action_writes_history_when_row_present(): source_ids = [uuid.uuid4()] conn = AsyncMock() - conn.execute = AsyncMock(return_value="UPDATE 1") # 1 row matched + conn.execute_rows_affected = AsyncMock(return_value=1) # observation row present → 1 row matched conn.transaction = MagicMock(return_value=_AsyncNullCtx(None)) memory_engine = MagicMock() @@ -308,6 +308,36 @@ async def test_update_action_writes_history_when_row_present(): 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 " — 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): """