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

"""Integration: phase 184 — what one import run RETAINS, and its bound.

**VERDICT (task 01): CONFIRMED — the run kept every row it walked and
nothing released it — the accumulation lives on the UNCHANGED path, not the
added path.** The owner incident (2026-10-15, the same run phase 183 fixed
the prune of) is the reason: an import run of a big knowledge base grew in
memory with the SIZE OF THE KB rather than with the work the run actually
did.

**→ FIXED (task 02):** :func:`app.rag.importer.import_sources` now calls
``session.expunge_all()`` every ``Settings.sync_session_reset_files`` files
(top of the per-file loop) plus once per source before the two post-passes,
so the map is bounded by the CADENCE instead of by the knowledge base — the
pre-fix and post-fix map sizes are both measured below, and the post-fix
ceiling is :data:`MAP_BOUND`.

**The mechanism, exactly.**

* ``app/db.py:30`` builds every session with
  ``expire_on_commit=False`` — deliberately, and this phase does **NOT**
  change it: every request path reads attributes after commit. What that
  flag controls is attribute EXPIRY, not object retention; on its own it
  keeps nothing alive.
* The SQLAlchemy identity map is therefore **weak** for a CLEAN row: a row
  the run read, committed and no longer references is collected and leaves
  the map by itself (pinned by
  :func:`test_a_clean_row_leaves_the_map_and_a_dirty_one_never_does`). That
  is why the **added** path is bounded instead of retained: nothing in the
  session points at a finished file's cluster, so the next cyclic
  collection frees it. Refcounting alone does not — a ``Document`` and its
  ``Chunk`` rows form a cycle — which is why the added path's map size is a
  question of GC timing, never of what the session holds.
* The **unchanged** path is the opposite, and it is the path a production
  sync takes for nearly every file. ``_unchanged_doc``
  (``app/rag/importer.py``) stamps ``doc.file_hash = file_digest`` on EVERY
  unchanged file. The value already equals the stored one, so
  ``session.is_modified(doc)`` is ``False`` (no UPDATE, and the helper's
  commit is skipped) — but the assignment still marks the state
  ``modified``, which puts the row in ``Session.dirty``: a **strong**
  reference held by the session itself. Cyclic GC cannot help (the session
  IS the reference); ``commit()`` does not help
  (``expire_on_commit=False``). The row — its full ``content`` string
  included — is resident until the run ends.

**Measured BY THIS FILE** (12 markdown files of ~4.4 KB each, the dev
Postgres, the deterministic in-process fake embedder):

* objects the ADD run creates: 12 ``Document`` + 60 content ``Chunk`` + 12
  summary ``Chunk`` = **84**. Identity map at the end of that run: **84**
  inside the pytest process (no cyclic collection had been triggered yet —
  a document and its chunks form a reference cycle, so refcounting alone
  cannot free them) and **0** after ONE collection point
  (:func:`test_the_added_run_holds_its_objects_only_until_a_collection`).
  Over a longer 400-file run the same map sawtoothed between 40 and 200
  entries with ``len(gc.get_objects())`` flat, i.e. the added rows are
  released by the collector as the run goes: bounded by GC timing, NOT by
  the session.
* objects the UNCHANGED run walks: **12** documents. Identity map at the end
  of that run: **exactly 12** — every one a ``Document`` still holding its
  full ``content`` (53 KB of content strings in this fixture). Sampled
  through the ``progress`` hook it is a monotone staircase (before file 1:
  0 rows resident … before file 12: 11) — **one row gained per file, none
  ever released**. ``gc.collect()`` releases **0** of them;
  ``session.expunge_all()`` releases **all 12** without losing a row (task
  02's remedy, measured here so task 02 has a number to hit).
* extrapolated to the incident KB (17,407 documents, ``content`` averaging
  **15.2 KB** — the phase-184 baseline): a full unchanged sync retains
  ≈ **265 MB** of ``content`` strings plus row state on a 3,904 MB VM with
  no swap that shares its RAM with a 3.1 GB Postgres. That is the "memory
  grew with the knowledge base" the owner saw; it is proportional to the
  files walked, so the phase-147 fast path (which makes the walk faster)
  cannot reduce it.

**The hazard the fix respects (task 02, implemented).** The retained rows
are retained BECAUSE they are dirty, and ``expunge_all()`` DISCARDS
unflushed changes. That is safe on the fast path — the stamp writes the
value the database already holds, so there is nothing to lose — but it is
only safe at a point where every real write of the finished file has been
committed (``_unchanged_doc`` commits its date refresh and its
``is_modified`` stamp, ``_store_summary`` and the index paths commit per
file). Expunging before such a commit would silently drop that file's
write, which is why the cadence sits at the top of the per-file loop
(after the previous file's ``_index_file`` returned) and at the source
boundary (before the two post-passes, which select their own rows fresh)
— and nowhere else: never inside ``_index_file``, ``_unchanged_doc``,
``_store_summary`` or a post-pass, whose locals are live across
statements.

**Measured AFTER the fix (task 02, the same 12-file fixture).** With
``sync_session_reset_files=3`` the UNCHANGED walk is a sawtooth
``[0, 1, 2, 0, 1, 2, 0, 1, 2, 0, 1, 2]`` sampled at the ``progress`` hook
— a release every third file where the cadence-``0`` run ramps
monotonically to 12 — and it ends at **0** rows, because the
source-boundary release is the last thing the walk does and the two
post-passes select no rows for a source with no image docs. The ADD run
(whose rows are only weakly held, so its mid-run number moves with GC
timing) peaks at **14** entries sampled mid-run — two files' worth of
seven objects, the widest gap the cadence allows — against the
``0``-cadence run's 21→49 climb, and it too ends at **0**. Both runs
report a :func:`dataclasses.asdict`-identical ``ImportSummary`` and leave
a byte-identical database next to the cadence-``0`` run (see
:func:`test_reset_cadence_bounds_the_map`): the release costs nothing but
residency.

**Why RSS is never asserted here** (phase 183 LOCKED A4, inherited): the
allocator's behaviour is CI noise. ``VmRSS`` over a 400-file run of this
same fixture grew ≈ 35 KB/file and did NOT come back after
``gc.collect()`` — that is freed-but-unreturned allocator memory, NOT
session retention (``len(gc.get_objects())`` stayed flat across the whole
400-file run: 190k→210k sawtooth). ``len(session.identity_map)`` is the
exact, deterministic retention surface, and every assertion here measures
it through :func:`_map_size` — the one door tasks 02/03 assert through.

**The prune side (task 03) — the OTHER half of the incident.** The walk
above is what a sync costs per file; ``_prune`` is what it costs per
DOCUMENT IN THE SOURCE, and on 2026-10-15 it was where the box actually
died. It used to ``select(Document)`` the FULL row — ``content`` included,
averaging **15.2 KB** — for every document of every pruned source (17,407
rows ≈ **252 MB** of markdown) to decide a deletion it then threw away, and
to hold the whole delete in ONE transaction until the end. Post-task-03 it
selects six light columns (``id``, ``source``, ``path``, ``is_image``,
``image_path``, ``page_count`` — everything the guards and the copy cleanup
read) ``order_by(Document.id)`` and commits every
``Settings.sync_prune_batch`` deletes. Measured with the ``before_cursor_execute``
capture below over a 30-doc × 5-chunk × 8 KB fixture: **1** ``SELECT … FROM
documents`` for the whole walk (no per-row follow-up: the six columns really
are enough), ``documents.content`` ABSENT from its column list, and
**3** commits at ``sync_prune_batch=10`` vs exactly **1** at ``0`` (the
pre-phase rollback) with every count, ``changed_paths`` entry and row total
identical. The guard pins
(:func:`test_prune_guards_unaffected_with_light_rows`) are what prove the
light row is COMPLETE — a column missing from ``load_only`` would emit one
lazy SELECT per row and fail that test, not get papered over.

**What this file does NOT touch:** the walk's column set (phase 185),
``scripts/git_sync.py`` (phase 186), ``app/db.py:30``, and the
``doc.chunks = [...]`` re-embed path (phase 183 LOCKED A2).
"""
from __future__ import annotations

