diff --git a/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py b/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py index 4f331149e1..2dcce2ab39 100644 --- a/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py +++ b/hindsight-api-slim/hindsight_api/engine/consolidation/consolidator.py @@ -28,6 +28,7 @@ from itertools import combinations from typing import TYPE_CHECKING, Any, Literal +import asyncpg from pydantic import BaseModel, field_validator from ...config import get_config @@ -1775,15 +1776,22 @@ async def _append_observation_history( history from growing without bound. """ obs_uuid = uuid.UUID(observation_id) - await conn.execute( - f""" + try: + await conn.execute( + f""" INSERT INTO {fq_table("observation_history")} (observation_id, bank_id, content, changed_at) VALUES ($1, $2, $3::jsonb, now()) """, - obs_uuid, - bank_id, - json.dumps(asdict(snapshot)), - ) + obs_uuid, + bank_id, + json.dumps(asdict(snapshot)), + ) + except asyncpg.exceptions.ForeignKeyViolationError: + logger.warning( + f"FK violation writing observation_history for {observation_id}: " + "observation was removed before history could be written (race with parallel consolidation). Skipping." + ) + return if max_entries and max_entries > 0: await conn.execute( f""" diff --git a/hindsight-api-slim/tests/test_observation_history_fk_race.py b/hindsight-api-slim/tests/test_observation_history_fk_race.py new file mode 100644 index 0000000000..00dcde2731 --- /dev/null +++ b/hindsight-api-slim/tests/test_observation_history_fk_race.py @@ -0,0 +1,48 @@ +"""Regression: observation_history write for a since-deleted observation is skipped. + +Under parallel (or same-batch delete-then-update) consolidation, one write path +can remove an observation from ``memory_units`` before another writes its +``observation_history`` snapshot. The INSERT then trips +``observation_history_observation_id_fkey``. Because consolidation runs in +autocommit (no enclosing transaction), catching the FK violation and skipping +the best-effort history row is safe — the connection stays usable and the +current observation state remains the source of truth. + +Regression for #2597 / #2506: before the fix this raised +``asyncpg.ForeignKeyViolationError`` and failed the whole consolidation task. +""" + +import uuid + +import pytest + +from hindsight_api.engine.consolidation.consolidator import ( + _append_observation_history, + _ObservationHistorySnapshot, +) +from hindsight_api.engine.db_utils import acquire_with_retry + + +@pytest.mark.asyncio +async def test_append_history_for_missing_observation_is_skipped(memory, request_context): + """A history write targeting an absent observation is skipped, not fatal.""" + bank_id = f"test-obs-history-fk-{uuid.uuid4().hex[:8]}" + await memory.get_bank_profile(bank_id, request_context=request_context) + + snapshot = _ObservationHistorySnapshot( + previous_text="old", + previous_tags=[], + previous_occurred_start=None, + previous_occurred_end=None, + previous_mentioned_at=None, + new_source_memory_ids=[], + ) + # Never inserted into memory_units, so the FK target is absent. + missing_observation_id = str(uuid.uuid4()) + + pool = await memory._get_pool() + async with acquire_with_retry(pool) as conn: + # Pre-fix this raised asyncpg.ForeignKeyViolationError; the fix skips it. + await _append_observation_history(conn, bank_id, missing_observation_id, snapshot, max_entries=10) + # Autocommit: the failed INSERT did not poison the connection. + assert await conn.fetchval("SELECT 1") == 1