FabledGrimoire/tests/test_backfill_reads_current_text.py

"""A backfill repairs to current truth; it does not replay a snapshot.

The failure these pin: a backfill that SELECTs the text alongside the id and
then iterates that snapshot, sleeping between records. A run over a real corpus
takes minutes, so anything edited inside that window is re-embedded by its own
write path and then OVERWRITTEN with the pre-edit text the scan captured — and
stamped at the current chunker version, so the next boot considers it current
and never repairs it. The record stays findable by what it used to say,
silently, until someone happens to edit it again.

That window is exactly "forward work continues while a long backfill runs",
which is the normal condition, not an unlucky one.
"""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from grimoire.services import embeddings as emb


def _ctx(session):
    ctx = MagicMock()
    ctx.__aenter__ = AsyncMock(return_value=session)
    ctx.__aexit__ = AsyncMock(return_value=False)
    return ctx


def _scan_session(all_ids, current_ids, stale_ids=(), logged_ids=()):
    """The opening scan: ids at the current version, all ids, and the two
    staleness queries. Order matches backfill_note_embeddings."""
    def _rows(values):
        r = MagicMock()
        r.fetchall.return_value = [(v,) for v in values]
        return r

    session = MagicMock()
    session.execute = AsyncMock(side_effect=[
        _rows(current_ids),     # note_ids already at CHUNKER_VERSION
        _rows(all_ids),         # every note id
        _rows(stale_ids),       # vectors older than the record
        _rows(logged_ids),      # tasks logged since their vectors
    ])
    return session


# --- the fix: text is read when the record is embedded, not when scanned -----


@pytest.mark.asyncio
async def test_the_backfill_embeds_the_text_as_it_is_now_not_as_it_was_scanned():
    """THE regression. The scan sees the old body; by the time the loop reaches
    the note, someone has edited it. The vectors must encode the edit."""
    scan = _scan_session(all_ids=[1], current_ids=[])
    edited = ("the body AFTER the edit",)

    with (
        patch.object(emb, "async_session", return_value=_ctx(scan)),
        patch.object(emb, "_current_row",
                     AsyncMock(return_value=(42, "T", *edited, "note", None))),
        patch.object(emb, "upsert_note_embedding", AsyncMock()) as upsert,
        patch.object(emb.asyncio, "sleep", AsyncMock()),
    ):
        await emb.backfill_note_embeddings()

    upsert.assert_awaited_once()
    assert upsert.await_args.args[3] == "the body AFTER the edit"


@pytest.mark.asyncio
async def test_a_record_deleted_between_the_scan_and_the_loop_is_skipped():
    """An ordinary outcome of a long run, not an error — and embedding a row
    that no longer exists is either a FK violation or a resurrected chunk."""
    scan = _scan_session(all_ids=[1, 2], current_ids=[])

    with (
        patch.object(emb, "async_session", return_value=_ctx(scan)),
        patch.object(emb, "_current_row",
                     AsyncMock(side_effect=[None, (42, "T", "body", "note", None)])),
        patch.object(emb, "upsert_note_embedding", AsyncMock()) as upsert,
        patch.object(emb.asyncio, "sleep", AsyncMock()),
    ):
        await emb.backfill_note_embeddings()

    assert upsert.await_count == 1
    assert upsert.await_args.args[0] == 2


# --- the repair: stale vectors are re-embedded even at the current version ---


@pytest.mark.asyncio
async def test_a_record_whose_text_outran_its_vectors_is_re_embedded():
    """The repair half: a guard that refuses to write a bad value does not
    undo the bad values already stored. Note 5 IS at the current version,
    so the version comparison alone would skip it forever; its text is newer
    than its vectors, which is the only signal that separates "embedded from
    stale text" from "embedded correctly"."""
    scan = _scan_session(all_ids=[5], current_ids=[5], stale_ids=[5])

    with (
        patch.object(emb, "async_session", return_value=_ctx(scan)),
        patch.object(emb, "_current_row", AsyncMock(return_value=(42, "T", "b", "note", None))),
        patch.object(emb, "upsert_note_embedding", AsyncMock()) as upsert,
        patch.object(emb.asyncio, "sleep", AsyncMock()),
    ):
        await emb.backfill_note_embeddings()

    upsert.assert_awaited_once()
    assert upsert.await_args.args[0] == 5


@pytest.mark.asyncio
async def test_a_task_logged_since_its_vectors_is_re_embedded():
    """A work log is part of a task's document but lives in its own table, so
    writing one never moves `notes.updated_at` — the timestamp comparison is
    blind to it. These are the records the race was MOST likely to hit, because
    a session writing work logs is what "forward work" means."""
    scan = _scan_session(all_ids=[9], current_ids=[9], stale_ids=[], logged_ids=[9])

    with (
        patch.object(emb, "async_session", return_value=_ctx(scan)),
        patch.object(emb, "_current_row", AsyncMock(return_value=(42, "T", "b", "note", None))),
        patch.object(emb, "upsert_note_embedding", AsyncMock()) as upsert,
        patch.object(emb.asyncio, "sleep", AsyncMock()),
    ):
        await emb.backfill_note_embeddings()

    upsert.assert_awaited_once()
    assert upsert.await_args.args[0] == 9


@pytest.mark.asyncio
async def test_a_corpus_that_is_genuinely_current_embeds_nothing():
    """The guard must be able to stay quiet, or it is not a guard — and
    a backfill that re-embeds everything every boot is a different bug."""
    scan = _scan_session(all_ids=[1, 2, 3], current_ids=[1, 2, 3])

    with (
        patch.object(emb, "async_session", return_value=_ctx(scan)),
        patch.object(emb, "_current_row", AsyncMock()) as fresh,
        patch.object(emb, "upsert_note_embedding", AsyncMock()) as upsert,
        patch.object(emb.asyncio, "sleep", AsyncMock()),
    ):
        await emb.backfill_note_embeddings()

    upsert.assert_not_awaited()
    fresh.assert_not_awaited()


# --- the scan must not carry text at all -------------------------------------


def test_no_backfill_selects_a_records_text_in_its_opening_scan():
    """A structural guard, because this is the shape of the bug rather than one
    instance of it: all four backfills had it, and the fourth was written by
    copying the third. A scan that carries text is a snapshot that goes stale
    over the length of the run."""
    import ast
    import inspect

    source = inspect.getsource(emb)
    tree = ast.parse(source)
    text_columns = {
        "title", "body", "statement", "when_to_apply",
        "description", "name",
    }
    offenders = []
    for node in ast.walk(tree):
        if not (isinstance(node, ast.AsyncFunctionDef)
                and node.name.startswith("backfill_")):
            continue
        # Only the OPENING scan is at issue — it is the code inside the
        # `async with async_session()` block, before the embed loop.
        for inner in ast.walk(node):
            if not (isinstance(inner, ast.Call)
                    and getattr(inner.func, "id", None) == "select"):
                continue
            for arg in inner.args:
                if (isinstance(arg, ast.Attribute)
                        and arg.attr in text_columns):
                    offenders.append(f"{node.name}: select(...{arg.attr})")

    assert not offenders, (
        "a backfill scan is carrying record text, which goes stale over the "
        f"length of the run: {offenders}"
    )