import asyncio
import dataclasses
import gc
import re
from collections.abc import Callable, Iterator
from pathlib import Path
from uuid import UUID

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

from app.config import Settings
from app.db import SessionLocal, engine
from app.models import Chunk, Document
from app.rag.importer import ImportSummary, _prune, import_sources
from app.rag.llm import EmbeddingError
from tests.fakes import FakeEmbedder

# ---------------------------------------------------------------------------
# The fixture: 12 files of ~4.4 KB, each producing 5 content chunks
# ---------------------------------------------------------------------------

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

#: One import run's worth of files. Small on purpose: what the tests pin is
#: the SHAPE of the retention (an unbounded ramp with the knob ``0``, a
#: cadence-bounded sawtooth with it on), so any count works — 12 keeps the
#: run in seconds, divides by both cadences used here (2 and 3) and keeps
#: the staircase readable. Task 04 drives the cadence through the UI with a
#: bigger fixture.
FILE_COUNT = 12

#: One repeated sentence, so every file has the same shape (and therefore
#: the same chunk count) as every other.
_SENTENCE = (
    "The homelab runs podman quadlets on the hnb host with a static DNS "
    "zone, nightly restic backups and a systemd timer that probes uptime. "
)

#: The pinned output of :func:`_seed_files`: 5 content chunks per file (4
#: sections of ~1.1 KB against ``chunk_target_chars=2000``) and 1 summary
#: chunk per document (phase 118 A2).
EXPECTED_CONTENT_CHUNKS = 60
EXPECTED_SUMMARY_CHUNKS = 12

#: Every ORM object a full ADD run creates: the 12 documents, their content
#: chunks and their summary chunks. What the identity map would have to
#: hold if the ADDED path retained what the phase-184 baseline describes.
OBJECTS_CREATED_BY_ADD_RUN = (
    FILE_COUNT + EXPECTED_CONTENT_CHUNKS + EXPECTED_SUMMARY_CHUNKS
)

#: The measured number of identity-map entries at the end of an UNCHANGED
#: run with the cadence OFF — one per file walked, each a full ``Document``
#: (module docstring). The counterpart
#: :func:`test_reset_cadence_bounds_the_unchanged_walk` pins what the same
#: walk does with the cadence ON: a sawtooth capped by the cadence, ending
#: at 0.
MEASURED_UNCHANGED_MAP_SIZE = FILE_COUNT

#: Task 02's cadence for the bounded tests: 12 files / every 3rd = FOUR
#: releases inside the walk, so a single missed reset cannot hide.
RESET_FILES = 3

#: Every ORM object ONE added file puts through the session: its
#: ``Document`` + its 5 content chunks + its 1 summary chunk.
OBJECTS_PER_ADDED_FILE = 1 + EXPECTED_CONTENT_CHUNKS // FILE_COUNT + 1  # 7

#: The post-fix ceiling (``sync_session_reset_files + slack``, task 02).
#: The arithmetic: between two cadence releases at most
#: ``RESET_FILES - 1`` files have been indexed, each contributing at most
#: :data:`OBJECTS_PER_ADDED_FILE` objects (the ``2 * 7 = 14`` the ADD run is
#: measured peaking at) — the last partial batch. ``RESET_FILES`` on top is
#: the slack for the file being indexed when the sample is taken. Measured:
#: the UNCHANGED run ends at 0 and the ADD run ends at 0, so 17 is a
#: ceiling, not an expectation — what matters is that it does not scale
#: with :data:`FILE_COUNT` (the pre-fix run measured
#: :data:`MEASURED_UNCHANGED_MAP_SIZE` = 12 grown from the KB, 49 on the
#: ADD path).
MAP_BOUND = RESET_FILES + (RESET_FILES - 1) * OBJECTS_PER_ADDED_FILE  # 17

#: The content token :class:`_BoomEmbedder` refuses on — file #4
#: (``probe-03.md`` carries ``# Probe document 3`` + ``of probe 3``).
BOOM_MARKER = "probe 3"


def _seed_files(tmp: Path) -> Path:
    """``FILE_COUNT`` markdown files of ~4.4 KB under a fresh source root."""
    root = tmp / PROBE_SOURCE
    root.mkdir()
    for n in range(FILE_COUNT):
        parts = [f"# Probe document {n}\n"]
        for section in range(4):
            parts.append(f"\n## Section {section} of probe {n}\n\n" + _SENTENCE * 8 + "\n")
        body = "".join(parts)
        assert 4_300 <= len(body) <= 4_700, len(body)
        (root / f"probe-{n:02d}.md").write_text(body)
    return root


