refactor(embeddings): one definition of the document a record is embedded as

`f"{title}\n{body}"` was written out four times. #2486 found three — the write
path, the recurring-task spawn, the startup backfill. The guard added here
found the fourth immediately, and it was the one that mattered most.

`dedup.find_duplicate_note` built the same string as a QUERY, compared against
embedded documents. Shaped differently from the corpus it searches, the gate
degrades silently: it still returns neighbours, just less apt ones, and nothing
says the query and the index stopped agreeing. The spawn path has the same
shape of risk — a recurring task embedded differently from everything else is
ranked against documents it doesn't match.

None of the four had diverged. That is what makes this worth doing now rather
than after: they are identical today, so collapsing them is a no-op, and the
whole point is that the next change to the shape can't hit three of four.

Which is imminent. #2486 measured a dev-log separating from five unrelated
dev-logs by 0.023 where a snippet separates by 0.153 — the difference being
that a snippet states its purpose twice in a short document. Whether that shape
is right is the open question; testing an alternative against four copies would
mean testing a shape that isn't the one in production. This is the precondition
the issue named.

The guard is source inspection, matching the f-string pattern rather than a
variable name, so a copy that renames its locals is still caught. A behavioural
test cannot see this: an inlined copy produces the same string today and
diverges the day the shape changes.

Refs #2486

src/scribe/services/dedup.py

@@ -258,7 +258,13 @@ async def find_duplicate_note(
 
     # --- Signal 3: semantic similarity (only with a substantial body) ---
     if body and len(body.strip()) >= _MIN_BODY_FOR_SEMANTIC:
-        query = f"{title}\n{body}".strip()
+        # Built by the SAME function the corpus was embedded with. This one is
+        # the copy that mattered most and was easiest to miss: it is a QUERY
+        # document, compared against embedded ones. Shaped differently from the
+        # corpus it searches, the gate degrades silently — it still returns
+        # neighbours, just less apt ones, and no signal says the query and the
+        # index stopped agreeing (found by the guard in test_embedding_text).
+        query = embeddings_svc.embedding_text(title, body)
         # Scope the semantic check the same way as the title check: a record in
         # project P compares only to P; a project-less (orphan) record compares
         # only to other orphans (orphan_only), NOT across every project — without

src/scribe/services/embeddings.py

@@ -86,6 +86,31 @@ def _cosine_similarity(a: list[float], b: list[float]) -> float:
     return dot / (mag_a * mag_b)
 
 
+def embedding_text(title: str | None, body: str | None) -> str:
+    """The document a record is embedded AS.
+
+    One definition, deliberately. This was written out three times — the write
+    path (`notes.embed_note`), the recurring-task spawn, and the startup
+    backfill — and identical copies of a formatting rule are three chances to
+    change one and not the others. The spawn path is the dangerous one: a
+    recurring task embedded to a different shape than everything else would be
+    ranked against a corpus it doesn't match, and nothing would report it.
+
+    It is also a PRECONDITION for changing the shape at all (#2486). Measured,
+    a dev-log's vector separates from five unrelated dev-logs by 0.023 while a
+    snippet's separates by 0.153 — the difference being that a snippet states
+    its purpose twice in a short document, so the purpose dominates. Testing an
+    alternative shape against three copies would mean testing a shape that is
+    not the one in production.
+
+    Whether `title\\n{body}` is the RIGHT shape is the open question. That it is
+    one shape is what makes the question answerable.
+    """
+    title = title or ""
+    body = body or ""
+    return f"{title}\n{body}".strip() if body else title
+
+
 async def upsert_note_embedding(note_id: int, user_id: int, text: str) -> None:
     """Generate and persist an embedding for a note. Safe to fire-and-forget."""
     if not text or not text.strip():
@@ -248,7 +273,7 @@ async def backfill_note_embeddings() -> None:
     logger.info("Embedding backfill: generating embeddings for %d notes", len(notes_to_embed))
     success = 0
     for note_id, user_id, title, body in notes_to_embed:
-        text = f"{title}\n{body}".strip() if body else (title or "")
+        text = embedding_text(title, body)
         if not text:
             continue
         await upsert_note_embedding(note_id, user_id, text)

src/scribe/services/notes.py

@@ -30,13 +30,13 @@ def embed_note(note) -> None:
     index refresh. No running loop (unit tests, scripts) is an ordinary case,
     not an error.
     """
-    text = f"{note.title}\n{note.body}".strip() if note.body else (note.title or "")
-    if not text:
-        return
     try:
         import asyncio
 
-        from scribe.services.embeddings import upsert_note_embedding
+        from scribe.services.embeddings import embedding_text, upsert_note_embedding
+        text = embedding_text(note.title, note.body)
+        if not text:
+            return
         asyncio.create_task(upsert_note_embedding(note.id, note.user_id, text))
     except RuntimeError:
         pass          # no running loop — a sync caller, not a failure

src/scribe/services/recurrence.py

@@ -103,7 +103,7 @@ async def spawn_recurring_tasks() -> int:
 
     Returns the number of tasks spawned.
     """
-    from scribe.services.embeddings import upsert_note_embedding
+    from scribe.services.embeddings import embedding_text, upsert_note_embedding
     from scribe.services.notes import create_note
 
     now = datetime.now(timezone.utc)
@@ -139,7 +139,7 @@ async def spawn_recurring_tasks() -> int:
                 milestone_id=task.milestone_id,
                 recurrence_rule=task.recurrence_rule,
             )
