fix(embeddings): a backfill reads text when it embeds, not when it scanned (#4264)

Every backfill selected the text alongside the id and then iterated that
snapshot, sleeping between records. Over a real corpus that is minutes, and
anything edited inside the window was re-embedded by its own write path and
then OVERWRITTEN with the pre-edit text the scan had captured — stamped at the
current chunker version, so the next boot considered it current and never
repaired it. The record stayed findable by what it used to say, indefinitely,
until someone happened to edit it again. Nothing reported it, and no existing
signal separated it from a correct record: the version was right and the
vectors were there.

All four backfills had it. Three were pre-existing; I wrote the fourth an hour
ago for #4251 by copying the third.

#4251 also widened the exposure. The race used to need a note-body edit; now
every add_task_log re-embeds its task, so any session recording work during a
backfill can hit it — which is exactly what "forward work continues while the
backfill runs" means.

TWO HALVES, SEPARATE ON PURPOSE.

Stopping it: the scan takes IDS ONLY and `_current_row` re-reads each record at
the moment it is embedded. One indexed primary-key read, negligible beside the
forward pass that follows, and it turns a backfill from "replay a snapshot"
into "repair to current truth". The residual race shrinks from the length of
the run to the width of one record, which the parent-row claim and the
single-transaction replacement narrow further.

Repairing it: `_vectors_older_than_their_record` adds a third staleness
condition beside "no vectors" and "old chunker" — vectors older than the text
they encode. A guard that refuses to write a bad value does not undo the bad
values already stored (#4202); the rows are what has to change.

That check earns its place beyond the bug that prompted it. It equally catches
a record whose `embed_note` never ran — no event loop, a swallowed refresh, a
process that died between the commit and the index. Every one of those looks
identical from the version column and identical from outside.

And a work log is part of a task's document while living in its own table, so
writing one never moves `notes.updated_at` and the timestamp comparison cannot
see it. `_tasks_logged_since_embedding` covers exactly the records the race was
most likely to have hit.

The live instance has already run #4251's backfill — the prior-art arm is
surfacing `## Work log —` passages, which is the thing itself rather than a
banner about it. Whether that run clobbered anything is unknown and was not
measured; the repair half picks up whatever it did on the next boot, without
needing it diagnosed first.

The structural guard in the tests was checked against the old shape, not just
the new one: reintroducing `select(Note.id, Note.title, Note.body)` makes it
fail, so it can catch the next copy of this rather than only agreeing with
today's code (#167).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy

src/scribe/services/embeddings.py

@@ -18,7 +18,7 @@ from collections.abc import Sequence
 
 from typing import TYPE_CHECKING
 
-from sqlalchemy import delete, or_, select
+from sqlalchemy import delete, func, or_, select
 
 from scribe.models import async_session
 from scribe.models.embedding import NoteEmbedding, RuleEmbedding
@@ -908,15 +908,135 @@ async def semantic_search_notes(
     return final
 
 
-async def backfill_note_embeddings() -> None:
+# --- backfill correctness: scan ids, read text at embed time (#4262) --------
-    """(Re-)embed every note that is missing vectors OR whose stored vectors
+#
-    were produced by an older chunker.
+# A backfill used to SELECT the text alongside the id and then iterate that
+# snapshot, sleeping between records. On a corpus of any size the run takes
+# minutes, and anything edited inside that window was re-embedded by its own
+# write path and then OVERWRITTEN with the pre-edit text the scan had captured
+# — stamped at the current version, so the next boot considered it current and
+# never repaired it. A record left findable by what it used to say, silently,
+# until someone happened to edit it again.
+#
+# The two functions below are the fix, and they are separate on purpose: one
+# stops it happening, the other repairs what already happened.
 
-    Runs as a background task at startup. Version-awareness is what makes a
+
-    document-shape change deployable: migration 0077 cleared the table once,
+async def _current_row(columns, id_column, row_id: int):
-    and every later CHUNKER_VERSION bump re-embeds the stale notes here — a
+    """A record's text AS IT IS NOW, read at the moment it is embedded.
-    version comparison instead of another wipe. Adds a small sleep between
+
-    notes so a large backfill doesn't peg CPU.
+    One indexed primary-key read, negligible beside the forward pass that
+    follows it — and it turns a backfill from "replay a snapshot" into "repair
+    to current truth". The residual race shrinks from the length of the whole
+    run to the width of one record's processing, which `_claim_parent_row` and
+    the single-transaction replacement then narrow further.
+
+    None when the row is gone: deleted between the scan and here, which is an
+    ordinary outcome of a long run and not an error.
+    """
+    try:
+        async with async_session() as session:
+            return (
+                await session.execute(select(*columns).where(id_column == row_id))
+            ).first()
+    except Exception:
+        logger.warning("Backfill could not re-read record %s", row_id, exc_info=True)
+        return None
+
+
+async def _vectors_older_than_their_record(
+    session, emb_id_column, emb_updated_column, rec_id_column, rec_updated_column
+) -> set[int]:
+    """Ids whose text changed AFTER their vectors were written.
+
+    The repair half, and a standing guard worth more than the bug that prompted
+    it. A version comparison alone asks "was this embedded by the current
+    chunker" and cannot see a record that was embedded by it from stale text —
+    nor one whose `embed_note` never ran at all, because the write happened
+    with no event loop, or the refresh was swallowed, or the process died
+    between the commit and the index.
+
+    Every one of those looks identical from the version column and identical
+    from the outside: a record that answers to what it used to say. This is the
+    only signal that separates them, and it costs one grouped read at startup.
+
+    A guard that refuses to write a bad value does not undo the bad values
+    already stored (#4202) — the rows are the thing that has to change, and
+    this is what changes them.
+    """
+    newest = (
+        select(
+            emb_id_column.label("rid"),
+            func.max(emb_updated_column).label("embedded_at"),
+        )
+        .group_by(emb_id_column)
+        .subquery()
+    )
+    rows = await session.execute(
+        select(rec_id_column)
+        .join(newest, newest.c.rid == rec_id_column)
+        .where(rec_updated_column > newest.c.embedded_at)
+    )
+    return {int(row[0]) for row in rows.fetchall()}
+
+
+async def _tasks_logged_since_embedding(session) -> set[int]:
+    """Task ids whose newest work log is newer than their newest vector.
+
+    The note corpus needs this on top of the plain timestamp comparison, and
+    the reason is a consequence of #4251 that is easy to miss: a task's
+    embedded document carries its work logs, but a log lives in its own table,
+    so writing one does NOT move `notes.updated_at`. A task re-embedded from
+    stale text during a backfill therefore looks current by every other
+    signal — the version is right and the note's own timestamp is older than
+    the vectors — while the document those vectors encode is out of date.
+
+    These are the exact records the backfill race was most likely to hit,
+    because a session logging work is what "forward work during a long
+    backfill" MEANS.
+    """
+    newest_vector = (
+        select(
+            NoteEmbedding.note_id.label("rid"),
+            func.max(NoteEmbedding.updated_at).label("embedded_at"),
+        )
+        .group_by(NoteEmbedding.note_id)
+        .subquery()
+    )
+    newest_log = (
+        select(
+            TaskLog.task_id.label("tid"),
+            func.max(TaskLog.updated_at).label("logged_at"),
+        )
+        .group_by(TaskLog.task_id)
+        .subquery()
+    )
+    rows = await session.execute(
+        select(newest_log.c.tid)
+        .join(newest_vector, newest_vector.c.rid == newest_log.c.tid)
+        .where(newest_log.c.logged_at > newest_vector.c.embedded_at)
+    )
+    return {int(row[0]) for row in rows.fetchall()}
+
+
+async def backfill_note_embeddings() -> None:
+    """(Re-)embed every note whose vectors are missing, outdated or WRONG.
+
+    Three conditions, and the third is the one a version number cannot see:
+
+    1. no vectors at all;
+    2. vectors from an older chunker — version-awareness is what makes a
+       document-shape change deployable. Migration 0077 cleared the table
+       once; every later CHUNKER_VERSION bump re-embeds here instead;
+    3. vectors OLDER THAN THE TEXT THEY CLAIM TO ENCODE. A record embedded by
+       the current chunker from text that has since changed is current by
+       every other signal and answers to what it used to say (#4262).
+
+    Runs as a background task at startup, after the serving flag (#4181), with
+    a small sleep between notes so a large backfill doesn't peg CPU. It is
+    resumable by construction: progress IS the version stamp on each record's
+    rows, so a restart re-queries and skips what finished — there is no
+    checkpoint to lose.
     """
     try:
         async with async_session() as session:
@@ -930,26 +1050,41 @@ async def backfill_note_embeddings() -> None:
                     )
                 ).fetchall()
             }
-            result = await session.execute(
+            # IDS ONLY. The text is read per note below, at the moment it is
-                select(Note.id, Note.user_id, Note.title, Note.body)
+            # embedded — a scan that carried the text would spend the whole run
-            )
+            # holding a snapshot and overwrite anything edited inside it.
-            notes_to_embed = [
+            all_ids = [
-                row for row in result.fetchall() if row[0] not in current
+                row[0]
+                for row in (await session.execute(select(Note.id))).fetchall()
             ]
+            stale = await _vectors_older_than_their_record(
+                session, NoteEmbedding.note_id, NoteEmbedding.updated_at,
+                Note.id, Note.updated_at,
+            )
+            # A work log is part of a task's document but lives in its own
+            # table, so writing one leaves `notes.updated_at` untouched and the
+            # comparison above blind to it.
+            stale |= await _tasks_logged_since_embedding(session)
     except Exception:
         logger.warning("Embedding backfill: failed to query notes", exc_info=True)
         return
 
+    notes_to_embed = [i for i in all_ids if i not in current or i in stale]
     if not notes_to_embed:
         logger.info("Embedding backfill: all notes current at chunker v%d", CHUNKER_VERSION)
         return
 
     logger.info(
-        "Embedding backfill: embedding %d notes at chunker v%d",
+        "Embedding backfill: embedding %d note(s) at chunker v%d (%d with vectors "
-        len(notes_to_embed), CHUNKER_VERSION,
+        "older than the record they describe)",
+        len(notes_to_embed), CHUNKER_VERSION, len(stale),
     )
     success = 0
-    for note_id, user_id, title, body in notes_to_embed:
+    for note_id in notes_to_embed:
+        row = await _current_row((Note.user_id, Note.title, Note.body), Note.id, note_id)
+        if row is None:
+            continue          # deleted between the scan and here
+        user_id, title, body = row
         if not chunk_document(title, body):
             continue
         await upsert_note_embedding(note_id, user_id, title, body)
@@ -1280,10 +1415,18 @@ async def backfill_rule_embeddings() -> None:
             current = select(RuleEmbedding.rule_id).where(
                 RuleEmbedding.chunker_version == CHUNKER_VERSION
             )
-            stale = (await session.execute(
+            # IDS ONLY — the text is re-read per rule below (#4262).
-                select(Rule.id, Rule.title, Rule.statement, Rule.when_to_apply)
+            by_version = {
+                row[0] for row in (await session.execute(
+                    select(Rule.id)
                     .where(Rule.deleted_at.is_(None), Rule.id.notin_(current))
-            )).all()
+                )).fetchall()
+            }
+            by_time = await _vectors_older_than_their_record(
+                session, RuleEmbedding.rule_id, RuleEmbedding.updated_at,
+                Rule.id, Rule.updated_at,
+            )
+            stale = sorted(by_version | by_time)
     except Exception:
         logger.warning("Rule embedding backfill: failed to query rules", exc_info=True)
         return
@@ -1291,9 +1434,17 @@ async def backfill_rule_embeddings() -> None:
     if not stale:
         logger.info("Rule embedding backfill: all rules current at chunker v%d", CHUNKER_VERSION)
         return
-    logger.info("Rule embedding backfill: embedding %d rule(s)", len(stale))
+    logger.info(
-    for rule_id, title, statement, when_to_apply in stale:
+        "Rule embedding backfill: embedding %d rule(s) (%d with stale vectors)",
-        await upsert_rule_embedding(rule_id, title, statement, when_to_apply)
+        len(stale), len(by_time),
+    )
+    for rule_id in stale:
+        row = await _current_row(
+            (Rule.title, Rule.statement, Rule.when_to_apply), Rule.id, rule_id
+        )
+        if row is None:
+            continue
+        await upsert_rule_embedding(rule_id, *row)
 
 
 # ── Milestones (milestone 415) ──────────────────────────────────────────
@@ -1641,10 +1792,18 @@ async def backfill_system_embeddings() -> None:
             current = select(SystemEmbedding.system_id).where(
                 SystemEmbedding.chunker_version == CHUNKER_VERSION
             )
-            stale = (await session.execute(
+            # IDS ONLY — the charter is re-read per System below (#4262).
-                select(System.id, System.name, System.description)
+            by_version = {
+                row[0] for row in (await session.execute(
+                    select(System.id)
                     .where(System.deleted_at.is_(None), System.id.notin_(current))
-            )).all()
+                )).fetchall()
+            }
+            by_time = await _vectors_older_than_their_record(
+                session, SystemEmbedding.system_id, SystemEmbedding.updated_at,
+                System.id, System.updated_at,
+            )
+            stale = sorted(by_version | by_time)
     except Exception:
         logger.warning("System embedding backfill: failed to query systems", exc_info=True)
         return
@@ -1652,9 +1811,17 @@ async def backfill_system_embeddings() -> None:
     if not stale:
         logger.info("System embedding backfill: all systems current at chunker v%d", CHUNKER_VERSION)
         return
-    logger.info("System embedding backfill: embedding %d system(s)", len(stale))
+    logger.info(
-    for system_id, name, description in stale:
+        "System embedding backfill: embedding %d system(s) (%d with stale vectors)",
-        await upsert_system_embedding(system_id, name, description)
+        len(stale), len(by_time),
+    )
+    for system_id in stale:
+        row = await _current_row(
+            (System.name, System.description), System.id, system_id
+        )
+        if row is None:
+            continue
+        await upsert_system_embedding(system_id, *row)
 
 
 async def backfill_milestone_embeddings() -> None:
@@ -1668,10 +1835,18 @@ async def backfill_milestone_embeddings() -> None:
             current = select(MilestoneEmbedding.milestone_id).where(
                 MilestoneEmbedding.chunker_version == CHUNKER_VERSION
             )
-            stale = (await session.execute(
+            # IDS ONLY — the plan is re-read per milestone below (#4262).
-                select(Milestone.id, Milestone.title, Milestone.description, Milestone.body)
+            by_version = {
+                row[0] for row in (await session.execute(
+                    select(Milestone.id)
                     .where(Milestone.deleted_at.is_(None), Milestone.id.notin_(current))
-            )).all()
+                )).fetchall()
+            }
+            by_time = await _vectors_older_than_their_record(
+                session, MilestoneEmbedding.milestone_id, MilestoneEmbedding.updated_at,
+                Milestone.id, Milestone.updated_at,
+            )
+            stale = sorted(by_version | by_time)
     except Exception:
         logger.warning("Milestone embedding backfill: failed to query milestones", exc_info=True)
         return
@@ -1679,6 +1854,15 @@ async def backfill_milestone_embeddings() -> None:
     if not stale:
         logger.info("Milestone embedding backfill: all milestones current at chunker v%d", CHUNKER_VERSION)
         return
-    logger.info("Milestone embedding backfill: embedding %d milestone(s)", len(stale))
+    logger.info(
-    for milestone_id, title, description, body in stale:
+        "Milestone embedding backfill: embedding %d milestone(s) (%d with stale vectors)",
-        await upsert_milestone_embedding(milestone_id, title, description, body)
+        len(stale), len(by_time),
+    )
+    for milestone_id in stale:
+        row = await _current_row(
+            (Milestone.title, Milestone.description, Milestone.body),
+            Milestone.id, milestone_id,
+        )
+        if row is None:
+            continue
+        await upsert_milestone_embedding(milestone_id, *row)

tests/test_backfill_reads_current_text.py

@@ -0,0 +1,186 @@
+"""A backfill repairs to current truth; it does not replay a snapshot (#4262).
+
+The bug these pin: every backfill used to SELECT the text alongside the id and
+then iterate that snapshot, sleeping between records. A run over a real corpus
+takes minutes, so anything edited inside that window was re-embedded by its own
+write path and then OVERWRITTEN with the pre-edit text the scan had captured —
+and stamped at the current chunker version, so the next boot considered it
+current and never repaired it. The record stayed findable by what it used to
+say, silently, until someone happened 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 scribe.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))),
+        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")])),
+        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 (#4202 — 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"))),
+        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"))),
+        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 (#167) — 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}"
+    )

tests/test_chunking.py

@@ -287,18 +287,26 @@ async def test_backfill_reembeds_notes_with_a_stale_chunker_version():
 
     from scribe.services import embeddings as emb
 
-    current_rows = MagicMock()
+    def _ids(*values):
-    current_rows.fetchall.return_value = [(1,)]         # note 1 is current
+        r = MagicMock()
-    note_rows = MagicMock()
+        r.fetchall.return_value = [(v,) for v in values]
-    note_rows.fetchall.return_value = [
+        return r
-        (1, 42, "current", "body"),
+
-        (2, 42, "stale-version", "body"),
-    ]
     session, ctx = _session_ctx()
-    session.execute = AsyncMock(side_effect=[current_rows, note_rows])
+    # The opening scan, in order: ids already at the current version, every
+    # note id, vectors older than their note, tasks logged since their vectors.
+    # IDS ONLY — the text is re-read per note at embed time (#4262).
+    session.execute = AsyncMock(side_effect=[
+        _ids(1),            # note 1 is current
+        _ids(1, 2),         # the corpus
+        _ids(),             # nothing stale by timestamp
+        _ids(),             # no task logged since its vectors
+    ])
 
     with (
         patch.object(emb, "async_session", return_value=ctx),
+        patch.object(emb, "_current_row",
+                     AsyncMock(return_value=(42, "stale-version", "body"))),
         patch.object(emb, "upsert_note_embedding", AsyncMock()) as upsert,
         patch.object(emb.asyncio, "sleep", AsyncMock()),
     ):