brain-of-reese/tests/integration/test_prune_cascade_statements.py

"""Integration: phase 183 — what a mass prune ACTUALLY loads (the probe).

**VERDICT (task 01): CONFIRMED — one chunk SELECT per pruned document →
FIXED (task 02, ``passive_deletes=True``).** Pruning 200 documents (25
chunks each, a real 768-dim ``chunks.embedding`` per chunk) through the
importer's own ``_prune`` used to issue **exactly 200** ``SELECT … FROM
chunks`` statements — one lazy child-collection load per
``session.delete(doc)`` — materialising every 768-dim embedding of every
document into Python on its way to the delete. Task 01 pinned that with
``EXPECTED_CHUNK_SELECTS_PER_PRUNED_DOC = 1`` GREEN before any code change
(the phase-179 task-01 "pin the defect with a probe first" precedent);
task 02 declared ``passive_deletes=True`` on
``Document.chunks`` and flipped the constant to ``0``. The same fixture
now prunes in **2 statements** (the prune's own ``SELECT documents`` plus
one batched ``DELETE FROM documents``), **0 chunk SELECTs**, **0 orphan
chunk rows** — the ``chunks_document_id_fkey`` ``ON DELETE CASCADE``
drops the rows in the database, so Python never sees them.

:func:`test_chunks_fk_cascades_at_database_level` is the other half of the
contract: ``passive_deletes`` is only correct while the FK really
cascades, so the live ``pg_constraint`` row is asserted here (a future
migration dropping ``ON DELETE CASCADE`` would otherwise let every prune
silently orphan chunk rows).

**The incident (owner, 2026-10-15).** The brain-of-code sync ``started
repos=4`` at 11:19:23, reached ``_prune`` at 12:14:14, pruned **3,008
litellm docs** — and the VM locked up so hard that boot −1 has no shutdown
record at all (4 GB of RAM). The app cgroup had already peaked at 3.3 GB
twice that same week. The box has 3,904 MB and **no swap**, and shares that
RAM with a 3.1 GB Postgres. 3,008 docs × ~24 chunks ≈ 72k rows whose
embeddings the prune pulled into Python to delete them.

**Mechanism (pre-phase-183).** ``Document.chunks`` carried
``cascade="all, delete-orphan"`` and **no ``passive_deletes``** — so
``session.delete(doc)`` had to LOAD the child collection first (the ORM has
to know which children to orphan/delete), SELECTing every chunk row
*including* its ``Vector(768)`` column. Nothing was released afterwards:
``app/db.py:30`` builds sessions with ``expire_on_commit=False``, so the 72k
hydrated ``Chunk`` objects (and their float lists) stayed resident until the
single ``session.commit()`` at ``app/rag/importer.py`` returned — the
``passive_deletes=True`` of task 02 is what takes that walk out.

**Measured unit costs.**

* A loaded 768-dim ``Chunk.embedding`` costs **24.8 KiB of Python** (list
  header + 768 float objects — pgvector hands back a ``list[float]``, never
  a buffer) against **3 KiB on disk**: an 8× expansion per embedding.
* **~0.24 MB resident per document deleted** at 25 chunks/doc — measured
  2026-10-15 against the dev DB with production-shaped rows (400 docs ×
  25 chunks, ~745-char chunk content): RSS grew linearly, +241 MB over
  400 deletes.

**Measured BY THIS FILE (2026-10-15, the fixture below: 200 docs × 25
chunks × 745 chars × a 768-float vector, the dev Postgres).**

* PRE-fix (no ``passive_deletes``) statement capture
  (``before_cursor_execute``, the fixture below): **203** statements for
  the whole prune — **200 of them ``SELECT … FROM chunks``** (one per
  pruned document, the load-bearing pin), plus 1 ``SELECT documents`` (the
  prune's own walk) and the 2 ``executemany`` ``DELETE``s (``chunks``, then
  ``documents``).
* PRE-fix ``VmRSS`` delta: **+165.6 MB for the 200-doc prune**
  (≈ 0.83 MB/doc). Higher than the 0.24 MB/doc above because this fixture's
  chunk content and vector are both fully hydrated and the session keeps
  everything (``expire_on_commit=False``); the ORDER is the point, not the
  figure.
* the same 200 deletes issued as plain ``DELETE FROM documents`` (i.e. what
  the DB's own ``ON DELETE CASCADE`` does — the shape task 02 lands,
  measured as a one-off while writing task 01): **0 chunk SELECTs,
  +0.0 MB RSS**, and still **0 orphan chunk rows** — the ceiling below is
  picked from that contrast, roughly one quarter of the pre-fix delta.
* POST-fix (``passive_deletes=True``, task 02, the same fixture and the
  same capture): **2** statements — 1 ``SELECT documents`` + 1 batched
  ``executemany`` ``DELETE FROM documents`` — **0 chunk SELECTs**, and the
  ``VmRSS`` delta is **+0.2 MB** (run-to-run +0.1…0.6 MB — allocator noise,
  two orders of magnitude under the 165.6 MB the hydrating walk cost, and
  still 200× inside the 40 MB ceiling). Still **0
  orphan chunk rows**, which is what the D4 assertions inside
  :func:`test_prune_never_selects_a_chunk_row` and the drift guard
  :func:`test_chunks_fk_cascades_at_database_level` exist to keep true.

**Why the memory test branches on the constant.** D1/LOCKED A4: the
statement count is the load-bearing, CI-safe pin; the RSS figure is
corroboration and must never flake a build on allocator noise. So
``test_prune_memory_delta_has_a_ceiling`` asserts the post-fix ceiling
(``< 40 MB``); its ``else`` branch, kept for the pre-fix state the constant
can be flipped back to, asserts the opposite (the delta lands WELL ABOVE
that ceiling) — which is how the ceiling was proven to have teeth before it
became the contract.

**What this file does NOT touch:** the prune's batching (phase 184), the
fast-path column set (phase 185),
``expire_on_commit``, and the ``doc.chunks = [...]`` orphan-replacement
path inside ``_replace_content_chunks`` (``app/rag/importer.py:1033``) —
``passive_deletes`` does not cover collection reassignment and a re-embed
genuinely needs the loaded collection (phase 183 LOCKED A2).
"""
from __future__ import annotations