-            text = f"{child.title}\n{child.body}".strip() if child.body else (child.title or "")
+            text = embedding_text(child.title, child.body)
             if text:
                 asyncio.create_task(upsert_note_embedding(child.id, task.user_id, text))
         except Exception:

tests/test_embedding_text.py

@@ -0,0 +1,81 @@
+"""The document shape a record is embedded as, and the guard that keeps it one.
+
+WHY THIS EXISTS
+
+`f"{title}\\n{body}"` was written out three times — the write path, the
+recurring-task spawn, and the startup backfill. Identical copies of a formatting
+rule are three chances to change one and not the others, and the spawn path is
+the dangerous one: a recurring task embedded to a different shape than the rest
+of the corpus is ranked against documents it doesn't match, and nothing reports
+it. A wrong vector returns results; it just returns the wrong ones.
+
+It is also the precondition for #2486. A dev-log's vector separates from five
+unrelated dev-logs by 0.023 where a snippet separates by 0.153, and the leading
+explanation is shape — a snippet states its purpose twice in a short document.
+Testing an alternative against three copies would mean testing a shape that
+isn't the one in production.
+"""
+from __future__ import annotations
+
+import ast
+import pathlib
+
+from scribe.services.embeddings import embedding_text
+
+SERVICES = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
+
+
+def test_title_and_body_are_joined_by_a_newline():
+    assert embedding_text("A title", "A body") == "A title\nA body"
+
+
+def test_a_bodyless_record_embeds_as_its_title_alone():
+    """Not "title\\n" — the trailing separator would be a token's worth of noise
+    on the shortest documents, which are the ones least able to spare it."""
+    assert embedding_text("Just a title", "") == "Just a title"
+    assert embedding_text("Just a title", None) == "Just a title"
+
+
+def test_an_empty_record_yields_an_empty_string():
+    """Callers gate on falsiness to skip embedding entirely, so this must be
+    empty rather than a stray newline."""
+    assert embedding_text("", "") == ""
+    assert embedding_text(None, None) == ""
+
+
+def test_surrounding_whitespace_is_stripped():
+    assert embedding_text("  A title  ", "  A body  ") == "A title  \n  A body"
+
+
+def test_nothing_else_builds_the_embedding_document_itself():
+    """The guard. A fourth copy is how the first three happened.
+
+    Source inspection, because this is the shape no behavioural test catches:
+    an inlined copy produces the same string today and diverges silently the day
+    the shape changes. Matches the f-string pattern itself rather than a
+    variable name, so a copy that renames its locals is still caught.
+    """
+    offenders = []
+    for path in SERVICES.rglob("*.py"):
+        source = path.read_text()
+        for node in ast.walk(ast.parse(source)):
+            if not isinstance(node, ast.JoinedStr):
+                continue
+            # An f-string whose literal parts are exactly a newline, with a
+            # substitution either side: `f"{x}\n{y}"`.
+            literals = [
+                v.value for v in node.values
+                if isinstance(v, ast.Constant) and isinstance(v.value, str)
+            ]
+            subs = [v for v in node.values if isinstance(v, ast.FormattedValue)]
+            if literals == ["\n"] and len(subs) == 2:
+                offenders.append(f"{path.relative_to(SERVICES)}:{node.lineno}")
+
+    # embeddings.py holds the one definition.
+    offenders = [o for o in offenders if not o.startswith("services/embeddings.py")]
+    assert not offenders, (
+        f"these build the embedding document inline instead of calling "
+        f"embedding_text(): {offenders}. One definition — an inlined copy is "
+        f"ranked against a corpus it no longer matches the moment the shape "
+        f"changes, and nothing reports it (#2486)."
+    )