feat(embeddings): best-chunk-per-note on every retrieval surface (#280 step 4)

A note's relevance is now its best chunk's similarity, everywhere:

- semantic_search_notes keeps the indexed raw-distance top-k and over-fetches
  chunk rows (x4, composing with the x3 supersession over-fetch), then
  collapses to first-appearance-per-note — rows arrive distance-ordered, so
  first is best. Every ranked consumer (MCP/REST search, Browse, auto-inject,
  write-path, gate) inherits through the one function.
- list_notes semantic q swaps its join for a correlated MIN-distance
  subquery — the join would have repeated a long note once per matching chunk
  and made total count chunks.
- the duplicate report groups its self-join by note pair on MIN(distance):
  pair similarity = closest chunk pair, and the < join now also drops
  cross-chunk self-pairs that would flag every long note against itself.
- the write gate queries once per chunk of the candidate (capped at 8), so a
  note duplicating an existing record in ONE SECTION is caught — the
  whole-document query diluted exactly the section that mattered.

Integration test now seeds a two-chunk note and pins the collapse against
real pgvector.

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

alembic/versions/0077_chunked_embeddings.py

@@ -0,0 +1,54 @@
+"""Chunked embeddings: one note_embeddings row per chunk (#280)
+
+Revision ID: 0077
+Revises: 0076
+Create Date: 2026-08-09
+
+The embedding model reads at most 512 tokens and fastembed truncates the rest
+silently, so the old one-row-per-note shape permanently lost everything past
+~400 words of a record. A note now stores one row per chunk of
+`embeddings.chunk_document`: PK (note_id, chunk_index), plus the chunk's text
+(inspectability + future "matched section" surfacing) and the chunker version
+that produced it (so later shape changes re-embed by version comparison
+instead of repeating this wipe).
+
+Embeddings are DERIVED data (0067 precedent): rows are cleared here and the
+startup backfill regenerates the whole corpus at the new shape on next boot.
+The HNSW index is untouched — it indexes chunk rows exactly as it indexed
+note rows.
+"""
+from alembic import op
+
+
+revision = "0077"
+down_revision = "0076"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Derived data — the version-aware startup backfill re-embeds everything
+    # at the chunked shape. Old whole-document rows would be indistinguishable
+    # from properly-chunked single-chunk notes, so they cannot be carried over.
+    op.execute("DELETE FROM note_embeddings")
+
+    # Empty table, so NOT NULL columns need no defaults and the PK swap is
+    # instant.
+    op.execute("ALTER TABLE note_embeddings ADD COLUMN chunk_index integer NOT NULL")
+    op.execute("ALTER TABLE note_embeddings ADD COLUMN chunk_text text NOT NULL")
+    op.execute("ALTER TABLE note_embeddings ADD COLUMN chunker_version integer NOT NULL")
+    op.execute("ALTER TABLE note_embeddings DROP CONSTRAINT note_embeddings_pkey")
+    op.execute(
+        "ALTER TABLE note_embeddings ADD PRIMARY KEY (note_id, chunk_index)"
+    )
+
+
+def downgrade() -> None:
+    # Same reasoning in reverse: chunk rows make no sense to a whole-document
+    # reader, so clear and let the old backfill regenerate.
+    op.execute("DELETE FROM note_embeddings")
+    op.execute("ALTER TABLE note_embeddings DROP CONSTRAINT note_embeddings_pkey")
+    op.execute("ALTER TABLE note_embeddings DROP COLUMN chunk_index")
+    op.execute("ALTER TABLE note_embeddings DROP COLUMN chunk_text")
+    op.execute("ALTER TABLE note_embeddings DROP COLUMN chunker_version")
+    op.execute("ALTER TABLE note_embeddings ADD PRIMARY KEY (note_id)")

src/scribe/models/embedding.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timezone
 
 from pgvector.sqlalchemy import Vector
-from sqlalchemy import DateTime, ForeignKey, Integer
+from sqlalchemy import DateTime, ForeignKey, Integer, Text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from scribe.models import Base
@@ -14,7 +14,15 @@ EMBEDDING_DIM = 384
 
 
 class NoteEmbedding(Base):
-    """Stores the embedding vector for a note, used for semantic search."""
+    """One embedding vector per CHUNK of a note (#280, migration 0077).
+
+    The model reads at most 512 tokens, so a single whole-document vector
+    permanently lost everything past ~400 words. A note now stores one row per
+    chunk of `embeddings.chunk_document`, and a query matches the note if it
+    matches ANY chunk — retrieval collapses rows to best-chunk-per-note.
+    A short note has exactly one row (chunk_index 0) whose text is the
+    historical `title\\nbody` shape.
+    """
 
     __tablename__ = "note_embeddings"
 
@@ -23,8 +31,16 @@ class NoteEmbedding(Base):
         ForeignKey("notes.id", ondelete="CASCADE"),
         primary_key=True,
     )