import gc
import re
from collections.abc import Iterator
from pathlib import Path
from uuid import UUID

import pytest
from sqlalchemy import event, func, select, text
from sqlalchemy.orm import Session

from app.db import engine
from app.models import Chunk, Document
from app.rag.importer import ImportSummary, _prune

#: The probe's own source name — nothing else in the KB uses it.
PROBE_SOURCE = "cascade-probe"

#: The production-shaped fixture: 200 documents, 25 chunks each (the
#: incident pruned 3,008 docs × ~24 chunks; this is that walk scaled down
#: to what a test run can seed in seconds).
DOC_COUNT = 200
CHUNKS_PER_DOC = 25

#: ~745 chars, the measured production chunk size (phase 183 baseline).
CHUNK_CONTENT = ("The homelab runs podman quadlets on the hnb host. " * 18)[:745]

#: **0 post-phase-183 — the FK cascade does the work in the database;
#: 1 pre-phase-183 was the lazy child load** (one ``SELECT … FROM chunks``
#: per deleted document, the waste this phase removed: 5,000 chunk rows and
#: 5,000 768-dim embeddings hydrated into Python purely to be deleted).
#: The DB's ``ON DELETE CASCADE`` (alembic 0001 L54,
#: ``chunks_document_id_fkey`` confdeltype='c' — pinned live by
#: :func:`test_chunks_fk_cascades_at_database_level`) deletes the rows
#: without Python ever seeing them.
EXPECTED_CHUNK_SELECTS_PER_PRUNED_DOC = 0

#: The POST-phase-183 memory ceiling for a 200-doc prune (~1/4 of the
#: measured pre-fix delta of 165.6 MB; the FK-cascade path measures
#: +0.2 MB, module docstring).
RSS_CEILING_MB = 40.0

#: The loose upper bound that still applies while the hydrating behaviour
#: is the expected one — a runaway allocator must never go unnoticed.
PRE_FIX_RSS_SANITY_CEILING_MB = 400.0

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def clean_state(db: Session) -> Iterator[None]:
    """``documents`` / ``chunks`` are global state: reset around every
    test (the ``test_sync_api.py`` truncate pattern)."""
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()
    yield
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()