def _probe_llm(reset_files: int = 0) -> FakeEmbedder:
    """The fake embedder with BOTH phase-184 knobs explicit — ``0`` by
    default, the exact pre-phase behaviour (LOCKED A3: ``0`` is the
    operator's rollback, no re-deploy). The run is byte-identical to a run
    predating the settings entirely; the explicitness is the point — these
    tests stay true after tasks 02/03 ship, because they pin what ``0``
    guarantees. *reset_files* ``> 0`` asks for the shipped cadence."""
    llm = FakeEmbedder()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        sync_session_reset_files=reset_files,
        sync_prune_batch=0,
    )
    return llm


class _BoomEmbedder(FakeEmbedder):
    """A fake embedder that refuses every embed whose text carries
    :data:`BOOM_MARKER` — the importer's ``EmbeddingError`` path (the
    per-file ``session.rollback()``), driven here AFTER the identity map has
    already been reset, to pin that detached state breaks nothing."""

    async def embed(
        self,
        texts: list[str],
        *,
        on_progress: Callable[[int, int], None] | None = None,
    ) -> list[list[float]]:
        if any(BOOM_MARKER in t for t in texts):
            raise EmbeddingError(f"simulated endpoint refusal for {BOOM_MARKER}")
        return await super().embed(texts, on_progress=on_progress)


def _map_size(session: Session) -> int:
    """The run's retention surface — the ONE door every assertion here (and
    in tasks 02/03) measures through."""
    return len(session.identity_map)


def _map_classes(session: Session) -> dict[str, int]:
    """How many identity-map entries belong to each mapped class."""
    counts: dict[str, int] = {}
    for state in session.identity_map.all_states():
        name = state.class_.__name__
        counts[name] = counts.get(name, 0) + 1
    return counts


def _resident_content_chars(session: Session) -> int:
    """Content characters held resident by the ``Document`` rows still in
    the identity map — the per-run cost that scales with the KB."""
    total = 0
    for state in session.identity_map.all_states():
        if state.class_ is not Document:
            continue
        obj = state.obj()
        assert isinstance(obj, Document)
        total += len(obj.content)
    return total


def _truncate(db: Session) -> None:
    """Empty the KB mid-test AND drop every row *db* has cached, so the next
    read is the database's truth rather than a stale identity-map entry."""
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()
    db.expunge_all()


def _kb_snapshot(db: Session) -> dict[str, tuple[str, str | None, str, int]]:
    """path → ``(content, summary, file_hash, chunk count)`` — the DB-side
    ground truth the cadence-bounded run must match the pre-phase run for,
    byte for byte (task 02's identical-output contract)."""
    counts: dict[UUID, int] = {
        row[0]: row[1]
        for row in db.execute(
            select(Chunk.document_id, func.count(Chunk.id)).group_by(Chunk.document_id)
        )
    }
    snapshot: dict[str, tuple[str, str | None, str, int]] = {}
    for doc in db.scalars(
        select(Document)
        .where(Document.source == PROBE_SOURCE)
        .order_by(Document.path)
    ):
        assert doc.file_hash is not None  # the phase-147 stamp is always set
        snapshot[doc.path] = (
            doc.content,
            doc.summary,
            doc.file_hash,
            counts.get(doc.id, 0),
        )
    return snapshot


