feat(rules): rules become findable by meaning (#3030, milestone 307 step 4)

Rules were the only major record type with no vector, so `search` could never
return one and a rule could arrive only by being preloaded. That single fact is
what made every rule compete for one always-on budget.

THE DECISION THE TASK ASKED FOR, made explicitly: a sibling rule_embeddings
table, not a polymorphic embedding row. The ROW could have been generalised;
the SEARCH could not. semantic_search_notes is Note-specific scoping end to end
— the visibility clause, the supersession penalty, note_type/task_kind/system
filters — and a rule shares none of it, scoping instead by rulebook ownership
or project. Generalising the row while still needing two searches is the worst
of both: a key with referential integrity to neither table, on the path every
session start runs, to share four columns. What is genuinely common is
BEHAVIOUR — get_embedding, chunk_document, embedding_text, CHUNKER_VERSION —
and those are reused as-is. Sharing them is the DRY win; sharing the table
would have been the DRY costume.

The document shape is measured, not chosen (note 2485). That pass found the
snippet was the only discriminative record in the corpus — a 0.153
top-to-second gap against 0.010-0.023 — and that the cause was its SHAPE:
purpose stated twice in a short single-topic document. rule_document
reproduces it: the trigger in the title AND as the body's first line.

And it excludes `why`, which matters more than any of it. `why` is dated
incident narrative — rule 46's runs to 4,300 characters — and long multi-topic
prose is exactly what made sixteen dev-logs mutually indistinguishable. Adding
it would not give the vector more to work with; it would give every rule the
SAME thing to work with. rule_document takes no `why` parameter at all, so a
well-meaning caller cannot pass one.

A rule with no trigger degrades to title + statement — findable, less sharp.
That is an argument for backfilling triggers (step 6), not for padding the
document with whatever text is nearby.

search(content_type="rule") returns the rule WITH its why and how_to_apply:
they are its operational half, the session payload never carries them, and a
caller who went looking should not have to re-fetch. Writes re-index
fire-and-forget like notes; startup backfills in its own try block so neither
backfill can skip the other. rule_embeddings is derived, so it joins
note_embeddings in the backup's explicitly-NOT-included list.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/scribe/services/backup.py

@@ -81,7 +81,8 @@ _BACKED_UP = [
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
 # explicit rather than silent. ACL (groups/shares) is a coherent follow-up;
-# note_embeddings are derived (regenerated from note bodies); api_keys are
+# note_embeddings and rule_embeddings are derived (regenerated at startup
+# from the records themselves, which is also how a chunker bump is handled); api_keys are
 # sensitive credentials; retrieval_logs is observational telemetry that nothing
 # reads for correctness and that grows per query; the rest are
 # transient/operational.
@@ -91,7 +92,7 @@ _BACKED_UP = [
 # like coverage while naming nothing the schema could confirm.
 _NOT_INCLUDED = [
     "groups", "group_memberships", "project_shares", "note_shares",
-    "api_keys", "note_embeddings", "app_logs", "notifications",
+    "api_keys", "note_embeddings", "rule_embeddings", "app_logs", "notifications",
     "invitation_tokens", "password_reset_tokens", "user_profiles",
     "retrieval_logs",
     # Sensitive credentials, same reasoning as api_keys: a backup that carries

src/scribe/services/embeddings.py

@@ -19,7 +19,7 @@ from collections.abc import Sequence
 from sqlalchemy import delete, or_, select
 
 from scribe.models import async_session
-from scribe.models.embedding import NoteEmbedding
+from scribe.models.embedding import NoteEmbedding, RuleEmbedding
 from scribe.models.note import Note
 from scribe.services.access import notes_visibility_clause
 
@@ -612,3 +612,186 @@ async def backfill_note_embeddings() -> None:
         await asyncio.sleep(0.05)  # gentle pacing
 
     logger.info("Embedding backfill complete: %d/%d notes embedded", success, len(notes_to_embed))
+
+
+# ── Rules (milestone 307, note 3026) ────────────────────────────────────
+
+def rule_document(
+    title: str | None, statement: str | None, when_to_apply: str | None,
+) -> tuple[str | None, str | None]:
+    """The (title, body) a rule is EMBEDDED as — trigger first, `why` never.
+
+    Both halves of this are measured, not guessed (note 2485). That pass found
+    the snippet was the only sharp record in the corpus — a 0.153 top-to-second
+    gap against 0.010–0.023 for everything else — and that the cause was its
+    SHAPE: `{name} — {when_to_use}` as the title and `**When to use:** …`
+    repeated in the body, so purpose appears twice in a short document and
+    dominates the vector. This mirrors that exactly.
+
+    And it excludes `why` on the same evidence. `why` is dated incident
+    narrative — rule 46's runs to 4,300 characters of it — and long,
+    multi-topic prose is precisely what made sixteen dev-logs mutually
+    indistinguishable: the average lands on the centroid of "development",
+    which every one of them shares. Adding `why` would not give the vector more
+    to work with; it would give every rule the same thing to work with.
+
+    A rule with no trigger yet degrades to title + statement. It still embeds,
+    just less sharply — which is an argument for backfilling triggers, not an
+    argument for padding the document with whatever text is lying around.
+    """
+    trigger = (when_to_apply or "").strip()
+    name = (title or "").strip()
+    body = (statement or "").strip()
+    if not trigger:
+        return name or None, body or None
+    return (
+        f"{name} — {trigger}" if name else trigger,
+        f"When to apply: {trigger}\n\n{body}" if body else f"When to apply: {trigger}",
+    )
+
+
+async def upsert_rule_embedding(
+    rule_id: int, title: str | None, statement: str | None,
+    when_to_apply: str | None = None,
+) -> None:
+    """Chunk, embed and persist a rule's vectors. Safe to fire-and-forget.
+
+    The note twin's contract, for the same reasons: the document is built HERE
+    so the write path, the backfill and any re-embed share one definition, and
+    replacement is atomic per rule so a concurrent read sees the old chunk set
+    or the new one, never a mixture.
+    """
+    doc_title, doc_body = rule_document(title, statement, when_to_apply)
+    chunks = chunk_document(doc_title, doc_body)
+    try:
+        if not chunks:
+            async with async_session() as session:
+                await session.execute(
+                    delete(RuleEmbedding).where(RuleEmbedding.rule_id == rule_id)
+                )
+                await session.commit()
+            return
+    except Exception:
+        logger.warning("Failed to clear embedding for rule %d", rule_id, exc_info=True)
+        return
+
+    try:
+        vectors = await get_embeddings(chunks)
+    except Exception:
+        logger.debug("Skipping embedding for rule %d — embedder unavailable", rule_id)
+        return
+
+    try:
+        async with async_session() as session:
+            await session.execute(
+                delete(RuleEmbedding).where(RuleEmbedding.rule_id == rule_id)
+            )
+            for index, (chunk, vector) in enumerate(zip(chunks, vectors)):
+                session.add(
+                    RuleEmbedding(
+                        rule_id=rule_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 rule %d", rule_id, exc_info=True)
+
+
+async def semantic_search_rules(
+    user_id: int,
+    query: str,
+    limit: int = 5,
+    threshold: float = _SIMILARITY_THRESHOLD,
+) -> list[tuple[float, "Rule"]]:
+    """Return up to *limit* (score, rule) pairs most relevant to *query*.
+
+    Scoped by OWNERSHIP — a rule is the caller's if they own its rulebook or
+    its project. Deliberately not filtered to what currently BINDS a given
+    project: this answers "is there a rule about this", which a person asking
+    wants answered across their whole rulebook. Deciding which rules bind where
+    is the surfacing question, and it has its own machinery
+    (get_applicable_rules) rather than a second, subtly different copy here.
+
+    Collapses to best-chunk-per-rule like the note search, so a long rule split
+    across chunks competes once rather than crowding the results with itself.
+
+    Returns an empty list if the embedder is unavailable or on any error.
+    """
+    from scribe.models.project import Project
+    from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
+
+    if not query or not query.strip():
+        return []
+    try:
+        query_vec = await get_embedding(query)
+    except Exception:
+        logger.debug("Rule search skipped — embedder unavailable")
+        return []
+
+    max_distance = min(2.0, max(0.0, 1.0 - threshold))
+    distance = RuleEmbedding.embedding.cosine_distance(query_vec)
+
+    try:
+        async with async_session() as session:
+            rows = (await session.execute(
+                select(Rule, distance.label("distance"))
+                .select_from(RuleEmbedding)
+                .join(Rule, RuleEmbedding.rule_id == Rule.id)
+                .outerjoin(RulebookTopic, Rule.topic_id == RulebookTopic.id)
+                .outerjoin(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+                .outerjoin(Project, Rule.project_id == Project.id)
+                .where(
+                    Rule.deleted_at.is_(None),
+                    distance <= max_distance,
+                    # topic_id XOR project_id, so exactly one arm can match.
+                    or_(
+                        Rulebook.owner_user_id == user_id,
+                        Project.user_id == user_id,
+                    ),
+                )
+                # Overfetch so collapsing chunks to their best row still fills
+                # the page — the same reason the note search overfetches.
+                .order_by(distance)
+                .limit(limit * _CHUNK_OVERFETCH)
+            )).all()
+    except Exception:
+        logger.warning("Rule semantic search failed", exc_info=True)
+        return []
+
+    best: dict[int, tuple[float, object]] = {}
+    for rule, dist in rows:
+        score = 1.0 - float(dist)
+        if rule.id not in best or score > best[rule.id][0]:
+            best[rule.id] = (score, rule)
+    ranked = sorted(best.values(), key=lambda pair: pair[0], reverse=True)
+    return ranked[:limit]
+
+
+async def backfill_rule_embeddings() -> None:
+    """Embed rules that have no current vectors. Runs at startup beside the
+    note backfill; a CHUNKER_VERSION bump re-embeds rather than wiping."""
+    from scribe.models.rulebook import Rule
+
+    try:
+        async with async_session() as session:
+            current = select(RuleEmbedding.rule_id).where(
+                RuleEmbedding.chunker_version == CHUNKER_VERSION
+            )
+            stale = (await session.execute(
+                select(Rule.id, Rule.title, Rule.statement, Rule.when_to_apply)
+                .where(Rule.deleted_at.is_(None), Rule.id.notin_(current))
+            )).all()
+    except Exception:
+        logger.warning("Rule embedding backfill: failed to query rules", exc_info=True)
+        return
+
+    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))
+    for rule_id, title, statement, when_to_apply in stale:
+        await upsert_rule_embedding(rule_id, title, statement, when_to_apply)

src/scribe/services/rulebooks.py

@@ -334,6 +334,30 @@ def rule_brief(rule: Rule, **extra) -> dict:
     return out
 
 
+def _refresh_rule_embedding(rule: Rule) -> None:
+    """Re-index a rule after a write. Fire-and-forget, like the note twin.
+
+    Lazy import so this module doesn't pull in the embedder; every exception
+    swallowed because a rule that SAVED must not fail on its index refresh —
+    a stale vector costs a missed search hit, a raised exception costs the
+    write. No running loop (unit tests, scripts) is ordinary, not an error.
+    """
+    try:
+        import asyncio
+
+        from scribe.services.embeddings import upsert_rule_embedding
+
+        asyncio.create_task(
+            upsert_rule_embedding(
+                rule.id, rule.title, rule.statement, rule.when_to_apply,
+            )
+        )
+    except RuntimeError:
+        pass          # no running loop — a sync caller, not a failure
+    except Exception:  # noqa: BLE001 - never let indexing break a write
+        logger.exception("embedding refresh failed for rule %s", rule.id)
+
+
 async def rule_detail(user_id: int, rule: Rule, system_ids: list[int] | None = None) -> dict:
     """The full record, with its areas and edges attached.
 
@@ -380,6 +404,7 @@ async def create_rule(
         session.add(rule)
         await session.commit()
         await session.refresh(rule)
+        _refresh_rule_embedding(rule)
         return rule
 
 
@@ -410,6 +435,7 @@ async def create_project_rule(
         session.add(rule)
         await session.commit()
         await session.refresh(rule)
+        _refresh_rule_embedding(rule)
         return rule
 
 
@@ -609,6 +635,7 @@ async def update_rule(rule_id: int, user_id: int, **fields) -> Optional[Rule]:
                 setattr(rule, key, _valid_tier(value) if key == "tier" else value)
         await session.commit()
         await session.refresh(rule)
+        _refresh_rule_embedding(rule)
         return rule