@pytest.fixture(autouse=True)
def statement_capture() -> Iterator[list[str]]:
    """Every SQL the engine executes for the duration of ONE test.

    Installed on the engine (the session under test binds to it) and
    REMOVED in teardown, so no other test of the session ever sees the
    listener (AGENTS: the suite stays isolated — verified by the full
    ``uv run pytest`` run being green with this file in it).
    """
    captured: list[str] = []

    def _listener(
        conn: object,  # noqa: ARG001
        cursor: object,  # noqa: ARG001
        statement: str,
        parameters: object,  # noqa: ARG001
        context: object,  # noqa: ARG001
        executemany: bool,  # noqa: ARG001
    ) -> None:
        captured.append(statement)

    event.listen(engine, "before_cursor_execute", _listener)
    try:
        yield captured
    finally:
        event.remove(engine, "before_cursor_execute", _listener)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

#: ``SELECT … FROM chunks`` — the child-collection load. The word boundary
#: keeps ``documents`` (and its ``…_id`` aliases) out of the count while
#: still matching the real lazy load, whose SQL starts
#: ``SELECT chunks.id AS chunks_id, … FROM chunks WHERE chunks.document_id``.
_CHUNK_SELECT_RE = re.compile(r"\bfrom\s+chunks\b", re.IGNORECASE)


def chunk_selects(captured: list[str]) -> list[str]:
    """The captured statements that SELECT the ``chunks`` table."""
    return [
        stmt
        for stmt in captured
        if stmt.lstrip().lower().startswith("select") and _CHUNK_SELECT_RE.search(stmt)
    ]


def _seed_docs(session: Session) -> list[UUID]:
    """``DOC_COUNT`` indexed documents of :data:`PROBE_SOURCE`, each with
    ``CHUNKS_PER_DOC`` embedded chunks. Committed; returns the doc ids.

    The chunks are added BY ``document_id`` (never through
    ``doc.chunks``) so the relationship stays UNLOADED — the state a real
    prune starts from, and the only state in which the lazy load is
    observable.
    """
    ids: list[UUID] = []
    for n in range(DOC_COUNT):
        doc = Document(
            source=PROBE_SOURCE,
            path=f"docs/doc-{n:04d}.md",
            full_path=f"/srv/brain/{PROBE_SOURCE}/docs/doc-{n:04d}.md",
            title=f"Cascade probe doc {n}",
            content=CHUNK_CONTENT,
            content_hash=f"{n:064x}",
        )
        session.add(doc)
        session.flush()
        ids.append(doc.id)
        # Deterministic 768-float vector, rotated by the doc index (a real
        # pgvector payload — the point is the HYDRATION cost, so the values
        # only need to be stable).
        rot = n % 768
        embedding = [((i + rot) % 768) / 768 for i in range(768)]
        session.add_all(
            [
                Chunk(
                    document_id=doc.id,
                    position=position,
                    content=CHUNK_CONTENT,
                    embedding=embedding,
                )
                for position in range(CHUNKS_PER_DOC)
            ]
        )
    session.commit()
    return ids


def _run_prune(session: Session) -> tuple[int, ImportSummary]:
    """The prune exactly as ``import_sources`` calls it with a walk that saw
    NOTHING (``seen`` empty → every doc of the source prunes) and the
    images toggle OFF (``app/rag/importer.py:3860``)."""
    summary = ImportSummary()
    pruned = _prune(
        session,
        {PROBE_SOURCE},
        set(),
        images=False,
        summary=summary,
        page_dir=None,
    )
    return pruned, summary


def _vm_rss_mb() -> float | None:
    """``VmRSS`` from ``/proc/self/status`` in MB, ``None`` when unreadable
    (a non-Linux box skips the memory test rather than pinning noise)."""
    try:
        for line in Path("/proc/self/status").read_text().splitlines():
            if line.startswith("VmRSS:"):
                return int(line.split()[1]) / 1024.0
    except OSError:
        return None
    return None


# ---------------------------------------------------------------------------
# The probe
# ---------------------------------------------------------------------------


