feat(search): milestones are searchable by meaning — "is there already a plan for this?" (#4078)

`search` covered notes, tasks and rules, and a milestone — the record a plan
lives in — could not be found. A project whose roadmap was written as
milestones had every later plan opened beside the one that already described
it, because nothing could have told the session it existed.

- milestone_embeddings (migration 0102): the third sibling of note_ and
  rule_embeddings, for note 3163's reason — the search is milestone-specific.
  The document is title — description, then description and the plan body,
  so a roadmap milestone with no description is still found by its design.
- Written on create, on a title/description/body update, and for a plan made
  through start_planning / create_records, fire-and-forget with the parent-row
  claim (#3262); a startup backfill covers every existing milestone. Derived,
  so it joins _NOT_INCLUDED beside the other embeddings.
- semantic_search_milestones: a project's milestones when the caller can read
  it (access.can_read_project), otherwise the caller's own; optional status.
- search(content_type="milestone"): id, title, description, status, project
  and progress. Its own shape, and not part of "all", whose results are
  note-shaped. The docstring says what it is for: ask before start_planning.
- Integration test on real Postgres: found in its project and not another,
  status narrows, an unreadable project returns nothing.

Milestone 415 step 3.

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

src/scribe/services/embeddings.py

@@ -25,7 +25,8 @@ from scribe.models.embedding import NoteEmbedding, RuleEmbedding
 from scribe.models.note import Note
 from scribe.services.access import can_read_project, notes_visibility_clause
 
-if TYPE_CHECKING:  # resolves the Rule forward ref without importing at runtime
+if TYPE_CHECKING:  # resolves forward refs without importing at runtime
+    from scribe.models.milestone import Milestone
     from scribe.models.rulebook import Rule
 
 logger = logging.getLogger(__name__)
@@ -975,3 +976,169 @@ async def backfill_rule_embeddings() -> None:
     logger.info("Rule embedding backfill: embedding %d rule(s)", len(stale))
     for rule_id, title, statement, when_to_apply in stale:
         await upsert_rule_embedding(rule_id, title, statement, when_to_apply)
+
+
+# ── Milestones (milestone 415) ──────────────────────────────────────────
+
+def milestone_document(
+    title: str | None, description: str | None, body: str | None,
+) -> tuple[str | None, str | None]:
+    """The (title, body) a milestone is EMBEDDED as.
+
+    Title and one-line description lead, the way a snippet's name and purpose
+    lead its document (note 2485): the question this search answers is "does
+    a plan for this already exist?", and a plan is recognised by what it is
+    FOR. The plan body follows, chunked, so a milestone whose description is
+    empty — most roadmap milestones written by hand — is still findable by the
+    words of its design.
+    """
+    name = (title or "").strip()
+    purpose = (description or "").strip()
+    plan = (body or "").strip()
+    doc_title = f"{name} — {purpose}" if name and purpose else (name or purpose or None)
+    parts = [p for p in (purpose, plan) if p]
+    return doc_title, "\n\n".join(parts) or None
+
+
+async def upsert_milestone_embedding(
+    milestone_id: int, title: str | None, description: str | None, body: str | None,
+) -> None:
+    """Chunk, embed and persist a milestone's vectors. Safe to fire-and-forget.
+
+    The rule twin's contract: one document definition shared by the write path
+    and the backfill, and an atomic per-milestone replacement guarded by the
+    parent-row claim (#3262), so a milestone deleted mid-refresh wins.
+    """
+    from scribe.models.embedding import MilestoneEmbedding
+    from scribe.models.milestone import Milestone
+
+    doc_title, doc_body = milestone_document(title, description, body)
+    chunks = chunk_document(doc_title, doc_body)
+    try:
+        if not chunks:
+            async with async_session() as session:
+                await session.execute(
+                    delete(MilestoneEmbedding).where(MilestoneEmbedding.milestone_id == milestone_id)
+                )
+                await session.commit()
+            return
+    except Exception:
+        logger.warning("Failed to clear embedding for milestone %d", milestone_id, exc_info=True)
+        return
+
+    try:
+        vectors = await get_embeddings(chunks)
+    except Exception:
+        logger.debug("Skipping embedding for milestone %d — embedder unavailable", milestone_id)
+        return
+
+    try:
+        async with async_session() as session:
+            if not await _claim_parent_row(session, Milestone.id, milestone_id, "milestone"):
+                return
+            await session.execute(
+                delete(MilestoneEmbedding).where(MilestoneEmbedding.milestone_id == milestone_id)
+            )
+            for index, (chunk, vector) in enumerate(zip(chunks, vectors)):
+                session.add(MilestoneEmbedding(
+                    milestone_id=milestone_id, chunk_index=index, embedding=vector,
+                    chunk_text=chunk, chunker_version=CHUNKER_VERSION,
+                ))
+            await session.commit()
+    except Exception:
+        logger.warning("Failed to persist embedding for milestone %d", milestone_id, exc_info=True)
+
+
+async def semantic_search_milestones(
+    user_id: int,
+    query: str,
+    *,
+    project_id: int | None = None,
+    status: str | None = None,
+    limit: int = 5,
+    threshold: float = _SIMILARITY_THRESHOLD,
+) -> list[tuple[float, "Milestone"]]:
+    """Return up to *limit* (score, milestone) pairs most like *query*.
+
+    Answers "is there already a plan for this?" — the question a session asks
+    before start_planning, and the one the planning gate asks for it.
+
+    SCOPE. With `project_id`, that project's milestones, provided the caller
+    can read the project (access.can_read_project, rule 78) — a collaborator on
+    a shared project sees its plans. Without one, the milestones the caller
+    owns across their projects. `status` narrows to "active" or "done".
+
+    Collapses to best-chunk-per-milestone, like the sibling searches. Returns
+    an empty list if the embedder is unavailable, the project is not readable,
+    or on any error: a recall aid must never break the call it serves.
+    """
+    from scribe.models.embedding import MilestoneEmbedding
+    from scribe.models.milestone import Milestone
+
+    if not query or not query.strip():
+        return []
+    try:
+        query_vec = await get_embedding(query)
+    except Exception:
+        logger.debug("Milestone search skipped — embedder unavailable")
+        return []
+
+    distance = MilestoneEmbedding.embedding.cosine_distance(query_vec)
+    try:
+        if project_id:
+            if not await can_read_project(user_id, project_id):
+                return []
+            scope = Milestone.project_id == project_id
+        else:
+            scope = Milestone.user_id == user_id
+        async with async_session() as session:
+            rows = (await session.execute(
+                select(Milestone, distance.label("distance"))
+                .select_from(MilestoneEmbedding)
+                .join(Milestone, MilestoneEmbedding.milestone_id == Milestone.id)
+                .where(
+                    scope,
+                    Milestone.deleted_at.is_(None),
+                    *([Milestone.status == status] if status else []),
+                )
+                .order_by(distance)
+                .limit(limit * _CHUNK_OVERFETCH)
+            )).all()
+    except Exception:
+        logger.warning("Milestone semantic search failed", exc_info=True)
+        return []
+
+    best: dict[int, tuple[float, object]] = {}
+    for milestone, dist in rows:
+        score = 1.0 - float(dist)
+        if milestone.id not in best or score > best[milestone.id][0]:
+            best[milestone.id] = (score, milestone)
+    ranked = sorted(best.values(), key=lambda pair: pair[0], reverse=True)
+    return [pair for pair in ranked if pair[0] >= threshold][:limit]
+
+
+async def backfill_milestone_embeddings() -> None:
+    """Embed milestones that have no current vectors. Runs at startup beside
+    the note and rule backfills; a CHUNKER_VERSION bump re-embeds."""
+    from scribe.models.embedding import MilestoneEmbedding
+    from scribe.models.milestone import Milestone
+
+    try:
+        async with async_session() as session:
+            current = select(MilestoneEmbedding.milestone_id).where(
+                MilestoneEmbedding.chunker_version == CHUNKER_VERSION
+            )
+            stale = (await session.execute(
+                select(Milestone.id, Milestone.title, Milestone.description, Milestone.body)
+                .where(Milestone.deleted_at.is_(None), Milestone.id.notin_(current))
+            )).all()
+    except Exception:
+        logger.warning("Milestone embedding backfill: failed to query milestones", exc_info=True)
+        return
+
+    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))
+    for milestone_id, title, description, body in stale:
+        await upsert_milestone_embedding(milestone_id, title, description, body)