fix(migration): backsweep orphaned observation memory units (#584)

* 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

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

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()