def test_prune_never_selects_a_chunk_row(db: Session, statement_capture: list[str]) -> None:
    """Zero ``SELECT … FROM chunks`` for 200 pruned documents — the fix.

    POST-phase-183 assertion (``EXPECTED_CHUNK_SELECTS_PER_PRUNED_DOC`` is
    ``0``): ``passive_deletes=True`` means the prune's 200
    ``session.delete(doc)`` calls never hydrate a child collection, so no
    chunk row and no 768-dim embedding reaches Python — the whole walk is 2
    statements. PRE-phase-183 this same fixture measured 200 chunk SELECTs
    and 203 statements (module docstring).

    Every other assertion here (the counts, the ``changed_paths`` feed, and
    above all the ZERO orphan chunk rows) held byte-identically across the
    flip — they exist so the fix cannot pass by quietly stopping to delete
    children (phase 183 D4).
    """
    ids = _seed_docs(db)
    statement_capture.clear()  # the seed's own SQL is not part of the probe

    pruned, summary = _run_prune(db)

    assert pruned == DOC_COUNT
    expected_selects = DOC_COUNT * EXPECTED_CHUNK_SELECTS_PER_PRUNED_DOC
    assert len(chunk_selects(statement_capture)) == expected_selects

    # D4 — the no-orphan contract, asserted BEFORE the fix so that after it
    # "0 orphans" means "unchanged", never "never checked".
    assert db.scalar(
        select(func.count()).select_from(Chunk).where(Chunk.document_id.in_(ids))
    ) == 0
    assert db.scalar(
        select(func.count()).select_from(Document).where(Document.source == PROBE_SOURCE)
    ) == 0

    # The phase-127 changed_paths feed: one (source, path) per pruned doc,
    # appended BEFORE the delete (the folder-summary regeneration input).
    assert sorted(summary.changed_paths) == sorted(
        (PROBE_SOURCE, f"docs/doc-{n:04d}.md") for n in range(DOC_COUNT)
    )


def test_prune_memory_delta_has_a_ceiling(db: Session) -> None:
    """The corroborating memory figure — a ceiling, never the flaky pin.

    Same 200-doc prune, ``VmRSS`` read before and after (``gc.collect()``
    on both sides so the allocator gets its fair say). See the module
    docstring for the measured numbers: **+165.6 MB** while the ORM hydrates
    the child collections (PRE-phase-183), **+0.2 MB** on the FK-cascade
    path (POST, task 02) against the 40 MB ceiling pinned below.
    """
    before = _vm_rss_mb()
    if before is None:
        pytest.skip("/proc/self/status unreadable — the RSS ceiling is Linux-only")

    ids = _seed_docs(db)
    gc.collect()
    baseline = _vm_rss_mb()
    assert baseline is not None

    pruned, _ = _run_prune(db)
    gc.collect()
    after = _vm_rss_mb()
    assert after is not None
    delta = after - baseline

    assert pruned == DOC_COUNT
    assert db.scalar(
        select(func.count()).select_from(Chunk).where(Chunk.document_id.in_(ids))
    ) == 0

    if EXPECTED_CHUNK_SELECTS_PER_PRUNED_DOC == 0:
        # POST-phase-183: no embedding is ever hydrated, so the whole prune
        # must cost well under the ceiling (measured +0.2 MB).
        assert delta < RSS_CEILING_MB, (
            f"VmRSS grew {delta:.1f} MB during a cascade prune — chunk rows"
            " are being hydrated again (passive_deletes removed?)"
        )
    else:
        # PRE-phase-183 (the state the constant can be flipped back to): the
        # hydration is the behaviour under test, so this fixture must land
        # WELL ABOVE that ceiling — a ceiling nothing ever crossed would be
        # a pin in name only.
        assert delta > RSS_CEILING_MB
        assert delta < PRE_FIX_RSS_SANITY_CEILING_MB


