feat(embeddings): per-chunk rows — schema, write path, version-aware backfill (#280 steps 2+3)

note_embeddings becomes one row per chunk: PK (note_id, chunk_index), plus
chunk_text (what this vector actually encodes) and chunker_version. Migration
0077 clears the table — embeddings are derived (0067 precedent) and the old
whole-document rows are indistinguishable from single-chunk notes, so the
startup backfill regenerates the corpus at the new shape. The backfill is now
version-aware: a future shape change is a CHUNKER_VERSION bump that re-embeds
exactly the stale notes, not another wipe.

upsert_note_embedding takes (title, body) and chunks internally — one path
for the write path, the recurrence spawn and the backfill. The recurrence
spawn's own embed call is deleted outright: create_note already embeds via
embed_note (#2056), so the spawn was a second copy of the rule. An emptied
record now CLEARS its stale vectors instead of leaving them findable.

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

tests/test_chunking.py

@@ -134,3 +134,97 @@ 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 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"