+    chunk_index: Mapped[int] = mapped_column(Integer, primary_key=True)
     user_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
     embedding: Mapped[list] = mapped_column(Vector(EMBEDDING_DIM), nullable=False)
+    # Exactly what this vector encodes — inspectable when a ranking surprises,
+    # and the hook for surfacing WHICH section matched, later.
+    chunk_text: Mapped[str] = mapped_column(Text, nullable=False)
+    # embeddings.CHUNKER_VERSION at write time. The startup backfill re-embeds
+    # any note whose rows carry a stale version — shape changes become a
+    # version bump instead of a table wipe.
+    chunker_version: Mapped[int] = mapped_column(Integer, nullable=False)
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         default=lambda: datetime.now(timezone.utc),

src/scribe/services/dedup.py

@@ -69,6 +69,12 @@ _SEMANTIC_THRESHOLD = 0.90
 # structural signals cannot see.
 _SNIPPET_SEMANTIC_THRESHOLD = 0.96
 
+# The gate queries per CHUNK of the candidate (#280) — this caps how many
+# searches one save may cost. Eight chunks ≈ five thousand words of candidate;
+# a duplicate hiding past that is the duplicate report's job to find, not a
+# reason to stall the write path.
+_GATE_MAX_CHUNKS = 8
+
 
 @dataclass
 class DuplicateMatch:
