From f09ad9deac7b02cdce29f2dc87d46c8778f525e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Mon, 16 Mar 2026 14:06:33 +0100 Subject: [PATCH] fix(migration): backsweep orphaned observation memory units (#584) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(migration): backsweep orphaned observation memory units Delete observation rows whose every source_memory_id points to a deleted memory unit, left behind before PR #580 fixed the chunk FK cascade and before delete_document() called _delete_stale_observations_for_memories. Closes #572 (data cleanup for pre-existing installs). * fix(migration): broaden backsweep to cover all fact types and bank-level orphans - Pass 1: delete any memory_units row (all fact_types) whose bank_id no longer exists in banks — catches orphans from bank deletions that predate a FK cascade between the two tables. - Pass 2: delete observation rows whose every source_memory_id points to a deleted memory unit, regardless of document_id/chunk_id anchors. * test(migration): verify backsweep removes orphans and preserves legit rows Adds a focused migration test that: - Starts a fresh pg0 instance at revision f6g7h8i9j0k1 - Seeds orphaned rows for both backsweep passes (ghost-bank + all-dead-sources) - Seeds legitimate rows that must survive - Applies the backsweep migration to head - Asserts the expected rows are deleted/preserved --- ...8i9j0k1l2_backsweep_orphan_observations.py | 71 ++++++++ .../tests/test_migration_backsweep.py | 154 ++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 hindsight-api-slim/hindsight_api/alembic/versions/g7h8i9j0k1l2_backsweep_orphan_observations.py create mode 100644 hindsight-api-slim/tests/test_migration_backsweep.py diff --git a/hindsight-api-slim/hindsight_api/alembic/versions/g7h8i9j0k1l2_backsweep_orphan_observations.py b/hindsight-api-slim/hindsight_api/alembic/versions/g7h8i9j0k1l2_backsweep_orphan_observations.py new file mode 100644 index 00000000..7f6de36b --- /dev/null +++ b/hindsight-api-slim/hindsight_api/alembic/versions/g7h8i9j0k1l2_backsweep_orphan_observations.py @@ -0,0 +1,71 @@ +"""backsweep_orphan_memory_units + +Two-pass cleanup of memory_units rows that were never removed by earlier bugs: + +Pass 1 — any fact_type, bank gone: + memory_units whose bank_id no longer exists in banks. These accumulate when + a bank is deleted without a proper cascade (no FK from memory_units to banks + exists in the schema). + +Pass 2 — observations only, all sources gone: + observation rows whose bank still exists but every source_memory_id points + to a deleted memory unit. These were left behind before PR #580 fixed the + chunk FK cascade and before delete_document() called + _delete_stale_observations_for_memories. + +Revision ID: g7h8i9j0k1l2 +Revises: f6g7h8i9j0k1 +Create Date: 2026-03-16 +""" + +from collections.abc import Sequence + +from alembic import context, op + +revision: str = "g7h8i9j0k1l2" +down_revision: str | Sequence[str] | None = "f6g7h8i9j0k1" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def _get_schema_prefix() -> str: + schema = context.config.get_main_option("target_schema") + return f'"{schema}".' if schema else "" + + +def upgrade() -> None: + schema = _get_schema_prefix() + mu = f"{schema}memory_units" + banks = f"{schema}banks" + + # Pass 1: delete all memory_units (any fact_type) whose bank no longer exists. + # There is no FK from memory_units to banks, so these never cascade away. + op.execute( + f""" + DELETE FROM {mu} + WHERE NOT EXISTS ( + SELECT 1 FROM {banks} b WHERE b.bank_id = {mu}.bank_id + ) + """ + ) + + # Pass 2: delete orphaned observations whose bank still exists but every + # source_memory_id refers to a now-deleted memory unit (or the array is + # empty). Observations with at least one surviving source are left alone. + op.execute( + f""" + DELETE FROM {mu} orphan + WHERE orphan.fact_type = 'observation' + AND NOT EXISTS ( + SELECT 1 + FROM {mu} src + WHERE src.id = ANY(orphan.source_memory_ids) + AND src.bank_id = orphan.bank_id + ) + """ + ) + + +def downgrade() -> None: + # Deleted rows cannot be restored. + pass diff --git a/hindsight-api-slim/tests/test_migration_backsweep.py b/hindsight-api-slim/tests/test_migration_backsweep.py new file mode 100644 index 00000000..4973a686 --- /dev/null +++ b/hindsight-api-slim/tests/test_migration_backsweep.py @@ -0,0 +1,154 @@ +"""Tests for migration g7h8i9j0k1l2 (backsweep orphaned memory_units). + +Uses a dedicated pg0 instance (port 5562) so the test can control exactly +which migrations have run before inserting the orphan seed data. +""" + +import asyncio +import uuid +from pathlib import Path + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, text + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_SCRIPT_LOCATION = str(Path(__file__).parent.parent / "hindsight_api" / "alembic") + + +def _alembic_cfg(db_url: str) -> Config: + cfg = Config() + cfg.set_main_option("script_location", _SCRIPT_LOCATION) + cfg.set_main_option("sqlalchemy.url", db_url) + cfg.set_main_option("prepend_sys_path", ".") + cfg.set_main_option("path_separator", "os") + return cfg + + +def _upgrade(db_url: str, revision: str) -> None: + command.upgrade(_alembic_cfg(db_url), revision) + + +# --------------------------------------------------------------------------- +# Fixture: fresh database at the revision just before the backsweep +# --------------------------------------------------------------------------- + +@pytest.fixture(scope="module") +def pre_backsweep_db_url(): + """ + Spin up a dedicated pg0 instance and run all migrations up to (but not + including) the backsweep revision so each test can seed orphan data and + then apply the backsweep itself. + """ + from hindsight_api.pg0 import EmbeddedPostgres + + pg0 = EmbeddedPostgres(name="hindsight-backsweep-test", port=5562) + loop = asyncio.new_event_loop() + try: + url = loop.run_until_complete(pg0.ensure_running()) + finally: + loop.close() + + # Migrate up to the revision just before the backsweep. + _upgrade(url, "f6g7h8i9j0k1") + return url + + +# --------------------------------------------------------------------------- +# The test +# --------------------------------------------------------------------------- + +def test_backsweep_removes_orphans_and_preserves_legit_rows(pre_backsweep_db_url): + """ + Seed four kinds of rows then apply the backsweep migration and verify: + + Rows that MUST be deleted + ───────────────────────── + A. Any fact_type, bank_id missing from banks + → Pass 1 deletes these regardless of fact_type or source links. + + B. observation, bank exists, but ALL source_memory_ids are gone + → Pass 2 deletes these. + + Rows that MUST survive + ────────────────────── + C. observation, bank exists, at least ONE source_memory_id still live + → Pass 2 must not touch these. + + D. Non-observation (world), bank exists, no sources (not relevant) + → Pass 1 must not touch these (bank exists). + """ + db_url = pre_backsweep_db_url + engine = create_engine(db_url) + + alive_bank = f"bank_{uuid.uuid4().hex[:8]}" + ghost_bank = f"bank_{uuid.uuid4().hex[:8]}" # never inserted into banks + + # UUIDs for memory units + id_pass1_world = uuid.uuid4() # A: world unit, ghost bank + id_pass1_obs = uuid.uuid4() # A: observation, ghost bank + id_pass2_obs = uuid.uuid4() # B: observation, all sources gone + id_keep_obs = uuid.uuid4() # C: observation with one live source + id_keep_world = uuid.uuid4() # D: world unit, alive bank + id_live_source = uuid.uuid4() # live source for C + + with engine.connect() as conn: + # --- banks --- + conn.execute(text("INSERT INTO banks (bank_id) VALUES (:b)"), {"b": alive_bank}) + + # --- seed memory_units --- + def insert_mu(uid, bank, fact_type, sources=None): + src_arr = "{" + ",".join(str(s) for s in (sources or [])) + "}" + conn.execute( + text( + """ + INSERT INTO memory_units + (id, bank_id, text, fact_type, source_memory_ids) + VALUES + (:id, :bank, :text, :ft, CAST(:src AS uuid[])) + """ + ), + {"id": uid, "bank": bank, "text": "test", "ft": fact_type, "src": src_arr}, + ) + + # A: ghost-bank rows (Pass 1 targets) + insert_mu(id_pass1_world, ghost_bank, "world") + insert_mu(id_pass1_obs, ghost_bank, "observation", sources=[uuid.uuid4()]) + + # B: observation with all-dead sources (Pass 2 target) + insert_mu(id_pass2_obs, alive_bank, "observation", sources=[uuid.uuid4(), uuid.uuid4()]) + + # C: observation with one live source (must survive) + insert_mu(id_live_source, alive_bank, "world") + insert_mu(id_keep_obs, alive_bank, "observation", sources=[id_live_source, uuid.uuid4()]) + + # D: world unit in alive bank (must survive) + insert_mu(id_keep_world, alive_bank, "world") + + conn.commit() + + # --- apply the backsweep --- + _upgrade(db_url, "g7h8i9j0k1l2") + + # --- verify --- + with engine.connect() as conn: + def exists(uid): + return conn.execute( + text("SELECT 1 FROM memory_units WHERE id = :id"), {"id": uid} + ).fetchone() is not None + + # Must be gone + assert not exists(id_pass1_world), "Pass 1: world unit with ghost bank should be deleted" + assert not exists(id_pass1_obs), "Pass 1: observation with ghost bank should be deleted" + assert not exists(id_pass2_obs), "Pass 2: observation with all-dead sources should be deleted" + + # Must survive + assert exists(id_keep_obs), "observation with a live source must not be deleted" + assert exists(id_keep_world), "world unit in alive bank must not be deleted" + assert exists(id_live_source), "live source memory unit must not be deleted" + + engine.dispose()