@pytest.fixture(autouse=True)
def clean_kb(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()


# ---------------------------------------------------------------------------
# The runs
# ---------------------------------------------------------------------------


def test_the_added_run_holds_its_objects_only_until_a_collection(
    db: Session, tmp_path: Path
) -> None:
    """The added path is NOT session-held — one collection point clears it.

    A fresh session indexes :data:`FILE_COUNT` new files
    (``prune=False``), creating :data:`OBJECTS_CREATED_BY_ADD_RUN` ORM
    objects. Measured: all 84 are still in the map at the end of the run
    inside the pytest process, because a ``Document`` and its ``Chunk``
    rows form a reference cycle that refcounting cannot break — but every
    one of them left the session at their commit (``session.is_modified``
    is false, nothing points at them), so a single ``gc.collect()`` drops
    the map to **0**. Contrast that with
    :func:`test_the_unchanged_run_keeps_every_document_it_walks`, where no
    number of collections frees anything: THAT is the leak-shaped
    retention, and it is the path a production sync takes for nearly every
    file.
    """
    root = _seed_files(tmp_path)
    session = SessionLocal()
    try:
        summary = asyncio.run(
            import_sources([root], _probe_llm(), prune=False, session=session)
        )
        assert (summary.files, summary.added, summary.unchanged) == (
            FILE_COUNT,
            FILE_COUNT,
            0,
        )
        assert summary.chunks == EXPECTED_CONTENT_CHUNKS
        assert summary.summaries == EXPECTED_SUMMARY_CHUNKS
        assert summary.errors == 0

        # The objects really were created — the DB is the ground truth for
        # how many ORM objects the run put through the session.
        n_docs = db.scalar(select(func.count()).select_from(Document))
        n_chunks = db.scalar(select(func.count()).select_from(Chunk))
        assert n_docs == FILE_COUNT
        assert n_chunks == EXPECTED_CONTENT_CHUNKS + EXPECTED_SUMMARY_CHUNKS
        assert n_docs + n_chunks == OBJECTS_CREATED_BY_ADD_RUN

        # Whatever the collector has not reached yet is NOT held by the
        # session — it is only garbage the cycle collector has not visited.
        gc.collect()
        assert _map_size(session) == 0, (
            f"{_map_size(session)} entries of a finished file are still"
            " reachable from the session after a collection point — the"
            " weak-map mechanism of the module docstring no longer holds"
        )
    finally:
        session.close()


def test_the_unchanged_run_keeps_every_document_it_walks(
    db: Session, tmp_path: Path
) -> None:
    """THE PROBE: a repeat sync holds one full ``Document`` per file walked.

    The production-shaped case — the second sync of the same source, every
    file unchanged, the phase-147 fast path for all of them. The identity
    map grows by exactly one row per file and loses nothing: sampled
    through the ``progress`` hook (which fires before each file's
    ``_index_file``) it is the staircase ``0, 1, … FILE_COUNT-1``, and it
    ends at :data:`MEASURED_UNCHANGED_MAP_SIZE`.

    This is the accumulation phase 184 exists for, and it is NOT
    collectable: ``gc.collect()`` releases none of it — the session's own
    ``dirty`` set is what holds the rows (the ``doc.file_hash`` stamp of
    ``_unchanged_doc`` marks the state modified and never commits it) —
    while ``session.expunge_all()`` releases all of it (see
    :func:`test_a_clean_row_leaves_the_map_and_a_dirty_one_never_does`).
    """
    root = _seed_files(tmp_path)
    seed = SessionLocal()
    try:
        first = asyncio.run(import_sources([root], _probe_llm(), session=seed))
        assert first.added == FILE_COUNT
    finally:
        seed.close()
    gc.collect()

    session = SessionLocal()
    try:
        staircase: list[int] = []

        def progress(source: str, rel: str, done: int, total: int) -> None:
            """The per-file sample — the hook fires BEFORE this file is
            indexed, so the value is the residue of the files already
            walked (``source``/``rel``/``done``/``total`` are not needed to
            measure retention)."""
            del source, rel, done, total
            staircase.append(_map_size(session))

        summary = asyncio.run(
            import_sources(
                [root], _probe_llm(), prune=False, session=session, progress=progress
            )
        )
        assert (summary.files, summary.added, summary.unchanged) == (
            FILE_COUNT,
            0,
            FILE_COUNT,
        )

        # One row resident per file already walked — a staircase, never a
        # plateau and never a release.
        assert staircase == list(range(FILE_COUNT)), (
            "expected exactly one document retained per file already walked,"
            f" got {staircase}"
        )
        assert _map_size(session) == MEASURED_UNCHANGED_MAP_SIZE

        # …and every one of them is a full Document row, content included
        # (the 15.2 KB-per-document production figure is what makes this
        # scale with the knowledge base instead of with the work).
        assert _map_classes(session) == {"Document": FILE_COUNT}
        assert _resident_content_chars(session) >= FILE_COUNT * 4_000

        # Not collectable: the session IS the reference.
        gc.collect()
        assert _map_size(session) == MEASURED_UNCHANGED_MAP_SIZE, (
            "the retained rows became collectable — the probe's premise (and"
            " task 02's fix) needs re-measuring"
        )
    finally:
        session.close()


def test_reset_cadence_is_configurable_off(db: Session, tmp_path: Path) -> None:
    """``sync_session_reset_files=0`` IS the pre-phase run — the rollback
    guarantee (LOCKED A3), true before task 02 and after it ships.

    With both knobs explicitly ``0`` the unchanged-path run keeps exactly
    what :func:`test_the_unchanged_run_keeps_every_document_it_walks`
    measured (:data:`MEASURED_UNCHANGED_MAP_SIZE` rows, all ``Document``,
    none collectable), and reports byte-identical counts. This is not a
    pre-fix-only assertion: it is the operator's promise that
    ``BOR_SYNC_SESSION_RESET_FILES=0`` restores today's behaviour exactly.
    """
    llm = _probe_llm()
    assert llm.settings.sync_session_reset_files == 0
    assert llm.settings.sync_prune_batch == 0

    root = _seed_files(tmp_path)
    seed = SessionLocal()
    try:
        first = asyncio.run(import_sources([root], _probe_llm(), session=seed))
        assert first.added == FILE_COUNT
    finally:
        seed.close()
    gc.collect()

    session = SessionLocal()
    try:
        summary = asyncio.run(import_sources([root], llm, prune=False, session=session))
        assert (summary.files, summary.added, summary.unchanged, summary.pruned) == (
            FILE_COUNT,
            0,
            FILE_COUNT,
            0,
        )
        assert summary.chunks == 0  # the fast path re-embeds nothing
        assert _map_size(session) == MEASURED_UNCHANGED_MAP_SIZE
        assert _map_classes(session) == {"Document": FILE_COUNT}
        gc.collect()
        assert _map_size(session) == MEASURED_UNCHANGED_MAP_SIZE
    finally:
        session.close()


def test_reset_cadence_bounds_the_map(db: Session, tmp_path: Path) -> None:
    """THE FIX: with ``sync_session_reset_files=3`` the map stays inside one
    cadence window — and the run's OUTPUT is identical to the cadence-``0``
    run it replaces.

    The same :func:`_seed_files` fixture is imported twice into an empty KB:
    once with the knob ``0`` (the pre-phase reference) and once with the
    cadence at :data:`RESET_FILES`. The bounded run reports
    ``added=12`` + :data:`EXPECTED_CONTENT_CHUNKS` chunks, its whole
    ``ImportSummary`` compares equal to the reference's, and the database
    matches it path by path (content, summary, ``file_hash`` and the per-path
    chunk count) — the release drops residency, never a byte of work.

    The bound itself (:data:`MAP_BOUND`) is asserted twice: at the end of the
    run (measured **0** — the source-boundary release is the last thing the
    walk does, and the two post-passes select no rows for a source with no
    image docs) and mid-run through the ``progress`` hook, where the ceiling
    is the ``RESET_FILES - 1`` files indexed since the last release times
    :data:`OBJECTS_PER_ADDED_FILE` (measured **14**). Neither number moves
    when the fixture grows: that is the whole point of the phase.
    """
    root = _seed_files(tmp_path)

    # 1 — the reference run: the exact pre-phase behaviour (both knobs 0).
    reference = SessionLocal()
    try:
        ref_summary = asyncio.run(
            import_sources([root], _probe_llm(), prune=False, session=reference)
        )
    finally:
        reference.close()
    ref_kb = _kb_snapshot(db)

    # 2 — the same walk over an empty KB, now with the cadence on.
    _truncate(db)
    session = SessionLocal()
    try:
        staircase: list[int] = []

        def progress(source: str, rel: str, done: int, total: int) -> None:
            """Per-file sample — the hook fires BEFORE this file is indexed,
            so it measures what the previous files left behind."""
            del source, rel, done, total
            staircase.append(_map_size(session))

        summary = asyncio.run(
            import_sources(
                [root],
                _probe_llm(RESET_FILES),
                prune=False,
                session=session,
                progress=progress,
            )
        )
        assert _map_size(session) <= MAP_BOUND, (
            f"the finished run still holds {_map_size(session)} objects — the"
            f" cadence window allows {MAP_BOUND}"
        )
        assert staircase, "the progress hook never fired"
        assert max(staircase) <= (RESET_FILES - 1) * OBJECTS_PER_ADDED_FILE, (
            f"mid-run the walk held {max(staircase)} objects — more than the"
            f" {RESET_FILES - 1} files allowed between two releases"
        )
    finally:
        session.close()

    # 3 — the byte-identical contract (the phase's reason for existing).
    assert summary.added == FILE_COUNT
    assert summary.chunks == EXPECTED_CONTENT_CHUNKS
    assert summary.summaries == EXPECTED_SUMMARY_CHUNKS
    assert summary.errors == 0
    assert dataclasses.asdict(summary) == dataclasses.asdict(ref_summary)
    assert _kb_snapshot(db) == ref_kb


def test_reset_cadence_bounds_the_unchanged_walk(db: Session, tmp_path: Path) -> None:
    """The production path, bounded: a repeat sync sawtooths instead of
    ramping.

    :func:`test_the_unchanged_run_keeps_every_document_it_walks` is the
    PRE-fix picture (the staircase ``0, 1, … 11``). With the cadence at
    :data:`RESET_FILES` the SAME second sync — every file unchanged, every
    row dirty from the ``file_hash`` stamp, none of it collectable — samples
    ``[0, 1, 2]`` repeated: the release lands before files 4, 7 and 10, and
    the run ends at **0** rows resident, because the source-boundary release
    runs after the last file and the post-passes (no image docs here) select
    nothing. Both halves matter: the sawtooth pins the per-file reset, the
    end-of-run ``0`` pins the source boundary.
    """
    root = _seed_files(tmp_path)
    seed = SessionLocal()
    try:
        first = asyncio.run(import_sources([root], _probe_llm(), session=seed))
        assert first.added == FILE_COUNT
    finally:
        seed.close()
    gc.collect()

    session = SessionLocal()
    try:
        staircase: list[int] = []

        def progress(source: str, rel: str, done: int, total: int) -> None:
            del source, rel, done, total
            staircase.append(_map_size(session))

        summary = asyncio.run(
            import_sources(
                [root],
                _probe_llm(RESET_FILES),
                prune=False,
                session=session,
                progress=progress,
            )
        )
        assert (summary.files, summary.added, summary.unchanged) == (
            FILE_COUNT,
            0,
            FILE_COUNT,
        )
        assert staircase == [n % RESET_FILES for n in range(FILE_COUNT)], (
            f"the bounded walk sampled {staircase} — expected a release every"
            f" {RESET_FILES} files"
        )
        assert max(staircase) < MEASURED_UNCHANGED_MAP_SIZE
        assert _map_size(session) == 0, (
            "the source-boundary release did not run — one source's walk"
            f" residue ({_map_size(session)} rows) is still resident"
        )
        gc.collect()
        assert _map_size(session) == 0
    finally:
        session.close()


def test_reset_cadence_survives_an_embedding_error(db: Session, tmp_path: Path) -> None:
    """The error funnel with a map already reset — nothing blows up, nothing
    is lost.

    ``sync_session_reset_files=2`` means releases have already happened
    before the stub embedder refuses file #4 (``probe-03.md``), so the
    handler's ``session.rollback()`` runs on a session whose earlier objects
    are DETACHED. The run records ``errors=1``, keeps walking, and indexes
    the other 11 files completely: every survivor keeps all 6 of its chunks
    and its content, the failed file leaves no row at all, and
    ``changed_paths`` is the 11 survivors in walk order — the same shape the
    pre-phase error pin produces
    (``tests/unit/test_importer.py::test_embedding_failure_is_logged_and_import_continues`,
    which runs with the cadence at its ``0`` default).
    """
    root = _seed_files(tmp_path)
    llm = _BoomEmbedder()
    llm.settings = Settings(
        _env_file=None,  # pyright: ignore[reportCallIssue]
        sync_session_reset_files=2,
        sync_prune_batch=0,
    )

    session = SessionLocal()
    try:
        summary = asyncio.run(
            import_sources([root], llm, prune=False, session=session)
        )
        assert (summary.files, summary.errors) == (FILE_COUNT, 1)
        assert summary.added == FILE_COUNT - 1
        assert summary.chunks == EXPECTED_CONTENT_CHUNKS - (
            EXPECTED_CONTENT_CHUNKS // FILE_COUNT
        )
        assert summary.summaries == EXPECTED_SUMMARY_CHUNKS - 1
        assert _map_size(session) <= MAP_BOUND
    finally:
        session.close()

    failed = f"probe-{3:02d}.md"
    assert summary.changed_paths == [
        (PROBE_SOURCE, f"probe-{n:02d}.md") for n in range(FILE_COUNT) if n != 3
    ]
    snapshot = _kb_snapshot(db)
    assert set(snapshot) == {f"probe-{n:02d}.md" for n in range(FILE_COUNT)} - {failed}
    # Every file walked BEFORE the failure kept its full row and its chunks —
    # detaching them cost the committed work nothing.
    assert {path: row[3] for path, row in snapshot.items()} == {
        path: OBJECTS_PER_ADDED_FILE - 1 for path in snapshot
    }
    assert all(row[0] and row[2] for row in snapshot.values())


def test_a_clean_row_leaves_the_map_and_a_dirty_one_never_does(db: Session) -> None:
    """The mechanism in one test — why ``expunge_all()`` is the lever.

    A committed, CLEAN row is held weakly: drop the last strong reference
    and it leaves the identity map by itself (which is why the added run
    above stays flat). Assign a value to it — even the value it already
    has, which is exactly ``_unchanged_doc``'s ``doc.file_hash =
    file_digest`` stamp — and the state becomes ``modified``, the session's
    ``dirty`` set holds it STRONGLY, and the row survives every collection
    until the session ends. ``expire_on_commit=False`` is not what retains
    it, and ``commit()`` does not clear it; only ``expunge`` / ``close``
    does — and expunging drops no data, because the dirty value equals the
    stored one.
    """
    session = SessionLocal()
    try:
        doc = Document(
            source=PROBE_SOURCE,
            path="mechanism.md",
            full_path=f"/srv/brain/{PROBE_SOURCE}/mechanism.md",
            title="Mechanism",
            content="x" * 64,
            content_hash="a" * 64,
            file_hash="a" * 64,
        )
        session.add(doc)
        session.commit()
        assert _map_size(session) == 1

        del doc
        gc.collect()
        assert _map_size(session) == 0, "a clean committed row must be weakly held"

        row = session.scalar(
            select(Document).where(
                Document.source == PROBE_SOURCE, Document.path == "mechanism.md"
            )
        )
        assert row is not None
        row.file_hash = row.file_hash  # the stamp: same value, dirty state
        assert sa_inspect(row).modified is True
        assert session.is_modified(row) is False  # no UPDATE will be emitted
        del row
        gc.collect()
        assert _map_size(session) == 1, (
            "the stamped row is no longer held by the session — the unchanged"
            " path's retention (and this phase's premise) changed"
        )

        # And the lever: expunging releases the row and loses nothing — the
        # bytes in the database are already what the stamp wanted.
        session.expunge_all()
        gc.collect()
        assert _map_size(session) == 0
        assert (
            db.scalar(
                select(func.count())
                .select_from(Document)
                .where(Document.source == PROBE_SOURCE)
            )
            == 1
        )
    finally:
        session.close()


# ---------------------------------------------------------------------------
# Task 03 — the prune: its ROW (the light select) and its TRANSACTIONS
# ---------------------------------------------------------------------------

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

#: The task-03 fixture: 30 documents × 5 chunks. Small on purpose — what the
#: tests pin is the SHAPE of the prune (the columns it loads, the
#: transactions it runs), and 30 is more than the batch used here so one
#: missed commit boundary cannot hide.
PRUNE_DOC_COUNT = 30
PRUNE_CHUNKS_PER_DOC = 5

#: ~8 KB of markdown per document (production ``content`` averages 15.2 KB).
#: A full row is FAR bigger than a light row, which is why "``content`` is
#: not in the SELECT" is the load-bearing pin of this task.
PRUNE_CONTENT = ("The homelab runs podman quadlets on the hnb host. " * 200)[:8_192]

#: Task 03's batch cadence for the commit-count pin: 30 docs / every 10th =
#: THREE commits, so a single missed boundary cannot hide.
PRUNE_BATCH = 10

#: The SIX columns the prune loads — the ``load_only`` set of ``_prune``
#: (docstring: the light-row contract). An eighth name here means the walk
#: got heavier; one missing means a guard lost the column it reads.
PRUNE_COLUMNS = frozenset(
    {"id", "source", "path", "is_image", "image_path", "page_count"}
)

#: ``… FROM documents`` — the prune's own walk AND (that is the pin) any
#: per-document follow-up lazy load of a deferred column. The word boundary
#: keeps ``DELETE FROM documents`` out of the count.
_DOC_SELECT_RE = re.compile(r"\bfrom\s+documents\b", re.IGNORECASE)


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

    The ``before_cursor_execute`` technique of phase 183 task 01
    (``tests/integration/test_prune_cascade_statements.py``): the column
    list a SELECT uses is invisible in the ORM's Python API and plain to
    see in the statement it compiles, so a capture is the only
    deterministic way to pin what the prune's row actually contains.
    Installed on the engine and REMOVED in teardown, so no other test of
    the session ever sees the listener.
    """
    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)


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


def _projected_columns(statement: str) -> frozenset[str]:
    """The ``documents.<col>`` names in ONE SELECT's PROJECTION.

    Only the part before ``FROM`` counts — ``WHERE documents.source IN`` and
    ``ORDER BY documents.id`` are a predicate and a sort, not a column the
    row holds.
    """
    head = statement.split("\nFROM ", 1)[0]
    return frozenset(re.findall(r"documents\.\w+", head))


def _seed_prune_docs(
    session: Session, *, page_dir: Path | None = None
) -> list[UUID]:
    """``PRUNE_DOC_COUNT`` indexed docs of :data:`PRUNE_SOURCE`, each with
    ``PRUNE_CHUNKS_PER_DOC`` embedded chunks and a full-fat ``content``.

    Committed and then EXPLICITLY expunged, so the prune starts from where a
    real sync's prune starts: an empty identity map and a SELECT. The chunks
    go in by ``document_id`` (never through ``doc.chunks``), the phase-183
    convention that keeps the child collection UNLOADED.

    With *page_dir* every doc becomes a page-QUALIFYING one (``page_count=2``,
    phase 148) and its two ``<doc-id>-pN.png`` copies are written there — the
    state in which the prune reads ``id`` + ``page_count``, i.e. the sixth
    and seventh reads of the light row.
    """
    ids: list[UUID] = []
    for n in range(PRUNE_DOC_COUNT):
        doc = Document(
            source=PRUNE_SOURCE,
            path=f"docs/doc-{n:04d}.md",
            full_path=f"/srv/brain/{PRUNE_SOURCE}/docs/doc-{n:04d}.md",
            title=f"Prune probe doc {n}",
            content=PRUNE_CONTENT,
            content_hash=f"{n:064x}",
            page_count=2 if page_dir is not None else None,
        )
        session.add(doc)
        session.flush()
        ids.append(doc.id)
        if page_dir is not None:
            for page in (1, 2):
                (page_dir / f"{doc.id}-p{page}.png").write_bytes(b"\x89PNG page")
        rot = n % 768
        session.add_all(
            [
                Chunk(
                    document_id=doc.id,
                    position=position,
                    content=PRUNE_CONTENT,
                    embedding=[((i + rot) % 768) / 768 for i in range(768)],
                )
                for position in range(PRUNE_CHUNKS_PER_DOC)
            ]
        )
    session.commit()
    session.expunge_all()
    return ids


def _arm_commit_counter(session: Session) -> list[int]:
    """A commit counter on *session* (task 03's transaction pin).

    ``after_commit`` fires once per ``Session.commit()`` that ends a root
    transaction — including an empty one — so what the list counts is
    exactly the prune's own transaction cadence and nothing else.
    """
    commits: list[int] = []
    event.listen(session, "after_commit", lambda s: commits.append(1))
    return commits


def test_prune_holds_no_content(
    db: Session, tmp_path: Path, statement_capture: list[str]
) -> None:
    """THE PRUNE PROBE: the walk loads SIX columns and never ``content``.

    30 documents × 5 chunks are pruned through the importer's own ``_prune``
    with an empty *seen* (every doc of the source goes), and the captured
    SQL pins all three halves of the task-03 contract:

    * the walk is exactly **ONE** ``SELECT … FROM documents`` — no
      per-document follow-up SELECT, i.e. nothing inside the loop reads a
      deferred column. An eighth statement here means the light row is
      INCOMPLETE, and the fix is to add that column to ``load_only`` (never
      a ``raiseload``, never a paper-over);
    * that SELECT's projection is EXACTLY :data:`PRUNE_COLUMNS` — no
      ``documents.content`` (15.2 KB per row in production, ~252 MB across
      the incident KB), no ``summary``, no ``full_path``, neither hash, no
      date;
    * the deletes still remove everything: 0 documents, **0 orphan chunk
      rows** (phase 183's ``ON DELETE CASCADE`` doing the child work), the
      pruned docs' PDF page copies gone from disk (the phase-148 cleanup,
      which is what makes ``id`` + ``page_count`` real reads), plus the full
      phase-127 ``changed_paths`` feed.
    """
    page_dir = tmp_path / "pdf_pages"
    page_dir.mkdir()
    ids = _seed_prune_docs(db, page_dir=page_dir)
    statement_capture.clear()  # the seed's own SQL is not part of the probe

    summary = ImportSummary()
    pruned = _prune(
        db,
        {PRUNE_SOURCE},
        set(),
        images=False,
        summary=summary,
        page_dir=page_dir,
        prune_batch=0,
    )
    selects = _doc_selects(statement_capture)

    assert pruned == PRUNE_DOC_COUNT
    assert len(selects) == 1, (
        f"{len(selects)} SELECTs over documents for a {PRUNE_DOC_COUNT}-doc "
        "prune — a deferred column is being read per row, so the light row "
        "is incomplete (add the column to load_only)"
    )
    projected = {
        col.removeprefix("documents.")
        for col in _projected_columns(selects[0])
    }
    assert projected == PRUNE_COLUMNS, (
        f"the prune's row is not the six-column light row: {sorted(projected)}"
    )
    assert "content" not in projected and "summary" not in projected, (
        "the prune SELECTs content again — its resident set scales with the "
        "knowledge base instead of with the deletes (the 2026-10-15 incident)"
    )

    # Everything really went, children included (no orphans — phase 183 D4,
    # asserted here too so a light row cannot quietly change it).
    assert db.scalar(
        select(func.count()).select_from(Document).where(Document.source == PRUNE_SOURCE)
    ) == 0
    assert db.scalar(
        select(func.count()).select_from(Chunk).where(Chunk.document_id.in_(ids))
    ) == 0
    assert sorted(summary.changed_paths) == sorted(
        (PRUNE_SOURCE, f"docs/doc-{n:04d}.md") for n in range(PRUNE_DOC_COUNT)
    )
    # The page copies went too — read off the light row (id + page_count).
    assert list(page_dir.iterdir()) == []
    # And the light rows leave nothing resident behind either.
    gc.collect()
    assert _map_classes(db) == {}


def test_prune_without_sources_touches_nothing(
    db: Session, statement_capture: list[str]
) -> None:
    """The empty-source guard is a real short-circuit, not a walk for naught.

    ``_prune`` with no source names returns ``0`` before the SELECT — so a
    sync that imported nothing prunes nothing WITHOUT reading a single row
    (the caller passes the set of sources it actually walked).
    """
    _seed_prune_docs(db)
    statement_capture.clear()

    summary = ImportSummary()
    assert _prune(db, set(), set(), summary=summary, prune_batch=10) == 0
    assert statement_capture == []
    assert summary.changed_paths == []
    assert db.scalar(
        select(func.count()).select_from(Document).where(Document.source == PRUNE_SOURCE)
    ) == PRUNE_DOC_COUNT


def test_prune_batch_commits(db: Session) -> None:
    """The transaction cadence: THREE commits at ``10``, exactly ONE at ``0``.

    The same 30-doc prune run twice against an empty KB — once with
    ``prune_batch=10`` and once with ``prune_batch=0`` (the pre-phase
    rollback, LOCKED A3). The counter sees the three boundaries (10, 20, 30)
    and NOTHING after them, where the ``0`` run commits once at the end.
    Every other number is identical, which is the point: the cadence changes
    WHEN the deletes become durable, never how many there are —

    ``pruned=30``, the same 30 ``changed_paths``, 0 documents and 0 chunk
    rows left. (A batch that divides the count does NOT get a fourth, empty
    commit — the final commit is skipped exactly when the last boundary
    already committed, so ``prune_batch=0`` stays the one-transaction run it
    always was.)
    """

    def run(prune_batch: int) -> tuple[int, ImportSummary, int]:
        ids = _seed_prune_docs(db)
        session = SessionLocal()
        summary = ImportSummary()
        try:
            commits = _arm_commit_counter(session)
            pruned = _prune(
                session,
                {PRUNE_SOURCE},
                set(),
                images=False,
                summary=summary,
                page_dir=None,
                prune_batch=prune_batch,
            )
            return pruned, summary, len(commits)
        finally:
            session.close()
            # The FK cascade did the child work in EVERY transaction the
            # prune opened, not just the last one.
            assert db.scalar(
                select(func.count())
                .select_from(Chunk)
                .where(Chunk.document_id.in_(ids))
            ) == 0

    pruned, summary, commits = run(PRUNE_BATCH)
    assert pruned == PRUNE_DOC_COUNT
    assert commits == PRUNE_DOC_COUNT // PRUNE_BATCH, (
        f"a {PRUNE_DOC_COUNT}-doc prune at batch {PRUNE_BATCH} ran "
        f"{commits} commits — expected one per {PRUNE_BATCH} deletes"
    )
    assert sorted(summary.changed_paths) == sorted(
        (PRUNE_SOURCE, f"docs/doc-{n:04d}.md") for n in range(PRUNE_DOC_COUNT)
    )
    assert db.scalar(
        select(func.count()).select_from(Document).where(Document.source == PRUNE_SOURCE)
    ) == 0

    # …and the pre-phase run does the SAME work in ONE transaction.
    _truncate(db)
    pruned_0, summary_0, commits_0 = run(0)
    assert pruned_0 == pruned == PRUNE_DOC_COUNT
    assert commits_0 == 1, (
        f"prune_batch=0 ran {commits_0} commits — the rollback knob must "
        "reproduce the pre-phase single transaction exactly"
    )
    assert sorted(summary_0.changed_paths) == sorted(summary.changed_paths)
    assert db.scalar(
        select(func.count()).select_from(Document).where(Document.source == PRUNE_SOURCE)
    ) == 0


def test_prune_batch_setting_reaches_the_prune(
    db: Session, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The caller passes ``sync_prune_batch`` — the knob is not dead code.

    ``_prune`` takes the cadence as a keyword (it has no ``llm`` argument),
    so this pins the ONE wire from ``Settings`` to the loop: a real sync of a
    source with three files deleted, observed through a spy on ``_prune``.
    The spy records the keyword and forwards to the real function, so the
    run's numbers stay the real ones (``pruned=3``) — and the survivors stay
    fully embedded, which is what proves the batched deletes were durable
    rather than parked in a transaction that never ran.
    """
    root = tmp_path / PRUNE_SOURCE
    root.mkdir()
    for n in range(4):
        (root / f"keep-{n}.md").write_text(
            f"# Keep {n}\n\nThe homelab runs podman quadlets on the hnb "
            f"host with quadlet units, nightly restic backups and a static "
            f"DNS zone for probe {n}.\n"
        )

    llm = _probe_llm()
    llm.settings.sync_prune_batch = 3
    first = asyncio.run(import_sources([root], llm, session=db))
    assert (first.files, first.added, first.pruned) == (4, 4, 0)

    for n in range(3):
        (root / f"keep-{n}.md").unlink()

    import app.rag.importer as importer

    seen_kwargs: list[object] = []
    real_prune = importer._prune

    def spy(*args: object, **kwargs: object) -> int:
        assert "prune_batch" in kwargs, (
            "import_sources did not pass Settings.sync_prune_batch to _prune"
        )
        seen_kwargs.append(kwargs["prune_batch"])
        return real_prune(*args, **kwargs)  # type: ignore[operator]

    monkeypatch.setattr(importer, "_prune", spy)
    second = asyncio.run(import_sources([root], llm, prune=True, session=db))

    assert seen_kwargs == [3], (
        "import_sources did not pass Settings.sync_prune_batch to _prune"
    )
    assert (second.files, second.pruned, second.errors) == (1, 3, 0)
    assert sorted(path for (_src, path) in second.changed_paths) == [
        "keep-0.md",
        "keep-1.md",
        "keep-2.md",
    ]
    assert [path for (path,) in db.execute(select(Document.path))] == ["keep-3.md"]
    # The survivor kept every chunk; the pruned docs left no orphans.
    embedded = db.scalar(
        select(func.count()).select_from(Chunk).where(Chunk.embedding.is_not(None))
    )
    assert embedded is not None and embedded > 0
    assert db.scalar(
        select(func.count())
        .select_from(Chunk)
        .where(Chunk.document_id.not_in(select(Document.id)))
    ) == 0


def test_prune_guards_unaffected_with_light_rows(
    db: Session, tmp_path: Path
) -> None:
    """The guards run off SIX columns — the light row is COMPLETE.

    A column missing from ``load_only`` does not announce itself: the guard
    that needs it simply emits one lazy SELECT per row and carries on, so
    the bug would show up as a walk that is quietly unbounded again. This
    test drives BOTH prune guards over light rows and pins the survivors:

    * phase 122 (images OFF): an ``is_image`` doc whose file is gone
      SURVIVES — ``is_image`` decides it;
    * phase 126 (the embedded image's parent rule, toggle ON): a
      ``<pdf>#page=N#img=M`` doc SURVIVES while its parent PDF's rel is in
      *seen* — ``path`` decides it — while a standalone image whose file IS
      gone prunes and takes its ``image_dir`` copy with it, which reads
      ``image_path`` (:func:`app.rag.importer._delete_image_copy`).

    Both toggles run over the same five rows, one run after the other (the
    way two syncs of the same source actually happen): toggle OFF prunes ONE
    document (the missing markdown — and leaves BOTH copies alone), toggle ON
    then prunes the ONE row that is still there and still must go (the
    standalone image, WITH its copy), the embedded image surviving in both.
    The exact pre-phase guard semantics, on a row that never loaded
    ``content``.
    """
    image_dir = tmp_path / "image_dir"
    image_dir.mkdir()
    gone_copy = image_dir / "gone.png"
    kept_copy = image_dir / "parent.png"
    gone_copy.write_bytes(b"\x89PNG\r\n\x1a\n gone")
    kept_copy.write_bytes(b"\x89PNG\r\n\x1a\n parent")

    def add(path: str, *, is_image: bool, copy: Path | None = None) -> None:
        db.add(
            Document(
                source=PRUNE_SOURCE,
                path=path,
                full_path=f"/srv/brain/{PRUNE_SOURCE}/{path}",
                title=f"Guard probe {path}",
                content=PRUNE_CONTENT,
                content_hash=f"{abs(hash(path)):064x}"[:64],
                is_image=is_image,
                image_path=str(copy) if copy else None,
            )
        )

    add("notes/kept.md", is_image=False)
    add("notes/gone.md", is_image=False)
    add("img/gone.png", is_image=True, copy=gone_copy)
    add("deck.pdf", is_image=False)
    add("deck.pdf#page=1#img=0.png", is_image=True, copy=kept_copy)
    db.commit()

    seen = {(PRUNE_SOURCE, "notes/kept.md"), (PRUNE_SOURCE, "deck.pdf")}
    session = SessionLocal()
    try:
        pruned_toggle_off = _prune(
            session, {PRUNE_SOURCE}, seen, images=False, prune_batch=0
        )
    finally:
        session.close()
    # Toggle OFF: BOTH image docs are invisible to the walk, not deleted —
    # and neither copy goes with them.
    assert pruned_toggle_off == 1
    assert gone_copy.exists() and kept_copy.exists()

    session = SessionLocal()
    try:
        pruned_toggle_on = _prune(
            session, {PRUNE_SOURCE}, seen, images=True, prune_batch=0
        )
    finally:
        session.close()
    # Toggle ON: the deleted image file prunes its row AND its copy (the
    # markdown went in the run above, so this run's single delete is the
    # image's); the embedded image survives on its parent PDF's rel.
    assert pruned_toggle_on == 1
    assert not gone_copy.exists()
    assert kept_copy.exists()
    assert sorted(
        path
        for (path,) in db.execute(
            select(Document.path).where(Document.source == PRUNE_SOURCE)
        )
    ) == ["deck.pdf", "deck.pdf#page=1#img=0.png", "notes/kept.md"]