@@ -258,41 +264,45 @@ async def find_duplicate_note(
 
     # --- Signal 3: semantic similarity (only with a substantial body) ---
     if body and len(body.strip()) >= _MIN_BODY_FOR_SEMANTIC:
-        # Built by the SAME function the corpus was embedded with. This one is
+        # Query with the SAME chunker the corpus was embedded with (#280). This
-        # the copy that mattered most and was easiest to miss: it is a QUERY
+        # was the copy that mattered most and was easiest to miss: these are
-        # document, compared against embedded ones. Shaped differently from the
+        # QUERY documents, compared against embedded ones — shaped differently
-        # corpus it searches, the gate degrades silently — it still returns
+        # from the corpus, the gate degrades silently. Chunking also makes the
-        # neighbours, just less apt ones, and no signal says the query and the
+        # gate see what the whole-document query diluted: a long candidate that
-        # index stopped agreeing (found by the guard in test_embedding_text).
+        # duplicates an existing record IN ONE SECTION now matches on that
-        query = embeddings_svc.embedding_text(title, body)
+        # section. Capped so one pathological paste can't turn a save into
-        # Scope the semantic check the same way as the title check: a record in
+        # dozens of searches — a duplicate past the cap is the duplicate
-        # project P compares only to P; a project-less (orphan) record compares
+        # report's job, not the gate's.
-        # only to other orphans (orphan_only), NOT across every project — without
+        for query in embeddings_svc.chunk_document(title, body)[:_GATE_MAX_CHUNKS]:
-        # this, semantic_search_notes applies no project filter when project_id
+            # Scope the semantic check the same way as the title check: a record
-        # is None and would match an orphan note against any project's notes.
+            # in project P compares only to P; a project-less (orphan) record
-        hits = await embeddings_svc.semantic_search_notes(
+            # compares only to other orphans (orphan_only), NOT across every
-            user_id, query, project_id=project_id, is_task=is_task,
+            # project — without this, semantic_search_notes applies no project
-            orphan_only=(project_id is None),
+            # filter when project_id is None and would match an orphan note
-            limit=3,
+            # against any project's notes.
-            threshold=(_SNIPPET_SEMANTIC_THRESHOLD
+            hits = await embeddings_svc.semantic_search_notes(
-                       if note_type == SNIPPET_NOTE_TYPE else _SEMANTIC_THRESHOLD),
+                user_id, query, project_id=project_id, is_task=is_task,
-            # Owner-only, deliberately: this gate BLOCKS a create and tells the
+                orphan_only=(project_id is None),
-            # caller to update the match instead. Matching someone else's record
+                limit=3,
-            # would refuse their write and point them at something they may not
+                threshold=(_SNIPPET_SEMANTIC_THRESHOLD
-            # be able to edit.
+                           if note_type == SNIPPET_NOTE_TYPE else _SEMANTIC_THRESHOLD),
-            scope="own",
+                # Owner-only, deliberately: this gate BLOCKS a create and tells
-            # NOT demoted by supersession (#278). A superseded record is still a
+                # the caller to update the match instead. Matching someone
-            # duplicate of what you are about to write — the claim is that it is
+                # else's record would refuse their write and point them at
-            # no longer CURRENT, not that it is gone. Demoting it here would let
+                # something they may not be able to edit.
-            # the same note be recorded a second time, and the second copy would
+                scope="own",
-            # be the one nothing warns about.
+                # NOT demoted by supersession (#278). A superseded record is
-            demote_superseded=False,
+                # still a duplicate of what you are about to write — the claim
-        )
+                # is that it is no longer CURRENT, not that it is gone. Demoting
-        for score, note in hits:
+                # it here would let the same note be recorded a second time, and
-            # semantic_search_notes doesn't filter note_type — enforce it here so
+                # the second copy would be the one nothing warns about.
-            # a note doesn't shadow a task of the same wording, etc.
+                demote_superseded=False,
-            if note.note_type == note_type:
+            )
-                return DuplicateMatch(note.id, note.title, round(score, 3), "semantic")
+            for score, note in hits:
+                # semantic_search_notes doesn't filter note_type — enforce it
+                # here so a note doesn't shadow a task of the same wording, etc.
+                if note.note_type == note_type:
+                    return DuplicateMatch(note.id, note.title, round(score, 3), "semantic")
 
     return None
 
@@ -502,15 +512,22 @@ async def find_duplicate_records(
     left_note = aliased(Note, name="left_note")
     right_note = aliased(Note, name="right_note")
     distance = left.embedding.cosine_distance(right.embedding)
+    # Chunk grain (#280): a note-pair's similarity is its closest CHUNK pair —
+    # two records duplicate each other where their most similar sections do,
+    # which is the honest definition when one section of a long note restates
+    # another record. GROUP BY collapses the chunk cross-product to one row
+    # per note pair.
+    best = func.min(distance)
 
     pairs: list[tuple[int, int, float]] = []
     try:
         async with async_session() as session:
             stmt = (
-                select(left.note_id, right.note_id, distance.label("distance"))
+                select(left.note_id, right.note_id, best.label("distance"))
                 .select_from(left)
                 # `<` not `!=`: each unordered pair exactly once, and it drops
-                # the self-pair (distance 0) that would otherwise dominate.
+                # the self-pairs (including cross-chunk self-pairs, which would
+                # otherwise flag every multi-chunk note against itself).
                 .join(right, left.note_id < right.note_id)
                 .join(left_note, left_note.id == left.note_id)
                 .join(right_note, right_note.id == right.note_id)
@@ -523,9 +540,10 @@ async def find_duplicate_records(
                     # report is bounded by what merge can actually act on.
                     left_note.user_id == user_id,
                     right_note.user_id == user_id,
-                    distance <= max_distance,
                 )
-                .order_by(distance.asc())
+                .group_by(left.note_id, right.note_id)
+                .having(best <= max_distance)
+                .order_by(best.asc())
                 .limit(max(1, limit))
             )
             rows = list((await session.execute(stmt)).all())

src/scribe/services/embeddings.py

@@ -75,10 +75,19 @@ async def get_embedding(text: str) -> list[float]:
     Raises if the fastembed model fails to load. Callers should catch and
     degrade to keyword search.
     """
+    return (await get_embeddings([text]))[0]
+
+
+async def get_embeddings(texts: list[str]) -> list[list[float]]:
+    """Embed several texts in one model call (the chunked write path).
+
+    fastembed batches internally, so N chunks cost far less than N single
+    calls. Raises like get_embedding; callers catch and degrade.
+    """
     embedder = await _get_model()
     # embed() is synchronous CPU work; offload so we don't block the event loop.
-    vecs = await asyncio.to_thread(lambda: list(embedder.embed([text])))
+    vecs = await asyncio.to_thread(lambda: list(embedder.embed(texts)))
-    return vecs[0].tolist()
+    return [v.tolist() for v in vecs]
 
 
 def _cosine_similarity(a: list[float], b: list[float]) -> float:
@@ -115,6 +124,14 @@ _SUPERSESSION_PENALTY = 0.05
 # whose neighbours sit ~0.01-0.02 apart.
 _SUPERSESSION_OVERFETCH = 3
 
+# Chunk rows fetched per requested result (#280). The HNSW top-k runs at CHUNK
+# grain — several chunks of one strong note can occupy consecutive ranks, and
+# each collapses into a single result. Four ranks of headroom per result keeps
+# the top-k indexed while making it effectively impossible for collapsing to
+# starve the result list: that would need every requested note to be shadowed
+# by four chunks of notes ranked above it.
+_CHUNK_OVERFETCH = 4
+
 
 async def _apply_supersession_penalty(
     scored: list[tuple[float, "Note"]], limit: int
@@ -322,12 +339,36 @@ def chunk_document(title: str | None, body: str | None) -> list[str]:
     return chunks
 
 
-async def upsert_note_embedding(note_id: int, user_id: int, text: str) -> None:
+async def upsert_note_embedding(
-    """Generate and persist an embedding for a note. Safe to fire-and-forget."""
+    note_id: int, user_id: int, title: str | None, body: str | None
-    if not text or not text.strip():
+) -> None:
-        return
+    """Chunk, embed and persist a note's vectors. Safe to fire-and-forget.
+
+    Takes title/body rather than pre-built text so the chunking happens HERE —
+    one path for the write path, the recurrence spawn and the startup backfill,
+    which is the same single-definition discipline embedding_text existed for.
+
+    Replacement is atomic per note: old rows are deleted and the new chunk set
+    inserted in one transaction, so a concurrent read sees the old shape or the
+    new one, never a mixture.
+    """
+    chunks = chunk_document(title, body)
     try:
-        embedding = await get_embedding(text)
+        if not chunks:
+            # A record emptied of content should stop being findable by its
+            # old content — clear stale vectors rather than leaving them.
+            async with async_session() as session:
+                await session.execute(
+                    delete(NoteEmbedding).where(NoteEmbedding.note_id == note_id)
+                )
+                await session.commit()
+            return
+    except Exception:
+        logger.warning("Failed to clear embedding for note %d", note_id, exc_info=True)
+        return
+
+    try:
+        vectors = await get_embeddings(chunks)
     except Exception:
         logger.debug("Skipping embedding for note %d — embedder unavailable", note_id)
         return
@@ -337,9 +378,19 @@ async def upsert_note_embedding(note_id: int, user_id: int, text: str) -> None:
             await session.execute(
                 delete(NoteEmbedding).where(NoteEmbedding.note_id == note_id)
             )
-            session.add(NoteEmbedding(note_id=note_id, user_id=user_id, embedding=embedding))
+            for index, (chunk, vector) in enumerate(zip(chunks, vectors)):
+                session.add(
+                    NoteEmbedding(
+                        note_id=note_id,
+                        chunk_index=index,
+                        user_id=user_id,
+                        embedding=vector,
+                        chunk_text=chunk,
+                        chunker_version=CHUNKER_VERSION,
+                    )
+                )
             await session.commit()
-        logger.debug("Upserted embedding for note %d", note_id)
+        logger.debug("Upserted %d chunk embedding(s) for note %d", len(chunks), note_id)
     except Exception:
         logger.warning("Failed to persist embedding for note %d", note_id, exc_info=True)
 
@@ -481,7 +532,9 @@ async def semantic_search_notes(
             # penalty far smaller than the window's score spread, that case
             # needs the true answer to be more than _SUPERSESSION_OVERFETCH
             # ranks down, which no observed query comes close to.
-            fetch = limit * _SUPERSESSION_OVERFETCH if demote_superseded else limit
+            fetch = limit * _CHUNK_OVERFETCH * (
+                _SUPERSESSION_OVERFETCH if demote_superseded else 1
+            )
             stmt = (
                 stmt.where(distance <= max_distance)
                 .order_by(distance.asc())
@@ -492,48 +545,69 @@ async def semantic_search_notes(
         logger.warning("Failed to query note embeddings", exc_info=True)
         return []
 
-    # Recover similarity (1 - distance) and preserve the highest-first contract.
+    # Collapse chunk rows to BEST-CHUNK-PER-NOTE (#280): rows arrive ordered by
-    scored = [(1.0 - float(dist), note) for note, dist in rows]
+    # distance, so the first appearance of a note is its best chunk and later
+    # appearances are the same note matched less well. A note's relevance IS
+    # its best section's relevance — a query about one topic of a long record
+    # must find that record as strongly as if the topic were the whole record.
+    # Recover similarity (1 - distance); order stays highest-first.
+    scored: list[tuple[float, Note]] = []
+    seen: set[int] = set()
+    for note, dist in rows:
+        if int(note.id) in seen:
+            continue
+        seen.add(int(note.id))
+        scored.append((1.0 - float(dist), note))
     if not demote_superseded:
         return scored[:limit]
     return await _apply_supersession_penalty(scored, limit)
 
 
 async def backfill_note_embeddings() -> None:
-    """Generate embeddings for all notes that don't have one yet.
+    """(Re-)embed every note that is missing vectors OR whose stored vectors
+    were produced by an older chunker.
 
-    Runs as a background task at startup. Adds a small sleep between notes
+    Runs as a background task at startup. Version-awareness is what makes a
-    so a large backfill doesn't peg CPU.
+    document-shape change deployable: migration 0077 cleared the table once,
+    and every later CHUNKER_VERSION bump re-embeds the stale notes here — a
+    version comparison instead of another wipe. Adds a small sleep between
+    notes so a large backfill doesn't peg CPU.
     """
     try:
         async with async_session() as session:
-            existing = {
+            current = {
                 row[0]
                 for row in (
-                    await session.execute(select(NoteEmbedding.note_id))
+                    await session.execute(
+                        select(NoteEmbedding.note_id).where(
+                            NoteEmbedding.chunker_version == CHUNKER_VERSION
+                        )
+                    )
                 ).fetchall()
             }
             result = await session.execute(
                 select(Note.id, Note.user_id, Note.title, Note.body)
             )
             notes_to_embed = [
-                row for row in result.fetchall() if row[0] not in existing
+                row for row in result.fetchall() if row[0] not in current
             ]
     except Exception:
         logger.warning("Embedding backfill: failed to query notes", exc_info=True)
         return
 
     if not notes_to_embed:
-        logger.info("Embedding backfill: all notes already have embeddings")
+        logger.info("Embedding backfill: all notes current at chunker v%d", CHUNKER_VERSION)
         return
 
-    logger.info("Embedding backfill: generating embeddings for %d notes", len(notes_to_embed))
+    logger.info(
+        "Embedding backfill: embedding %d notes at chunker v%d",
+        len(notes_to_embed), CHUNKER_VERSION,
+    )
     success = 0
     for note_id, user_id, title, body in notes_to_embed:
-        text = embedding_text(title, body)
+        if not chunk_document(title, body):
-        if not text:
             continue
-        await upsert_note_embedding(note_id, user_id, text)
+        await upsert_note_embedding(note_id, user_id, title, body)
         success += 1
         await asyncio.sleep(0.05)  # gentle pacing
 

src/scribe/services/notes.py

@@ -33,11 +33,12 @@ def embed_note(note) -> None:
     try:
         import asyncio
 
-        from scribe.services.embeddings import embedding_text, upsert_note_embedding
+        from scribe.services.embeddings import upsert_note_embedding
-        text = embedding_text(note.title, note.body)
+        # Chunking and the empty-record gate live inside upsert_note_embedding —
-        if not text:
+        # one path for every writer (#280).
-            return
+        asyncio.create_task(
-        asyncio.create_task(upsert_note_embedding(note.id, note.user_id, text))
+            upsert_note_embedding(note.id, note.user_id, note.title, note.body)
+        )
     except RuntimeError:
         pass          # no running loop — a sync caller, not a failure
     except Exception:  # noqa: BLE001 - never let indexing break a write
@@ -235,15 +236,25 @@ async def list_notes(
             if query_vec is not None:
                 from scribe.models.embedding import NoteEmbedding
                 from scribe.services.embeddings import INTERACTIVE_SEARCH_THRESHOLD
-                distance = NoteEmbedding.embedding.cosine_distance(query_vec)
+                # Best-chunk-per-note as a correlated MIN, not a join (#280):
-                sem_filter = distance <= (1.0 - INTERACTIVE_SEARCH_THRESHOLD)
+                # a note stores one embedding row PER CHUNK, so the plain join
-                query = query.join(
+                # this used to be would repeat a long note once per matching
-                    NoteEmbedding, NoteEmbedding.note_id == Note.id
+                # chunk — duplicated list rows and a total that counts chunks.
-                ).where(sem_filter)
+                # This query is filter-heavy and paginated, never HNSW-bound,
-                count_query = count_query.join(
+                # so the scalar subquery costs what the join did.
-                    NoteEmbedding, NoteEmbedding.note_id == Note.id
+                best_distance = (
-                ).where(sem_filter)
+                    select(
-                semantic_order = distance.asc()
+                        func.min(
+                            NoteEmbedding.embedding.cosine_distance(query_vec)
+                        )
+                    )
+                    .where(NoteEmbedding.note_id == Note.id)
+                    .scalar_subquery()
+                )
+                sem_filter = best_distance <= (1.0 - INTERACTIVE_SEARCH_THRESHOLD)
+                query = query.where(sem_filter)
+                count_query = count_query.where(sem_filter)
+                semantic_order = best_distance.asc()
             else:
                 terms = _strip_type_nouns(q)
                 for term in terms:

src/scribe/services/recurrence.py

@@ -1,5 +1,4 @@
 """Recurring task rule validation, date calculation, and scheduled spawning."""
-import asyncio
 import calendar
 import logging
 from datetime import date, datetime, timedelta, timezone
@@ -103,7 +102,6 @@ async def spawn_recurring_tasks() -> int:
 
     Returns the number of tasks spawned.
     """
-    from scribe.services.embeddings import embedding_text, upsert_note_embedding
     from scribe.services.notes import create_note
 
     now = datetime.now(timezone.utc)
@@ -139,9 +137,9 @@ async def spawn_recurring_tasks() -> int:
                 milestone_id=task.milestone_id,
                 recurrence_rule=task.recurrence_rule,
             )
-            text = embedding_text(child.title, child.body)
+            # No explicit embed here: create_note calls embed_note itself, so
-            if text:
+            # the spawn path stopped being a second copy of the embedding rule
-                asyncio.create_task(upsert_note_embedding(child.id, task.user_id, text))
+            # the moment that moved into the service (#2056, #280).
         except Exception:
             logger.exception("Failed to spawn recurring task %d", task.id)
             continue

tests/test_chunking.py

@@ -134,3 +134,127 @@ def test_a_monster_single_paragraph_is_hard_split_not_dropped():
     assert len(chunks) > 1
     total_words = sum(chunk.count("word") for chunk in chunks)
     assert total_words == 2000
+
+
+# --- the read path: best chunk wins (#280 step 4) ----------------------------
+
+
+async def test_search_collapses_chunk_rows_to_best_chunk_per_note():
+    """Rows arrive at CHUNK grain ordered by distance; a note appearing via
+    several chunks must come back ONCE, scored by its best chunk — otherwise a
+    long record fills the top-k with copies of itself."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from scribe.services import embeddings as emb
+
+    note_a, note_b = MagicMock(id=1), MagicMock(id=2)
+    rows = [(note_a, 0.10), (note_b, 0.20), (note_a, 0.25), (note_a, 0.30)]
+    result = MagicMock()
+    result.all.return_value = rows
+    session, ctx = _session_ctx()
+    session.execute = AsyncMock(return_value=result)
+
+    with (
+        patch.object(emb, "async_session", return_value=ctx),
+        patch.object(emb, "get_embedding", AsyncMock(return_value=[0.0] * 384)),
+    ):
+        out = await emb.semantic_search_notes(
+            1, "a query", limit=8, demote_superseded=False
+        )
+
+    assert [note.id for _s, note in out] == [1, 2]
+    assert out[0][0] == 1.0 - 0.10  # the BEST chunk's score, not a later one
+
+
+# --- the write path: one row per chunk (#280 step 3) -------------------------
+
+
+def _session_ctx():
+    from unittest.mock import AsyncMock, MagicMock
+
+    session = MagicMock()
+    session.execute = AsyncMock()
+    session.commit = AsyncMock()
+    ctx = MagicMock()
+    ctx.__aenter__ = AsyncMock(return_value=session)
+    ctx.__aexit__ = AsyncMock(return_value=False)
+    return session, ctx
+
+
+async def test_upsert_stores_one_versioned_row_per_chunk():
+    from unittest.mock import AsyncMock, patch
+
+    from scribe.services import embeddings as emb
+
+    body = "\n\n".join(
+        f"## Section {i}\n\n{_long_section(f'sec-{i}')}" for i in range(6)
+    )
+    chunks = chunk_document("T", body)
+    assert len(chunks) > 1
+
+    session, ctx = _session_ctx()
+    with (
+        patch.object(emb, "async_session", return_value=ctx),
+        patch.object(
+            emb, "get_embeddings",
+            AsyncMock(return_value=[[0.0] * 384 for _ in chunks]),
+        ),
+    ):
+        await emb.upsert_note_embedding(7, 42, "T", body)
+
+    rows = [call.args[0] for call in session.add.call_args_list]
+    assert [r.chunk_index for r in rows] == list(range(len(chunks)))
+    assert [r.chunk_text for r in rows] == chunks
+    assert {r.chunker_version for r in rows} == {emb.CHUNKER_VERSION}
+    assert {r.user_id for r in rows} == {42}
+    session.execute.assert_awaited()  # the delete that makes replacement atomic
+
+
+async def test_upsert_of_an_emptied_record_clears_rows_instead_of_embedding():
+    """An empty embedding is worse than none, and a STALE one is worse than
+    that — a record emptied of content must stop being findable by what it no
+    longer says."""
+    from unittest.mock import patch
+
+    from scribe.services import embeddings as emb
+
+    session, ctx = _session_ctx()
+    with (
+        patch.object(emb, "async_session", return_value=ctx),
+        patch.object(emb, "get_embeddings") as embedder,
+    ):
+        await emb.upsert_note_embedding(7, 42, "", "")
+
+    embedder.assert_not_called()
+    session.execute.assert_awaited()   # the delete
+    session.add.assert_not_called()
+    session.commit.assert_awaited()
+
+
+async def test_backfill_reembeds_notes_with_a_stale_chunker_version():
+    """The reason chunker_version exists: a shape change becomes a version
+    bump that re-embeds exactly the stale notes, instead of another 0077-style
+    table wipe. Only rows AT the current version count as done."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from scribe.services import embeddings as emb
+
+    current_rows = MagicMock()
+    current_rows.fetchall.return_value = [(1,)]         # note 1 is current
+    note_rows = MagicMock()
+    note_rows.fetchall.return_value = [
+        (1, 42, "current", "body"),
+        (2, 42, "stale-version", "body"),
+    ]
+    session, ctx = _session_ctx()
+    session.execute = AsyncMock(side_effect=[current_rows, note_rows])
+
+    with (
+        patch.object(emb, "async_session", return_value=ctx),
+        patch.object(emb, "upsert_note_embedding", AsyncMock()) as upsert,
+        patch.object(emb.asyncio, "sleep", AsyncMock()),
+    ):
+        await emb.backfill_note_embeddings()
+
+    embedded = [call.args[0] for call in upsert.call_args_list]
+    assert embedded == [2], "only the stale note is re-embedded"

tests/test_integration_pgvector_search.py

@@ -31,6 +31,20 @@ def _vec(*nonzero_first):
     return v[:EMBEDDING_DIM]
 
 
+def _emb(note_id, user_id, chunk_index, vec):
+    """A chunk row at the current chunker version (#280, migration 0077)."""
+    from scribe.services.embeddings import CHUNKER_VERSION
+
+    return NoteEmbedding(
+        note_id=note_id,
+        chunk_index=chunk_index,
+        user_id=user_id,
+        embedding=vec,
+        chunk_text=f"chunk {chunk_index} of note {note_id}",
+        chunker_version=CHUNKER_VERSION,
+    )
+
+
 @pytest_asyncio.fixture(autouse=True)
 async def _dispose_engine():
     # Per-loop pool: dispose after each test (see test_integration_db_maintenance).
@@ -54,8 +68,12 @@ async def seeded():
         await s.flush()
         # query vector will be [1,0,0,...]; near ~ identical (sim≈1.0),
         # far is orthogonal (sim≈0.0 -> filtered by the default threshold).
-        s.add(NoteEmbedding(note_id=near.id, user_id=user.id, embedding=_vec(1.0)))
+        # near gets a SECOND, weaker chunk (sim≈0.6) — the collapse to
-        s.add(NoteEmbedding(note_id=far.id, user_id=user.id, embedding=_vec(0.0, 1.0)))
+        # best-chunk-per-note (#280) is under test: near must come back once,
+        # at its best chunk's score, not twice.
+        s.add(_emb(near.id, user.id, 0, _vec(1.0)))
+        s.add(_emb(near.id, user.id, 1, _vec(0.6, 0.8)))
+        s.add(_emb(far.id, user.id, 0, _vec(0.0, 1.0)))
         await s.commit()
         ids = (user.id, near.id, far.id)
     yield ids
@@ -82,6 +100,9 @@ async def test_semantic_search_ranks_and_thresholds_via_pgvector(seeded):
     assert near_id in ids
     assert far_id not in ids
     assert ids[0] == near_id
+    # Chunk collapse (#280): near has TWO chunk rows above the floor (sim≈1.0
+    # and ≈0.6) and must appear exactly once, at its best chunk's score.
+    assert ids.count(near_id) == 1
     top_score = results[0][0]
     assert top_score == pytest.approx(1.0, abs=1e-3)
 

tests/test_recurrence.py

@@ -254,7 +254,6 @@ async def test_spawn_recurring_tasks_creates_child():
             new_callable=AsyncMock,
             return_value=mock_child,
         ) as mock_create,
-        patch("scribe.services.embeddings.upsert_note_embedding"),
     ):
         from scribe.services.recurrence import spawn_recurring_tasks
         count = await spawn_recurring_tasks()

tests/test_services_dedup.py

@@ -68,6 +68,33 @@ async def test_semantic_match_when_body_substantial():
     assert dup.similarity == 0.93
 
 
+@pytest.mark.asyncio
+async def test_gate_catches_a_duplicate_hiding_in_a_later_chunk():
+    """The capability #280 adds to the gate: a long candidate that duplicates
+    an existing record in ONE SECTION is caught, where the whole-document
+    query this replaces diluted exactly the section that mattered. The gate
+    queries once per chunk and any chunk's hit blocks."""
+    para = ("This section restates an existing decision in enough words to be "
+            "a real paragraph of content for the chunker to keep. ") * 4
+    body = "\n\n".join(f"## Topic {i}\n\n{para} (t{i})" for i in range(8))
+
+    from scribe.services.embeddings import chunk_document
+    n_chunks = len(chunk_document("Title", body))
+    assert n_chunks > 1, "test body must actually chunk"
+
+    hit = _fake_note(id=30, title="The existing decision", note_type="note")
+    # Every chunk misses except the LAST one the gate will ask about.
+    sem = AsyncMock(side_effect=[[] for _ in range(n_chunks - 1)] + [[(0.94, hit)]])
+    with patch("scribe.services.dedup.async_session",
+               return_value=_session_returning(None)), \
+         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
+        dup = await find_duplicate_note(
+            7, "Title", body=body, project_id=2, is_task=False, note_type="note",
+        )
+    assert dup is not None and dup.id == 30
+    assert sem.await_count == n_chunks
+
+
 @pytest.mark.asyncio
 async def test_semantic_match_of_other_note_type_is_ignored():
     other = _fake_note(id=21, title="X", note_type="process")

tests/test_services_notes.py

@@ -29,18 +29,25 @@ def test_embed_note_uses_the_OWNER_not_the_caller():
     upsert.assert_called_once()
     assert upsert.call_args.args[0] == 5
     assert upsert.call_args.args[1] == 42          # owner, never the caller
-    assert upsert.call_args.args[2] == "T\nB"
+    # Title and body travel separately since #280 — chunking happens inside
+    # upsert_note_embedding, the one path every writer shares.
+    assert upsert.call_args.args[2] == "T"
+    assert upsert.call_args.args[3] == "B"
 
 
-def test_embed_note_skips_a_record_with_no_text():
+def test_embed_note_hands_even_an_empty_record_to_the_one_path():
-    """An empty embedding is worse than none — it is a row that matches nothing
+    """The empty-record decision moved INTO upsert_note_embedding (#280): an
-    and hides the fact that the record was never indexed."""
+    emptied record must have its stale vectors CLEARED, not merely skipped —
+    so embed_note schedules the call unconditionally rather than deciding
+    here. The clearing behaviour itself is pinned in test_chunking.py."""
     from unittest.mock import MagicMock, patch
 
     note = MagicMock(id=5, user_id=42, title="", body="")
-    with patch("asyncio.create_task") as create_task:
+    with patch("scribe.services.embeddings.upsert_note_embedding") as upsert, \
+         patch("asyncio.create_task") as create_task:
         notes_svc.embed_note(note)
-    assert not create_task.called
+    assert create_task.called
+    upsert.assert_called_once()
 
 
 def test_embed_note_without_a_running_loop_is_not_an_error():