All gates green. Report: **Implemented / verified (phase 185)** - `_doc_row_stmt` is the ONE deferral helper for all four fast-path row loads; only `Document.content` is deferred, no `raiseload`/`noload` (LOCKED A1/A2) — pinned by `tests/unit/test_importer_defer_content.py`. - `tests/integration/test_import_fast_path_columns.py` (5), `tests/e2e/test_fast_path_deferred_content.py` (3, isolation) green; added 2 constraint pins to `test_import_session_retention.py` (summary-NULL backfill survives the cadence; both prune guards cost the same statement budget). - **Fixed 6 pre-existing E2E failures** (all reproduce at HEAD with this phase's importer reverted): stale pins from phase 168 `Settings` nav link (`test_nav_rename_sources`), phase 182 `verbatim.js` script count (`test_steering`), phase 175 `/api/chats?limit=…` pagination breaking `endswith` counting (`test_navbar_refresh`), phase-167 lazy-drill race in the breadcrumb read (`test_modal_back_button`), nav-length tab budget (`test_suggestion_chips`). - **Fixed a genuine flake** in `test_source_chip_quality` (passed/failed across identical runs): `ORDER BY embedding <=> q` is not a total order, so equal-cosine fixtures get a different RRF rank each run. Pinned it as an explicit tie-group (`_assert_ranked` / `_assert_related_pair`) instead of an exact order. **Tests / lint (all 100% green)** - `uv run pytest --cov=app --cov-report=term-missing` → **4519 passed**, `app/rag/importer.py` **99%**, `TOTAL` **99%** (>90%). - `uv run ruff check .` → All checks passed; `uv run pyright` → 0 errors. - Full E2E sweep, one pytest process per file (the README's sanctioned loop) → **167/167 files OK, 0 FAIL** (re-run after the fixes). **Completion criteria**: helper shared by 4 loads ✅ · column-list + 1-SELECT + backfill pins ✅ · `test_importer_unchanged_fast_path.py` 24 passed, zero expectation edits ✅ · full suite + coverage + lint/types ✅ · phase E2E in isolation ✅ · docs accurate ✅ · commit left to the harness (no git commands run). **Documentation**: README + AGENTS.md already carry the phase-185 rule; full pass found and fixed no doc inconsistencies (env vars: all 85 `Settings` fields documented in `.env.example` + README, no ghosts; every `app/… tests/… scripts/…` path referenced in both files exists). **Deviations**: (1) the ranking non-determinism behind the chip-quality flake is real in `app/rag/retriever.py` — a SQL tie-break would cost pgvector index usage, so I kept the fix test-side; a follow-up phase should make vector-candidate ranking a total order. (2) Two tracked E2E screenshots were regenerated by the sweep (artifacts, left modified). **Next pending phase**: `186_git_log_streaming_dates`.
1370 lines
58 KiB
Python
1370 lines
58 KiB
Python
"""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)), (
|
||
f"expected exactly one document retained per file already walked, 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 {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_the_summary_backfill_survives_the_reset_cadence(db: Session, tmp_path: Path) -> None:
|
||
"""The constraint that fixes WHERE the cadence may sit, in ONE file.
|
||
|
||
``_unchanged_doc`` (phase 118) backfills a NULL summary by calling
|
||
:func:`app.rag.summarizer.summarize_document`, which reads
|
||
``doc.content`` AFTER the row was loaded and its ``file_hash`` stamped.
|
||
That local is live across statements: a
|
||
``session.expunge_all()`` moved into ``_unchanged_doc`` — or a
|
||
``noload``/``raiseload`` added to the row it walks — would discard or
|
||
break the very row the backfill reads. Phase 185 task 01 chose
|
||
``defer`` over ``noload``/``raiseload`` (LOCKED A2) for the same reason,
|
||
and task 05 of THIS phase pinned the choice on the statement itself
|
||
(``tests/unit/test_importer_defer_content.py``); this pins the SAME
|
||
constraint on the run.
|
||
|
||
The fixture is a document whose ``is_summary`` chunk is missing while
|
||
its ``file_hash`` matches (a re-sync with a NULL summary — exactly the
|
||
row the backfill exists for), imported at the tightest cadence the
|
||
setting allows that still resets inside a 12-file walk. Three
|
||
observables, none of them a memory proxy:
|
||
|
||
* the backfill RUNS — one and only one ``is_summary`` chunk appears,
|
||
carrying the fake-embedder's summary text;
|
||
* every other file is still ``unchanged`` and the summary count is
|
||
exactly the file count afterwards, i.e. the reset cadence destroyed
|
||
no summary on the way through;
|
||
* the doc's own text is still intact afterwards (``content`` was never
|
||
rewritten by a re-index).
|
||
"""
|
||
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()
|
||
|
||
# Hand the next sync ONE document with a NULL summary: it takes
|
||
# ``_unchanged_doc``'s backfill branch instead of the early return.
|
||
probe_rel = "probe-00.md"
|
||
before = _kb_snapshot(db)[probe_rel]
|
||
target = db.execute(
|
||
select(Document).where(Document.source == PROBE_SOURCE, Document.path == probe_rel)
|
||
).scalar_one()
|
||
db.query(Chunk).filter(Chunk.document_id == target.id, Chunk.is_summary.is_(True)).delete()
|
||
target.summary = None
|
||
db.commit()
|
||
db.expunge_all()
|
||
assert _kb_snapshot(db)[probe_rel][1] is None
|
||
|
||
session = SessionLocal()
|
||
try:
|
||
summary = asyncio.run(
|
||
import_sources(
|
||
[root], _probe_llm(reset_files=RESET_FILES), prune=False, session=session
|
||
)
|
||
)
|
||
finally:
|
||
session.close()
|
||
# Everything stayed unchanged — the backfill never re-indexed.
|
||
assert (summary.files, summary.added, summary.updated, summary.unchanged) == (
|
||
FILE_COUNT,
|
||
0,
|
||
0,
|
||
FILE_COUNT,
|
||
)
|
||
assert (summary.errors, summary.summary_backfilled) == (0, 1)
|
||
|
||
snap = _kb_snapshot(db)
|
||
backfilled = snap[probe_rel]
|
||
assert backfilled[1] is not None, "the summary-NULL backfill did not run"
|
||
# The one doc that needed the deferred read still has its OWN text and
|
||
# its own hash: a summary was added, the document was not rewritten.
|
||
assert backfilled[0] == before[0]
|
||
assert backfilled[2] == before[2]
|
||
# And the cadence destroyed nothing on its way through: every file is
|
||
# back to 5 content chunks + 1 summary chunk.
|
||
assert {p: c for p, (_, _, _, c) in snap.items()} == {
|
||
f"probe-{n:02d}.md": 6 for n in range(FILE_COUNT)
|
||
}
|
||
|
||
|
||
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"]
|
||
|
||
|
||
def test_both_prune_guards_cost_the_same_statement_budget(
|
||
db: Session, tmp_path: Path, statement_capture: list[str]
|
||
) -> None:
|
||
"""The guards add NO hidden read — proved by statement count, both ways.
|
||
|
||
:func:`test_prune_guards_unaffected_with_light_rows` above proves the
|
||
guards DECIDE correctly off the six light columns. This pins the other
|
||
half of the same contract, the half that is invisible in behaviour: a
|
||
guard reaching for a deferred attribute would still decide correctly and
|
||
only show up as ONE MORE SELECT PER ROW. So the assertion here is a
|
||
statement count, not an outcome:
|
||
|
||
* the SAME projection runs under both toggles — one candidate walk
|
||
apiece, naming the same six columns (images OFF only skips rows
|
||
through ``is_image``, a column the walk already reads);
|
||
* neither walk projects ``documents.content``;
|
||
* deleting an image row + copying its file costs no extra document
|
||
SELECT (the ``image_path`` the copy cleanup uses is in the projection).
|
||
|
||
Total: exactly TWO document SELECTs for the two syncs, whatever the
|
||
guards do.
|
||
"""
|
||
image_dir = tmp_path / "image_dir"
|
||
image_dir.mkdir()
|
||
gone_copy = image_dir / "gone.png"
|
||
gone_copy.write_bytes(b"\x89PNG\r\n\x1a\n gone")
|
||
|
||
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"Budget 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#page=1#img=0.png", is_image=True)
|
||
db.commit()
|
||
|
||
seen = {(PRUNE_SOURCE, "notes/kept.md")}
|
||
projections: list[frozenset[str]] = []
|
||
for toggle in (False, True):
|
||
since = len(statement_capture)
|
||
session = SessionLocal()
|
||
try:
|
||
_prune(session, {PRUNE_SOURCE}, seen, images=toggle, prune_batch=0)
|
||
finally:
|
||
session.close()
|
||
# Only what THIS sync emitted — the capture is cumulative.
|
||
walks = [
|
||
s for s in _doc_selects(statement_capture[since:]) if "WHERE documents.source" in s
|
||
]
|
||
assert len(walks) == 1, (
|
||
f"the images={toggle} prune ran {len(walks)} candidate walks; "
|
||
"one per sync is the budget (a second is a per-row lazy load)"
|
||
)
|
||
projections.append(_projected_columns(walks[0]))
|
||
|
||
assert projections[0] == projections[1], (
|
||
"the two toggles read DIFFERENT columns — a guard grew a deferred "
|
||
f"read:\n off: {sorted(projections[0])}\n on: {sorted(projections[1])}"
|
||
)
|
||
assert "documents.content" not in projections[0]
|
||
assert len(_doc_selects(statement_capture)) == 2, (
|
||
"two syncs must cost exactly two document SELECTs in total, guards "
|
||
"included — any extra is a per-row read the light row must cover"
|
||
)
|