def test_chunks_fk_cascades_at_database_level(db: Session) -> None:
    """The drift guard: every ``chunks`` FK really is ``ON DELETE CASCADE``.

    ``passive_deletes=True`` tells the ORM to trust the database with the
    child rows. That trust is only sound while ``chunks_document_id_fkey``
    cascades (alembic 0001 L54; production confirmed
    ``confdeltype='c'``); a future migration that recreates the FK without
    the clause would turn every prune into a silent orphan-generator — the
    documents would vanish, the chunk rows (and their vectors) would stay,
    and nothing in the app would notice. So read the LIVE catalog and fail
    loudly if the cascade is gone (phase 183 A3 — a guard instead of a new
    migration).

    **The guard is not a tautology — proven 2026-10-15 against a scratch
    database** (``bor_183_fk_scratch``, created + dropped for the check;
    the shared test DB was never touched): ``alembic upgrade head``
    (``confdeltype='c'``, this test GREEN) → ``ALTER TABLE chunks DROP
    CONSTRAINT chunks_document_id_fkey`` + re-add it ``ON DELETE NO
    ACTION`` (``confdeltype='a'``) → this exact test FAILED, verbatim:

    ```
    AssertionError: chunks FKs not ON DELETE CASCADE:
    chunks_document_id_fkey confdeltype='a' (NO ACTION) — Document.chunks'
    passive_deletes=True would silently orphan chunk rows; restore the
    clause (alembic 0001 L54)
    ```

    and with the clause still present-but-NO-ACTION the same fixture's
    prune died on ``psycopg.errors.ForeignKeyViolation`` rather than
    orphaning rows unnoticed — the ORM really does delegate the child rows
    to the database now (which is the point, and exactly why the clause is
    load-bearing).
    """
    # ``pg_constraint.confdeltype`` codes → the DDL clause, so the failure
    # message names whatever the FK actually does (not always NO ACTION).
    deltype = {
        "a": "NO ACTION",
        "r": "RESTRICT",
        "c": "CASCADE",
        "n": "SET NULL",
        "d": "SET DEFAULT",
    }
    rows = db.execute(
        text(
            "select conname, confdeltype from pg_constraint"
            " where conrelid = 'chunks'::regclass and contype = 'f'"
            " order by conname"
        )
    ).all()
    # 'c' = ON DELETE CASCADE (pg_constraint.confdeltype).
    assert rows, "no foreign key on chunks at all — the cascade cannot hold"
    assert all(r[1] == "c" for r in rows), (
        "chunks FKs not ON DELETE CASCADE: "
        + ", ".join(
            f"{r[0]} confdeltype={r[1]!r} ({deltype.get(r[1], 'UNKNOWN')})" for r in rows
        )
        + " — Document.chunks' passive_deletes=True would silently orphan"
        " chunk rows; restore the clause (alembic 0001 L54)"
    )


def test_prune_with_a_loaded_collection_still_deletes_the_chunks(db: Session) -> None:
    """The other caller state: the collection IS loaded (phase 183 D4).

    ``passive_deletes=True`` only promises not to LOAD the children — a
    caller that already holds them (what ``_replace_content_chunks`` leaves
    behind in the same sync session, or any future caller that queries
    eagerly) must still lose the rows. Verified 2026-10-15: the ORM deletes
    the LOADED children with its own ``DELETE FROM chunks`` and the FK
    cascade covers whatever is not loaded — no ``SELECT`` of ``chunks``
    either way, and zero orphan rows in BOTH states. The keyword changes
    what Python loads, never what the database deletes.
    """
    doc = Document(
        source=PROBE_SOURCE,
        path="loaded.md",
        full_path=f"/srv/brain/{PROBE_SOURCE}/loaded.md",
        title="Loaded collection",
        content=CHUNK_CONTENT,
        content_hash="f" * 64,
    )
    db.add(doc)
    db.flush()
    db.add_all(
        [
            Chunk(
                document_id=doc.id,
                position=position,
                content=CHUNK_CONTENT,
                embedding=[(i + position) % 768 / 768 for i in range(768)],
            )
            for position in range(5)
        ]
    )
    db.commit()

    row = db.scalar(select(Document).where(Document.source == PROBE_SOURCE))
    assert row is not None
    assert len(row.chunks) == 5  # the collection is now IN the identity map

    db.delete(row)
    db.commit()

    assert db.scalar(
        select(func.count()).select_from(Document).where(Document.id == doc.id)
    ) == 0
    assert db.scalar(
        select(func.count()).select_from(Chunk).where(Chunk.document_id == doc.id)
    ) == 0