bvandeusenandClaude Opus 5.5
66e21a6c60
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 14s
CI & Build / integration (push) Successful in 52s
CI & Build / TypeScript typecheck (push) Successful in 53s
CI & Build / Python tests (push) Successful in 1m35s
CI & Build / Build & push image (push) Successful in 32s
The title was `subject — trigger` because the stored title WAS the embedded one, and the join is what makes these kinds rank on the situation they apply to (#2485). Every surface that shows a title then showed the trigger too -- menus, lists and search rows ran to kilobytes. - embeddings.document_title(title, note_type, data, body) joins the trigger from `data` (body fallback) at embed time. Idempotent: an un-migrated composed title comes out the same, never doubled. The embed path, the startup backfill and the dedup gate's semantic signal all use it, so the embedded text -- and every vector -- is unchanged. - Writers store the subject: snippet create/update (service, REST, MCP) and lesson_document. Both compose_title helpers are removed. - Readers: dedup takes `data`; the menus strip the embedded title from a passage; list rows project `when_to_use`, which SnippetListView reads. - 0108 rewrites existing rows on an exact `' — ' || <own trigger>` suffix with raw SQL, leaving updated_at alone so the backfill does not re-embed the corpus for identical vectors. Downgrade recomposes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1920 lines
86 KiB
Python
1920 lines
86 KiB
Python
"""Semantic note search via fastembed (in-process ONNX, no external service).
|
||
|
||
Embeddings are stored as JSONB lists in the note_embeddings table (one row per
|
||
note). All search operations degrade gracefully — if the embedder fails to
|
||
initialize the callers fall back to keyword search.
|
||
|
||
Model: BAAI/bge-small-en-v1.5 (384-dim). The first call downloads the model
|
||
into `FASTEMBED_CACHE_DIR` (defaults to /data/fastembed-cache, a mounted
|
||
volume so subsequent boots are instant).
|
||
"""
|
||
|
||
import asyncio
|
||
import logging
|
||
import math
|
||
import os
|
||
|
||
from collections.abc import Sequence
|
||
|
||
from typing import TYPE_CHECKING
|
||
|
||
from sqlalchemy import delete, func, or_, select
|
||
|
||
from scribe.models import async_session
|
||
from scribe.models.embedding import NoteEmbedding, RuleEmbedding
|
||
from scribe.models.note import Note
|
||
from scribe.models.task_log import TaskLog
|
||
from scribe.services.access import can_read_project, notes_visibility_clause
|
||
|
||
if TYPE_CHECKING: # resolves forward refs without importing at runtime
|
||
from scribe.models.milestone import Milestone
|
||
from scribe.models.rulebook import Rule
|
||
from scribe.models.system import System
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Minimum cosine similarity to include a note in context results.
|
||
# bge-small-en-v1.5 produces unit-normalized vectors, so range is [-1, 1].
|
||
# 0.45 keeps only genuinely relevant notes; lower values like 0.30 let in
|
||
# loosely-related results that pad the sidebar without adding real value.
|
||
_SIMILARITY_THRESHOLD = 0.45
|
||
|
||
# The floor for INTERACTIVE, human-facing feeds — REST /api/search, Browse
|
||
# search, and the list views' semantic `q`. Deliberately looser than the agent
|
||
# default above: a human scanning a result list gets value from loosely-related
|
||
# hits an agent would be misled by. One constant, because this number was
|
||
# written as `0.3` in two files with the reasoning attached to only one of
|
||
# them — the exact shape where a later tuner moves one and not the other
|
||
# (#2463 finding 3).
|
||
INTERACTIVE_SEARCH_THRESHOLD = 0.3
|
||
|
||
# Public alias so callers (and telemetry) can record the effective default
|
||
# threshold without reaching for the underscored name.
|
||
DEFAULT_SIMILARITY_THRESHOLD = _SIMILARITY_THRESHOLD
|
||
|
||
_MODEL_NAME = "BAAI/bge-small-en-v1.5"
|
||
_CACHE_DIR = os.environ.get("FASTEMBED_CACHE_DIR", "/data/fastembed-cache")
|
||
|
||
_model = None # lazy singleton; first call downloads model files
|
||
_model_lock = asyncio.Lock()
|
||
|
||
|
||
async def _get_model():
|
||
"""Return the singleton fastembed.TextEmbedding instance, loading on first call."""
|
||
global _model
|
||
if _model is None:
|
||
async with _model_lock:
|
||
if _model is None:
|
||
# Defer the import so module import doesn't pull in onnxruntime
|
||
# for non-embedding code paths (cheaper cold-start for tests etc.)
|
||
from fastembed import TextEmbedding
|
||
_model = await asyncio.to_thread(
|
||
TextEmbedding,
|
||
model_name=_MODEL_NAME,
|
||
cache_dir=_CACHE_DIR,
|
||
)
|
||
logger.info("Loaded fastembed model %s (cache: %s)", _MODEL_NAME, _CACHE_DIR)
|
||
return _model
|
||
|
||
|
||
async def get_embedding(text: str) -> list[float]:
|
||
"""Get an embedding vector for the given text.
|
||
|
||
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(texts)))
|
||
return [v.tolist() for v in vecs]
|
||
|
||
|
||
def _cosine_similarity(a: list[float], b: list[float]) -> float:
|
||
"""Cosine similarity between two vectors. Returns 0 for zero-length or
|
||
mismatched-length inputs (defensive — mixed-dim vectors can sneak in
|
||
across the migration boundary)."""
|
||
if not a or not b or len(a) != len(b):
|
||
return 0.0
|
||
dot = sum(x * y for x, y in zip(a, b))
|
||
mag_a = math.sqrt(sum(x * x for x in a))
|
||
mag_b = math.sqrt(sum(x * x for x in b))
|
||
if mag_a == 0.0 or mag_b == 0.0:
|
||
return 0.0
|
||
return dot / (mag_a * mag_b)
|
||
|
||
|
||
# How much a superseded record is pushed down the ranking (#278).
|
||
#
|
||
# Chosen against a measurement, not by feel. On 2026-08-07 dev-log #2420 sat at
|
||
# 0.6120 on a query made of its own title phrase, 8th, behind #1759 at 0.6506 —
|
||
# a deficit of 0.039 to the top and ~0.014 to its nearest neighbours. A penalty
|
||
# of 0.05 clears that whole band, so demoting a cluster's stale members actually
|
||
# reorders it rather than shuffling within a tie.
|
||
#
|
||
# It is deliberately NOT large. Supersession is a claim about SOME of a record's
|
||
# content, so a superseded note that strongly answers a question nothing else
|
||
# answers should still surface — just behind anything comparable that is
|
||
# current. A penalty big enough to bury it outright would be hiding by another
|
||
# name, which is the thing the operator ruled out.
|
||
_SUPERSESSION_PENALTY = 0.05
|
||
|
||
# Candidates fetched per requested result when a re-rank follows. Three ranks of
|
||
# headroom is far more than a 0.05 penalty can move anything through in a corpus
|
||
# 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
|
||
) -> list[tuple[float, "Note"]]:
|
||
"""Push superseded records below their equals, then take the top `limit`.
|
||
|
||
The penalty is applied to the RANKING score and the returned score, so
|
||
downstream gates see the adjusted value — the auto-inject margin band in
|
||
particular, which exists to stop near-ties dragging in neighbours and would
|
||
otherwise re-tie exactly what this just separated.
|
||
|
||
It is NOT applied to the relevance threshold: the floor decides whether a
|
||
record is relevant at all, the penalty decides which relevant record comes
|
||
first. Applying it to the floor would drop a superseded record out of the
|
||
results entirely — hiding, which is the one thing this must not do.
|
||
|
||
Stable within a tie: Python's sort preserves the distance order the database
|
||
already established, so equal-scoring records keep their original sequence
|
||
rather than reshuffling per call.
|
||
"""
|
||
if not scored:
|
||
return []
|
||
from scribe.services.supersession import superseded_ids
|
||
|
||
try:
|
||
stale = await superseded_ids([int(note.id) for _score, note in scored])
|
||
except Exception:
|
||
# Fail OPEN, and the direction matters: ranking without the penalty is
|
||
# the behaviour that shipped for months. Returning nothing, or raising,
|
||
# would turn a supersession-lookup hiccup into a broken search.
|
||
logger.warning("Supersession lookup failed — ranking unpenalised", exc_info=True)
|
||
return scored[:limit]
|
||
|
||
if not stale:
|
||
return scored[:limit]
|
||
adjusted = [
|
||
(score - _SUPERSESSION_PENALTY if int(note.id) in stale else score, note)
|
||
for score, note in scored
|
||
]
|
||
adjusted.sort(key=lambda pair: pair[0], reverse=True)
|
||
return adjusted[:limit]
|
||
|
||
|
||
def embedding_text(title: str | None, body: str | None) -> str:
|
||
"""`title\\n{body}` — the atomic join every embedded document is built from.
|
||
|
||
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.
|
||
|
||
Since the chunking build (#280) this is a BUILDING BLOCK, not the whole
|
||
story: the document shape a record is embedded as is `chunk_document`
|
||
below, which calls this once per chunk. Callers that want "the text this
|
||
note is embedded as" want `chunk_document`; this stays public because the
|
||
two functions are one contract and the guard in test_embedding_text pins
|
||
both.
|
||
"""
|
||
title = title or ""
|
||
body = body or ""
|
||
return f"{title}\n{body}".strip() if body else title
|
||
|
||
|
||
# The join between a situation-keyed record's subject and its trigger. A
|
||
# CONSTANT because `untrigger_title` below has to spell the same thing to undo
|
||
# it, and two literals that must match are one edit away from not matching.
|
||
TRIGGER_SEP = " — "
|
||
|
||
|
||
def trigger_title(subject: str | None, trigger: str | None) -> str:
|
||
"""`{subject} — {trigger}` — the title half of a situation-keyed document.
|
||
|
||
ONE definition, because this join had three — rules, snippets, and a
|
||
fourth for lessons (milestone 385) — the shape #3207 records, where a fix
|
||
or an improvement then has to be found in N places by someone who does not
|
||
know N. Since milestone 427 it builds EMBEDDED titles only: `rule_document`
|
||
for rules and `document_title` for snippets and lessons. No stored title
|
||
carries it.
|
||
|
||
WHY THE JOIN MATTERS AT ALL, measured in note #2485: 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 the cause was this title plus the
|
||
same trigger repeated in the body, so purpose appears twice in a short
|
||
document and dominates the vector. Every kind that must be findable by WHEN
|
||
IT APPLIES rather than what it is about is built on this line.
|
||
|
||
Either side alone is returned as-is: a record with no trigger yet degrades
|
||
to its subject and still embeds, just less sharply — which is an argument
|
||
for backfilling triggers, not for padding the title with whatever text is
|
||
to hand.
|
||
"""
|
||
subject = (subject or "").strip()
|
||
trigger = (trigger or "").strip()
|
||
if subject and trigger:
|
||
return f"{subject}{TRIGGER_SEP}{trigger}"
|
||
return subject or trigger
|
||
|
||
|
||
def untrigger_title(title: str | None, trigger: str | None) -> str:
|
||
"""The subject back out of a `trigger_title` — the inverse of the join.
|
||
|
||
Kept HERE, beside the join, for the reason the join itself was
|
||
consolidated: a separator spelled in two files is a separator that will one
|
||
day be changed in one of them. #3207 records the shape — derive it before
|
||
the third copy — and an inverse written in a caller is that third copy
|
||
wearing a different name.
|
||
|
||
Needs the trigger passed in rather than guessing at the separator, because
|
||
a subject may legitimately contain an em dash. Given the trigger, the
|
||
suffix is exact and the split cannot be wrong.
|
||
|
||
Degrades to the whole title when the suffix is absent — a record written
|
||
before the join existed, or one with no trigger yet, still answers with
|
||
something a human recognises rather than with "".
|
||
"""
|
||
title = (title or "").strip()
|
||
trigger = (trigger or "").strip()
|
||
if not trigger:
|
||
return title
|
||
suffix = f"{TRIGGER_SEP}{trigger}"
|
||
if title.endswith(suffix):
|
||
return title[: -len(suffix)].strip()
|
||
return title
|
||
|
||
|
||
# The `data` key each trigger-keyed note kind mirrors its trigger under. Rules
|
||
# are not here: they keep the trigger in a column and `rule_document` builds
|
||
# their document from it.
|
||
_TRIGGER_DATA_KEYS = {"snippet": "when_to_use", "lesson": "when_to_apply"}
|
||
|
||
|
||
def document_title(
|
||
title: str | None, note_type: str | None, data: dict | None = None,
|
||
body: str | None = None,
|
||
) -> str | None:
|
||
"""The title a note is EMBEDDED under — its stored title, plus its trigger.
|
||
|
||
Milestone 427. A snippet's or lesson's STORED title is its subject alone;
|
||
the trigger lives in `data` (decision #4157). It still has to reach the
|
||
vector — the `subject — trigger` join is what makes these kinds rank on
|
||
the situation they apply to (#2485) — so it is joined HERE, at embed time,
|
||
rather than being carried in a title every listing then has to show.
|
||
|
||
IDEMPOTENT, and that is what makes the migration safe: a title that is
|
||
already composed (a row not yet migrated, an old backup restored) is
|
||
untriggered first, so it comes out the same and never doubled. The text is
|
||
byte-identical to what these kinds were embedded as before, so no vector
|
||
moves and the floors tuned against them stay calibrated.
|
||
|
||
`body` is the fallback when the mirror is missing, read by the kind's own
|
||
parser — the same degrade-to-the-body each kind's reader already has.
|
||
Every other kind, and a record with no trigger, keeps its title as-is.
|
||
"""
|
||
key = _TRIGGER_DATA_KEYS.get(note_type or "")
|
||
if key is None:
|
||
return title
|
||
trigger = ((data or {}).get(key) or "").strip() if isinstance(data, dict) else ""
|
||
if not trigger and body:
|
||
from types import SimpleNamespace
|
||
if note_type == "lesson":
|
||
from scribe.services.lessons import lesson_trigger
|
||
trigger = lesson_trigger(SimpleNamespace(data=None, body=body))
|
||
else:
|
||
from scribe.services.snippets import parse_snippet_fields
|
||
trigger = parse_snippet_fields(title or "", body).get("when_to_use", "")
|
||
if not trigger:
|
||
return title
|
||
return trigger_title(untrigger_title(title, trigger), trigger)
|
||
|
||
|
||
# --- chunking (#280): the document shape ------------------------------------
|
||
#
|
||
# bge-small reads at most 512 tokens and fastembed silently truncates the rest,
|
||
# so a single whole-document vector loses everything past ~400 words — for a
|
||
# long dev-log, three quarters of the record was PERMANENTLY invisible to
|
||
# search. The fix is the document shape: one vector per meaningful chunk, and a
|
||
# record is as findable as its best-matching section.
|
||
|
||
# Bumped whenever THE DOCUMENT A RECORD IS EMBEDDED AS can change for a record
|
||
# that itself has not changed. Stored on every note_embeddings row so the
|
||
# startup backfill re-embeds exactly the stale notes — a version comparison
|
||
# instead of the table wipe migrations 0067/0077 had to do.
|
||
#
|
||
# Stated that way rather than as "chunk_document's output for the same input",
|
||
# which is what it used to say: `chunk_document` is only the last step, and
|
||
# version 2 moves without touching it. A task's document now carries its work
|
||
# logs (#4251), so every task that has one embeds differently than it did while
|
||
# its title, body and the chunker are all untouched — exactly the case the
|
||
# narrower wording would have read as "nothing to re-embed".
|
||
#
|
||
# 1 → 2: work logs joined the task document.
|
||
CHUNKER_VERSION = 2
|
||
|
||
|
||
# The public name of the space every score lives in, and the two facts that
|
||
# can invalidate a tuned number (#4104).
|
||
#
|
||
# EMBEDDING_MODEL is `_MODEL_NAME` under a name other modules may read. It was
|
||
# private until this step, which is precisely why nothing outside this file
|
||
# could state what space a threshold was measured in — a floor is a distance in
|
||
# THIS model's geometry and means nothing in another's.
|
||
#
|
||
# The pair is what a stamp is made of, and the pairing is the point: a score
|
||
# changes when the model changes (different geometry) OR when the document
|
||
# shape changes (different text embedded for the same record). Either one
|
||
# invalidates a number that was measured before it.
|
||
EMBEDDING_MODEL = _MODEL_NAME
|
||
|
||
|
||
def calibration_stamp() -> dict:
|
||
"""What a tuned retrieval number was measured against.
|
||
|
||
ONE definition, because the alternative is each reader assembling the pair
|
||
and one of them forgetting a half. Returned as a dict rather than a string
|
||
so a mismatch can say WHICH half moved — "the model changed" and "the
|
||
chunker changed" call for different responses, and a fused string can only
|
||
report that something did.
|
||
|
||
Deliberately says nothing about the CHAT model. A Claude upgrade changes no
|
||
score here and must never raise a recalibration prompt: a false alarm on
|
||
this surface teaches an operator to ignore the true one. `tests/
|
||
test_calibration_stamp.py` asserts that absence rather than trusting it.
|
||
"""
|
||
return {"embedding_model": EMBEDDING_MODEL, "shape_version": CHUNKER_VERSION}
|
||
|
||
# Character budget approximating the model window. Tokens-per-char varies by
|
||
# content — ~4 chars/token for prose, closer to 3 for code and tables — so 1400
|
||
# chars sits at roughly 350-470 tokens, leaving headroom for the title prefixed
|
||
# to every chunk. Deliberately conservative: our own measurement (#2485) says
|
||
# shorter, single-topic documents embed SHARPER, so the cost of over-splitting
|
||
# is a few extra cheap vectors while the cost of under-splitting is truncation —
|
||
# the exact data loss this exists to end.
|
||
_CHUNK_CHAR_BUDGET = 1400
|
||
|
||
_HEADING_RE = None # compiled lazily below to keep re import local
|
||
|
||
|
||
def _split_sections(body: str) -> list[str]:
|
||
"""Split a markdown body at heading lines, fence-aware.
|
||
|
||
Each section is a heading line plus everything under it; text before the
|
||
first heading is its own section. Heading-looking lines inside ``` / ~~~
|
||
code fences do not split — a commented `# step` in a recorded shell snippet
|
||
is content, not structure.
|
||
"""
|
||
import re
|
||
global _HEADING_RE
|
||
if _HEADING_RE is None:
|
||
_HEADING_RE = re.compile(r"^#{1,6}\s")
|
||
|
||
sections: list[list[str]] = [[]]
|
||
in_fence = False
|
||
for line in body.splitlines():
|
||
if line.lstrip().startswith(("```", "~~~")):
|
||
in_fence = not in_fence
|
||
if not in_fence and _HEADING_RE.match(line) and sections[-1]:
|
||
sections.append([line])
|
||
else:
|
||
sections[-1].append(line)
|
||
return ["\n".join(chunk).strip() for chunk in sections if any(s.strip() for s in chunk)]
|
||
|
||
|
||
def _split_paragraphs(section: str, budget: int) -> list[str]:
|
||
"""Break one oversize section into budget-sized pieces at paragraph
|
||
boundaries, hard-splitting only a single paragraph that alone exceeds the
|
||
budget (a monster table or code block — split at line boundaries so no
|
||
content is dropped, which is the entire point of this module)."""
|
||
pieces: list[str] = []
|
||
current = ""
|
||
for para in section.split("\n\n"):
|
||
while len(para) > budget:
|
||
# Hard split: prefer the last newline inside the budget so lines
|
||
# stay whole, then the last space so words do; a clean char cut is
|
||
# the final resort for one enormous unbroken token.
|
||
cut = para.rfind("\n", 0, budget)
|
||
if cut <= 0:
|
||
cut = para.rfind(" ", 0, budget)
|
||
if cut <= 0:
|
||
cut = budget
|
||
head, para = para[:cut], para[cut:].lstrip("\n ")
|
||
if current:
|
||
pieces.append(current)
|
||
current = ""
|
||
pieces.append(head.strip())
|
||
if not para.strip():
|
||
continue
|
||
candidate = f"{current}\n\n{para}" if current else para
|
||
if len(candidate) > budget and current:
|
||
pieces.append(current)
|
||
current = para
|
||
else:
|
||
current = candidate
|
||
if current:
|
||
pieces.append(current)
|
||
return pieces
|
||
|
||
|
||
def chunk_document(title: str | None, body: str | None) -> list[str]:
|
||
"""The document(s) a record is embedded AS — one string per chunk.
|
||
|
||
The contract every retrieval surface builds on:
|
||
|
||
- A record that fits the model window yields EXACTLY ONE chunk, identical
|
||
to the historical `title\\nbody` shape — snippets and reference notes,
|
||
the corpus's sharpest records, are byte-for-byte unaffected.
|
||
- A longer record is split at markdown heading boundaries (fence-aware),
|
||
small neighbouring sections merged, oversize sections split at paragraph
|
||
boundaries, so every chunk fits the window. NOTHING is dropped: every
|
||
line of the body lands in some chunk.
|
||
- Every chunk is prefixed with the record's title — each vector carries
|
||
its own topical anchor, the property that makes snippets discriminative
|
||
(#2485). Pieces sub-split from one section also repeat that section's
|
||
heading line, so "which part of which topic" survives the split.
|
||
- An empty record yields [] (callers gate on falsiness to skip embedding).
|
||
|
||
Bump CHUNKER_VERSION when changing anything observable here.
|
||
"""
|
||
single = embedding_text(title, body)
|
||
if not single:
|
||
return []
|
||
if len(single) <= _CHUNK_CHAR_BUDGET:
|
||
return [single]
|
||
|
||
title = title or ""
|
||
# Budget for section content, net of the title prefix added to every chunk.
|
||
budget = max(200, _CHUNK_CHAR_BUDGET - len(title) - 1)
|
||
|
||
# Merge small adjacent sections upward so tiny sections don't each spend a
|
||
# vector, then split anything still over budget at paragraph boundaries.
|
||
merged: list[str] = []
|
||
for section in _split_sections(body or ""):
|
||
if merged and len(merged[-1]) + 2 + len(section) <= budget:
|
||
merged[-1] = f"{merged[-1]}\n\n{section}"
|
||
else:
|
||
merged.append(section)
|
||
|
||
chunks: list[str] = []
|
||
for section in merged:
|
||
if len(section) <= budget:
|
||
chunks.append(embedding_text(title, section))
|
||
continue
|
||
pieces = _split_paragraphs(section, budget)
|
||
first_line = section.split("\n", 1)[0]
|
||
heading = first_line if first_line.lstrip().startswith("#") else ""
|
||
for i, piece in enumerate(pieces):
|
||
# Repeat the section heading on continuation pieces so each vector
|
||
# still knows what topic it is part of.
|
||
if i > 0 and heading and not piece.startswith(heading):
|
||
piece = f"{heading}\n{piece}"
|
||
chunks.append(embedding_text(title, piece))
|
||
return chunks
|
||
|
||
|
||
async def _claim_parent_row(session, id_column, row_id: int, label: str) -> bool:
|
||
"""Lock the record a vector belongs to BEFORE rewriting that vector (#3262).
|
||
|
||
An embedding write and a cascading delete of the same record take the same
|
||
two row locks in OPPOSITE orders. The embedder deletes the old chunk rows
|
||
and then, on INSERT, needs the foreign key's lock on the parent; a delete
|
||
of the parent — or of the rulebook, topic or project above it — locks the
|
||
parent first and cascades down into the chunk rows. That is a cycle, and
|
||
Postgres breaks it by killing one side at random: sometimes the embedding
|
||
write, which is swallowed and invisible, and sometimes the operator's
|
||
delete, which surfaces as a 500 on an operation that should have worked.
|
||
|
||
Claiming the parent first REMOVES the cycle rather than narrowing it.
|
||
Either the embedder arrives first and the delete waits its turn behind it,
|
||
or the delete already holds the row and NOWAIT makes the embedder lose at
|
||
once. The embedder is the side that should lose: a skipped refresh costs a
|
||
stale vector until the next write or the startup backfill, and the other
|
||
outcome costs a person their request.
|
||
|
||
FOR KEY SHARE, not FOR UPDATE — it is precisely the lock the INSERT's
|
||
foreign key would take anyway, so it conflicts with a delete of the parent
|
||
and with nothing else. An ordinary edit of the same record, or a second
|
||
refresh racing this one, is unaffected.
|
||
|
||
Returns False when the row is locked or already gone; the caller skips.
|
||
"""
|
||
try:
|
||
held = (await session.execute(
|
||
select(id_column)
|
||
.where(id_column == row_id)
|
||
# BOTH flags: SQLAlchemy spells the four Postgres row locks as a
|
||
# read/key_share pair, and key_share alone is FOR NO KEY UPDATE —
|
||
# which would make two refreshes of one record fight each other.
|
||
.with_for_update(read=True, key_share=True, nowait=True)
|
||
)).scalar_one_or_none()
|
||
except Exception:
|
||
# LockNotAvailable: this record is being deleted right now. Not an
|
||
# error — the delete wins by design.
|
||
logger.debug("Skipping embedding for %s %d — row is being deleted", label, row_id)
|
||
return False
|
||
if held is None:
|
||
logger.debug("Skipping embedding for %s %d — row is gone", label, row_id)
|
||
return False
|
||
return True
|
||
|
||
|
||
async def _work_log_sections(note_id: int) -> list[tuple[object, str | None]]:
|
||
"""A task's work logs, oldest first, for its embedded document (#4251).
|
||
|
||
Reads the table directly rather than through `task_logs.logs_for_task`,
|
||
because that function asks a PERMISSION question — may this user read this
|
||
task — and there is no user here. An index build acts for the record, and
|
||
the record's vectors carry the owner's `user_id`, so the access decision is
|
||
made once at search time by the clause that already scopes every hit.
|
||
|
||
That also settles what happens on a shared task: a collaborator's log is
|
||
part of the task's document, so it becomes findable by everyone who can
|
||
read the task and by nobody else. The same answer `logs_for_task` gives a
|
||
reader (#4241), which is the point — a log that can be read and not found
|
||
is the half-surface that issue was about.
|
||
|
||
Asked for every note, not only tasks. `upsert_note_embedding` is handed a
|
||
note_id and no kind — and the three writers that call it would each have to
|
||
learn to pass one — so a non-task simply has no rows and gets []. One
|
||
indexed lookup beside an ONNX forward pass over every chunk is not the
|
||
expense worth adding a parameter to three call sites for.
|
||
|
||
Returns [] on any failure. A task whose logs could not be read should embed
|
||
as its own prose rather than not embed at all: less findable is recoverable
|
||
at the next write, unindexed is not.
|
||
"""
|
||
try:
|
||
async with async_session() as session:
|
||
result = await session.execute(
|
||
select(TaskLog.created_at, TaskLog.content)
|
||
.where(TaskLog.task_id == note_id)
|
||
.order_by(TaskLog.created_at.asc(), TaskLog.id.asc())
|
||
)
|
||
return list(result.all())
|
||
except Exception:
|
||
logger.warning(
|
||
"Could not read work logs for note %d; embedding its own prose only",
|
||
note_id, exc_info=True,
|
||
)
|
||
return []
|
||
|
||
|
||
async def upsert_note_embedding(
|
||
note_id: int, user_id: int, title: str | None, body: str | None
|
||
) -> None:
|
||
"""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.
|
||
"""
|
||
title, body = task_document(title, body, await _work_log_sections(note_id))
|
||
chunks = chunk_document(title, body)
|
||
try:
|
||
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
|
||
|
||
try:
|
||
async with async_session() as session:
|
||
if not await _claim_parent_row(session, Note.id, note_id, "note"):
|
||
return
|
||
await session.execute(
|
||
delete(NoteEmbedding).where(NoteEmbedding.note_id == note_id)
|
||
)
|
||
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 %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)
|
||
|
||
|
||
# Kinds that belong to no single project, and are therefore reachable from a
|
||
# project-scoped search of a DIFFERENT project when a caller asks for them
|
||
# (milestone 385 step 3).
|
||
#
|
||
# A lesson is the whole reason this exists. "A better way to think about this
|
||
# problem" is not true only where it was learned, and a lesson confined to its
|
||
# origin project would be unreachable exactly where it is most useful — on the
|
||
# next project, which is the case the kind was created for (#3727).
|
||
# `semantic_search_rules` has always had this property: it scopes by OWNERSHIP
|
||
# rather than by what binds a given project, because "is there a rule about
|
||
# this" is a question asked across a whole rulebook. A lesson asks the same
|
||
# kind of question.
|
||
#
|
||
# Spelled as a literal rather than imported from `services.lessons`, which
|
||
# imports `trigger_title` from this module and would close a cycle. A guard
|
||
# pins the two equal instead.
|
||
GLOBAL_NOTE_TYPES: tuple[str, ...] = ("lesson",)
|
||
|
||
|
||
# Both searches rank WITHOUT the threshold and apply it in Python, so the best
|
||
# rejected score stays observable (#3670). The qualifying set is provably
|
||
# unchanged: rows arrive ordered by distance ascending, so every above-bar row
|
||
# sorts ahead of every below-bar one, and an over-fetch that used to return N
|
||
# above-bar rows returns the same N plus some losers. What changes is only that
|
||
# the losers are now visible instead of discarded inside the query.
|
||
#
|
||
# That visibility is the entire point. A bar can only be judged from the calls
|
||
# it TURNED AWAY — a 0.72 bar rejecting a stream of 0.71s is set too high by a
|
||
# hair, one rejecting 0.30s is working — and those two are indistinguishable
|
||
# from any arrangement of the columns that survive the filter.
|
||
#
|
||
# `report` is how the score gets out without changing what a search RETURNS.
|
||
# Eight of the eleven call sites want hits and nothing else; the three that
|
||
# write telemetry pass a dict and read `best_available_score` back out of it.
|
||
|
||
|
||
def record_best_chunk(report: dict | None, chunks: dict[int, dict]) -> None:
|
||
"""Publish the winning chunk per record into `report["best_chunk"]`.
|
||
|
||
Every semantic search here collapses several chunk rows to the best one per
|
||
record, which means each of them KNOWS which passage earned the hit — and
|
||
each of them used to drop it, leaving every caller to preview the head of
|
||
the document instead. The head is a different span, one the search has
|
||
already scored lower, and nothing in the result said so (#4243).
|
||
|
||
It rides in `report` rather than in the return value because all three
|
||
searches return `list[tuple[float, Record]]` and roughly thirty sites
|
||
unpack that pair; widening it would be an interface change to every one of
|
||
them with nothing to catch a miss (lesson #4207). `report` is already the
|
||
side-channel these functions use for `searched` and `best_available_score`,
|
||
so this adds a key to a channel callers already open.
|
||
|
||
Shape: {record_id: {"index": int, "text": str}}. A caller that passed no
|
||
report simply doesn't get it, and every consumer falls back to the body.
|
||
"""
|
||
if report is None:
|
||
return
|
||
report["best_chunk"] = chunks
|
||
|
||
|
||
async def semantic_search_notes(
|
||
user_id: int,
|
||
query: str,
|
||
exclude_ids: set[int] | None = None,
|
||
limit: int = 8,
|
||
threshold: float = _SIMILARITY_THRESHOLD,
|
||
project_id: int | None = None,
|
||
is_task: bool | None = None,
|
||
note_type: str | Sequence[str] | None = None,
|
||
task_kind: str | Sequence[str] | None = None,
|
||
orphan_only: bool = False,
|
||
include_global_kinds: bool = False,
|
||
scope: str = "own",
|
||
demote_superseded: bool = True,
|
||
system_id: int | None = None,
|
||
report: dict | None = None,
|
||
) -> list[tuple[float, Note]]:
|
||
"""Return up to *limit* (score, note) pairs most relevant to *query*.
|
||
|
||
Scores are cosine similarities in [-1, 1]; only notes at or above
|
||
*threshold* are returned, sorted highest-first.
|
||
|
||
Pass `report` (an empty dict) to learn what the threshold turned away:
|
||
the function sets `report["best_available_score"]` to the highest score
|
||
anything reached, or None when the corpus offered nothing at all. It is
|
||
the only figure that survives a call returning nothing, and therefore the
|
||
only one a bar can be judged from (#3670).
|
||
|
||
It also sets `report["searched"]`: False before anything can return, True
|
||
only where a real result set exists. So an empty query, an unavailable
|
||
embedder and a failed database query all leave it FALSE, and a caller can
|
||
tell a search that found nothing from one that never ran. A caller logging
|
||
telemetry must check it — recording a failed search as a zero-result call
|
||
reports a decline the ranker never made (#3765). ABSENT means no search
|
||
touched the dict at all, which is a stand-in in a test, not a real call.
|
||
|
||
`note_type` narrows to a record kind, or several (e.g. "snippet", or
|
||
("snippet", "note")), for callers that want prior art rather than everything
|
||
embedded.
|
||
|
||
`task_kind` restricts TASKS to the given kinds while leaving non-task notes
|
||
untouched. That asymmetry is the point: "recorded experience" is issues plus
|
||
dev-logs, and those differ on `is_task`, so neither `note_type` nor `is_task`
|
||
alone can express it. With `note_type="note", task_kind="issue"` a caller
|
||
gets fixed problems and durable notes without the open to-do list.
|
||
|
||
`include_global_kinds` lets a project-scoped search ALSO reach the kinds in
|
||
GLOBAL_NOTE_TYPES — records that belong to no single project — so a lesson
|
||
written on one project is found from another. It widens the PROJECT filter
|
||
only: a caller that also passes `note_type` still gets exactly the kinds it
|
||
asked for, so narrowing to snippets does not quietly acquire lessons.
|
||
|
||
Off by default, because two callers depend on the project filter holding.
|
||
The near-duplicate gate compares a record only against its own project on
|
||
purpose, and a globally-visible kind would let a lesson block an unrelated
|
||
note's create on a project its author never touched. Ordinary note recall
|
||
is project-scoped for the same reason — the point of the carve-out is that
|
||
ONE kind escapes, not that scoping is weaker.
|
||
|
||
`scope` ("own" | "browse" | "read", see access.notes_visibility_clause)
|
||
decides how far this may see. It exists because this one function serves
|
||
three different kinds of act: an explicit search, which should reach
|
||
everything the caller may read; passive auto-injection, which must not pull
|
||
an unrequested record into their context; and the near-duplicate gate, whose
|
||
verdict must not depend on other people's notes at all. Defaults to "own" so
|
||
a caller that forgets is wrong in the safe direction.
|
||
|
||
Ranking and the top-k cut happen in Postgres via pgvector's cosine-distance
|
||
operator (`<=>`, exposed as ``Vector.cosine_distance``) backed by the HNSW
|
||
index from migration 0067 — so this is an indexed ``ORDER BY ... LIMIT k``
|
||
rather than a full-table scan. Cosine distance is ``1 - cosine_similarity``,
|
||
so a similarity floor of *threshold* is a distance ceiling of
|
||
``1 - threshold`` and similarity is recovered as ``1 - distance``.
|
||
|
||
`demote_superseded` applies the supersession penalty (#278): a record a
|
||
later note claims to have overtaken ranks below its equals. Callers asking
|
||
"what is the current answer" want it; the near-duplicate gate does NOT, and
|
||
passes False — a superseded record is still a duplicate of what you are
|
||
about to write, and demoting it there would let the same note be recorded
|
||
twice, the second time invisibly.
|
||
|
||
Returns an empty list if the embedder is unavailable or on any error.
|
||
"""
|
||
# Stamped FALSE before anything can return, flipped True only where a real
|
||
# result set exists (#3765). Every early return below leaves it false, so a
|
||
# caller can tell a search that found nothing from one that never ran. It
|
||
# has to be the first thing done to `report`: a return added above this
|
||
# line would leave the key ABSENT, which reads as "no caller asked".
|
||
if report is not None:
|
||
report["searched"] = False
|
||
if not query or not query.strip():
|
||
return []
|
||
try:
|
||
query_vec = await get_embedding(query)
|
||
except Exception:
|
||
logger.debug("Semantic search skipped — embedder unavailable")
|
||
return []
|
||
|
||
# Distance ceiling equivalent to the similarity floor. Clamp to the valid
|
||
# cosine-distance range [0, 2] so a threshold of, say, -1 doesn't produce a
|
||
# nonsensical ceiling.
|
||
distance = NoteEmbedding.embedding.cosine_distance(query_vec)
|
||
|
||
try:
|
||
async with async_session() as session:
|
||
# Scope on Note, not NoteEmbedding.user_id: the embedding row belongs
|
||
# to the note's owner, so filtering it would pin every scope to "own"
|
||
# and leave shared records unreachable by meaning.
|
||
stmt = (
|
||
# chunk_index/chunk_text ride along so the collapse below can
|
||
# say WHICH passage matched. Without them the caller is left
|
||
# previewing the head of the body — a span this query has
|
||
# already determined is not why the record ranked (#4243).
|
||
select(
|
||
Note,
|
||
distance.label("distance"),
|
||
NoteEmbedding.chunk_index,
|
||
NoteEmbedding.chunk_text,
|
||
)
|
||
.select_from(NoteEmbedding)
|
||
.join(Note, NoteEmbedding.note_id == Note.id)
|
||
.where(
|
||
notes_visibility_clause(user_id, scope),
|
||
Note.deleted_at.is_(None),
|
||
)
|
||
)
|
||
if orphan_only:
|
||
stmt = stmt.where(Note.project_id.is_(None))
|
||
elif project_id is not None:
|
||
in_project = Note.project_id == project_id
|
||
if include_global_kinds:
|
||
in_project = or_(
|
||
in_project, Note.note_type.in_(GLOBAL_NOTE_TYPES)
|
||
)
|
||
stmt = stmt.where(in_project)
|
||
# Narrow to records tagged to one System (subsystem/area). An
|
||
# association filter, not a ranking signal — membership in the
|
||
# candidate set, decided before scoring, like project_id above.
|
||
if system_id is not None:
|
||
from scribe.models.system import RecordSystem
|
||
stmt = stmt.where(
|
||
select(RecordSystem.id)
|
||
.where(
|
||
RecordSystem.note_id == Note.id,
|
||
RecordSystem.system_id == system_id,
|
||
)
|
||
.exists()
|
||
)
|
||
if is_task is True:
|
||
stmt = stmt.where(Note.status.isnot(None))
|
||
elif is_task is False:
|
||
stmt = stmt.where(Note.status.is_(None))
|
||
# Narrow to one kind of record, or several. Composes with is_task
|
||
# rather than replacing it — 'snippet' is a non-task note_type, so a
|
||
# caller asking for prior art gets snippets and not the dev-log that
|
||
# mentions them.
|
||
if note_type:
|
||
kinds = [note_type] if isinstance(note_type, str) else list(note_type)
|
||
stmt = stmt.where(Note.note_type.in_(kinds))
|
||
# Restrict TASKS to certain kinds while leaving notes alone. A note
|
||
# has no task_kind that means anything, so a plain `.in_()` would
|
||
# drop every dev-log — which is exactly the record a caller asking
|
||
# for prior experience wants most.
|
||
if task_kind:
|
||
tkinds = [task_kind] if isinstance(task_kind, str) else list(task_kind)
|
||
stmt = stmt.where(
|
||
or_(Note.status.is_(None), Note.task_kind.in_(tkinds))
|
||
)
|
||
if exclude_ids:
|
||
stmt = stmt.where(NoteEmbedding.note_id.notin_(exclude_ids))
|
||
# OVER-FETCH when a re-rank follows, so the demotion can actually
|
||
# move something. Demoting after a LIMIT k would be theatre: the cut
|
||
# already happened, so a superseded record pushed down still sits in
|
||
# the results and the live record that should have replaced it was
|
||
# never fetched.
|
||
#
|
||
# Ordering stays on RAW distance so pgvector's HNSW index still
|
||
# serves it (migration 0067). Ordering by `distance + penalty`
|
||
# instead would be exact, and would turn an indexed top-k into a
|
||
# scan-and-sort of every embedded note.
|
||
#
|
||
# The cost of that trade, stated plainly: a live record outside the
|
||
# over-fetch window cannot be promoted into the results. With a
|
||
# 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 * _CHUNK_OVERFETCH * (
|
||
_SUPERSESSION_OVERFETCH if demote_superseded else 1
|
||
)
|
||
# NO threshold predicate — see the note above this function. The
|
||
# bar is applied after the collapse, where the rejected scores can
|
||
# still be seen.
|
||
stmt = stmt.order_by(distance.asc()).limit(fetch)
|
||
rows = list((await session.execute(stmt)).all())
|
||
except Exception:
|
||
logger.warning("Failed to query note embeddings", exc_info=True)
|
||
return []
|
||
|
||
# Collapse chunk rows to BEST-CHUNK-PER-NOTE (#280): rows arrive ordered by
|
||
# 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()
|
||
# The winning row IS the best chunk, by the ordering above — so this is the
|
||
# one place that knows which passage earned the hit. Kept beside the score
|
||
# rather than returned with it: the return type is list[tuple[float, Note]]
|
||
# and ten callers unpack it at ~18 sites, so widening the tuple would be an
|
||
# interface change to every one of them with nothing to catch the misses
|
||
# (lesson #4207). `report` is the side-channel this function already uses
|
||
# for best_available_score.
|
||
best_chunk: dict[int, dict] = {}
|
||
for note, dist, chunk_index, chunk_text in rows:
|
||
if int(note.id) in seen:
|
||
continue
|
||
seen.add(int(note.id))
|
||
scored.append((1.0 - float(dist), note))
|
||
best_chunk[int(note.id)] = {
|
||
"index": int(chunk_index),
|
||
"text": chunk_text or "",
|
||
}
|
||
# The best score anything reached, bar or no bar. Recorded BEFORE the
|
||
# filter because a call that returns nothing is exactly when it matters.
|
||
if report is not None:
|
||
# `searched` is what stops a null score meaning four things (#3765).
|
||
# Every early return above — empty query, embedder down, and the broad
|
||
# `except` around the query itself — leaves this key ABSENT, so a
|
||
# caller can tell "I looked and there was nothing" from "I never
|
||
# looked" and from "the query failed". Set here, at the one point past
|
||
# which a real result set exists.
|
||
report["searched"] = True
|
||
# ONE unpack, so the score and the id cannot describe different records
|
||
# (#3807). Splitting these into two expressions is how a later edit
|
||
# pairs a score with its neighbour's id.
|
||
best = scored[0] if scored else None
|
||
report["best_available_score"] = best[0] if best else None
|
||
report["best_available_id"] = int(best[1].id) if best else None
|
||
scored = [pair for pair in scored if pair[0] >= threshold]
|
||
if not demote_superseded:
|
||
final = scored[:limit]
|
||
else:
|
||
final = await _apply_supersession_penalty(scored, limit)
|
||
# Only for what actually came back, so a caller can key straight off the
|
||
# results without carrying chunks for records it never saw.
|
||
record_best_chunk(report, {
|
||
int(n.id): best_chunk[int(n.id)]
|
||
for _s, n in final
|
||
if int(n.id) in best_chunk
|
||
})
|
||
return final
|
||
|
||
|
||
# --- backfill correctness: scan ids, read text at embed time (#4262) --------
|
||
#
|
||
# A backfill used to SELECT the text alongside the id and then iterate that
|
||
# snapshot, sleeping between records. On a corpus of any size the run takes
|
||
# minutes, and anything edited inside that window was re-embedded by its own
|
||
# write path and then OVERWRITTEN with the pre-edit text the scan had captured
|
||
# — stamped at the current version, so the next boot considered it current and
|
||
# never repaired it. A record left findable by what it used to say, silently,
|
||
# until someone happened to edit it again.
|
||
#
|
||
# The two functions below are the fix, and they are separate on purpose: one
|
||
# stops it happening, the other repairs what already happened.
|
||
|
||
|
||
async def _current_row(columns, id_column, row_id: int):
|
||
"""A record's text AS IT IS NOW, read at the moment it is embedded.
|
||
|
||
One indexed primary-key read, negligible beside the forward pass that
|
||
follows it — and it turns a backfill from "replay a snapshot" into "repair
|
||
to current truth". The residual race shrinks from the length of the whole
|
||
run to the width of one record's processing, which `_claim_parent_row` and
|
||
the single-transaction replacement then narrow further.
|
||
|
||
None when the row is gone: deleted between the scan and here, which is an
|
||
ordinary outcome of a long run and not an error.
|
||
"""
|
||
try:
|
||
async with async_session() as session:
|
||
return (
|
||
await session.execute(select(*columns).where(id_column == row_id))
|
||
).first()
|
||
except Exception:
|
||
logger.warning("Backfill could not re-read record %s", row_id, exc_info=True)
|
||
return None
|
||
|
||
|
||
async def _vectors_older_than_their_record(
|
||
session, emb_id_column, emb_updated_column, rec_id_column, rec_updated_column
|
||
) -> set[int]:
|
||
"""Ids whose text changed AFTER their vectors were written.
|
||
|
||
The repair half, and a standing guard worth more than the bug that prompted
|
||
it. A version comparison alone asks "was this embedded by the current
|
||
chunker" and cannot see a record that was embedded by it from stale text —
|
||
nor one whose `embed_note` never ran at all, because the write happened
|
||
with no event loop, or the refresh was swallowed, or the process died
|
||
between the commit and the index.
|
||
|
||
Every one of those looks identical from the version column and identical
|
||
from the outside: a record that answers to what it used to say. This is the
|
||
only signal that separates them, and it costs one grouped read at startup.
|
||
|
||
A guard that refuses to write a bad value does not undo the bad values
|
||
already stored (#4202) — the rows are the thing that has to change, and
|
||
this is what changes them.
|
||
"""
|
||
newest = (
|
||
select(
|
||
emb_id_column.label("rid"),
|
||
func.max(emb_updated_column).label("embedded_at"),
|
||
)
|
||
.group_by(emb_id_column)
|
||
.subquery()
|
||
)
|
||
rows = await session.execute(
|
||
select(rec_id_column)
|
||
.join(newest, newest.c.rid == rec_id_column)
|
||
.where(rec_updated_column > newest.c.embedded_at)
|
||
)
|
||
return {int(row[0]) for row in rows.fetchall()}
|
||
|
||
|
||
async def _tasks_logged_since_embedding(session) -> set[int]:
|
||
"""Task ids whose newest work log is newer than their newest vector.
|
||
|
||
The note corpus needs this on top of the plain timestamp comparison, and
|
||
the reason is a consequence of #4251 that is easy to miss: a task's
|
||
embedded document carries its work logs, but a log lives in its own table,
|
||
so writing one does NOT move `notes.updated_at`. A task re-embedded from
|
||
stale text during a backfill therefore looks current by every other
|
||
signal — the version is right and the note's own timestamp is older than
|
||
the vectors — while the document those vectors encode is out of date.
|
||
|
||
These are the exact records the backfill race was most likely to hit,
|
||
because a session logging work is what "forward work during a long
|
||
backfill" MEANS.
|
||
"""
|
||
newest_vector = (
|
||
select(
|
||
NoteEmbedding.note_id.label("rid"),
|
||
func.max(NoteEmbedding.updated_at).label("embedded_at"),
|
||
)
|
||
.group_by(NoteEmbedding.note_id)
|
||
.subquery()
|
||
)
|
||
newest_log = (
|
||
select(
|
||
TaskLog.task_id.label("tid"),
|
||
func.max(TaskLog.updated_at).label("logged_at"),
|
||
)
|
||
.group_by(TaskLog.task_id)
|
||
.subquery()
|
||
)
|
||
rows = await session.execute(
|
||
select(newest_log.c.tid)
|
||
.join(newest_vector, newest_vector.c.rid == newest_log.c.tid)
|
||
.where(newest_log.c.logged_at > newest_vector.c.embedded_at)
|
||
)
|
||
return {int(row[0]) for row in rows.fetchall()}
|
||
|
||
|
||
async def backfill_note_embeddings() -> None:
|
||
"""(Re-)embed every note whose vectors are missing, outdated or WRONG.
|
||
|
||
Three conditions, and the third is the one a version number cannot see:
|
||
|
||
1. no vectors at all;
|
||
2. vectors from an older chunker — version-awareness is what makes a
|
||
document-shape change deployable. Migration 0077 cleared the table
|
||
once; every later CHUNKER_VERSION bump re-embeds here instead;
|
||
3. vectors OLDER THAN THE TEXT THEY CLAIM TO ENCODE. A record embedded by
|
||
the current chunker from text that has since changed is current by
|
||
every other signal and answers to what it used to say (#4262).
|
||
|
||
Runs as a background task at startup, after the serving flag (#4181), with
|
||
a small sleep between notes so a large backfill doesn't peg CPU. It is
|
||
resumable by construction: progress IS the version stamp on each record's
|
||
rows, so a restart re-queries and skips what finished — there is no
|
||
checkpoint to lose.
|
||
"""
|
||
try:
|
||
async with async_session() as session:
|
||
current = {
|
||
row[0]
|
||
for row in (
|
||
await session.execute(
|
||
select(NoteEmbedding.note_id).where(
|
||
NoteEmbedding.chunker_version == CHUNKER_VERSION
|
||
)
|
||
)
|
||
).fetchall()
|
||
}
|
||
# IDS ONLY. The text is read per note below, at the moment it is
|
||
# embedded — a scan that carried the text would spend the whole run
|
||
# holding a snapshot and overwrite anything edited inside it.
|
||
all_ids = [
|
||
row[0]
|
||
for row in (await session.execute(select(Note.id))).fetchall()
|
||
]
|
||
stale = await _vectors_older_than_their_record(
|
||
session, NoteEmbedding.note_id, NoteEmbedding.updated_at,
|
||
Note.id, Note.updated_at,
|
||
)
|
||
# A work log is part of a task's document but lives in its own
|
||
# table, so writing one leaves `notes.updated_at` untouched and the
|
||
# comparison above blind to it.
|
||
stale |= await _tasks_logged_since_embedding(session)
|
||
except Exception:
|
||
logger.warning("Embedding backfill: failed to query notes", exc_info=True)
|
||
return
|
||
|
||
notes_to_embed = [i for i in all_ids if i not in current or i in stale]
|
||
if not notes_to_embed:
|
||
logger.info("Embedding backfill: all notes current at chunker v%d", CHUNKER_VERSION)
|
||
return
|
||
|
||
logger.info(
|
||
"Embedding backfill: embedding %d note(s) at chunker v%d (%d with vectors "
|
||
"older than the record they describe)",
|
||
len(notes_to_embed), CHUNKER_VERSION, len(stale),
|
||
)
|
||
success = 0
|
||
for note_id in notes_to_embed:
|
||
row = await _current_row(
|
||
(Note.user_id, Note.title, Note.body, Note.note_type, Note.data), Note.id, note_id,
|
||
)
|
||
if row is None:
|
||
continue # deleted between the scan and here
|
||
user_id, title, body, note_type, data = row
|
||
# The EMBEDDED title, as the write path builds it (milestone 427).
|
||
title = document_title(title, note_type, data, body)
|
||
if not chunk_document(title, body):
|
||
continue
|
||
await upsert_note_embedding(note_id, user_id, title, body)
|
||
success += 1
|
||
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) ────────────────────────────────────
|
||
|
||
# The heading a work log gets inside its task's embedded document. A CONSTANT
|
||
# because it is load-bearing twice over: `_split_sections` splits on it, so it
|
||
# is what keeps a log from being merged into the task's own prose, and it is
|
||
# what a reader sees at the top of a matched passage — "this is a log entry,
|
||
# not the task's description". Changing it changes the chunk boundaries of
|
||
# every task that has one, which is a CHUNKER_VERSION move.
|
||
WORK_LOG_HEADING = "## Work log"
|
||
|
||
|
||
def task_document(
|
||
title: str | None,
|
||
body: str | None,
|
||
logs: "Sequence[tuple[object, str | None]]" = (),
|
||
) -> tuple[str | None, str | None]:
|
||
"""The (title, body) a TASK is EMBEDDED as — its prose plus its work logs.
|
||
|
||
A synthesised embed-time shape, like `rule_document` and unlike a lesson:
|
||
the stored record is the task row, and the logs live in their own table, so
|
||
the document that should be searchable exists nowhere until it is built
|
||
here (#4251).
|
||
|
||
WHY THE LOGS BELONG IN THE TASK'S DOCUMENT rather than in rows of their
|
||
own. "Has anyone tried this before?" is answered by a log and asked of a
|
||
task — a hit on a bare log would have to be resolved back to its task to be
|
||
worth anything, so the useful result is the task either way. The objection
|
||
to folding them in is that a long log drowns a short title, and that was
|
||
true before #280: one vector per record meant a 2,000-word log averaged the
|
||
task's own subject away, and everything past ~400 words was truncated
|
||
unread. Chunking removed both. Each log becomes its own section, each
|
||
section its own title-anchored vector, each scored separately — so a task
|
||
is as findable as its best-matching log, and the task's own prose keeps the
|
||
chunk it always had.
|
||
|
||
That the result is LEGIBLE is the other half, and it is this session's
|
||
other build: a search now hands back the chunk that won (#4243), so a hit
|
||
earned by a log shows that log's passage under the task's title. Without it
|
||
the reader would get `body[:240]` of the task — the opening of a record
|
||
whose relevance lives three hundred lines further down.
|
||
|
||
Ordering is oldest-first, matching how the web renders the narrative. Only
|
||
the heading date distinguishes the sections, so it is part of the shape:
|
||
"when was this tried" is half of what a log answers.
|
||
|
||
An entry with no content is skipped rather than emitted as a bare heading —
|
||
an empty section is a vector with nothing in it but the task's title, which
|
||
competes with the task's real chunk and says nothing.
|
||
"""
|
||
sections = []
|
||
for created_at, content in logs:
|
||
text = (content or "").strip()
|
||
if not text:
|
||
continue
|
||
stamp = getattr(created_at, "date", None)
|
||
heading = (
|
||
f"{WORK_LOG_HEADING} — {stamp()}" if callable(stamp)
|
||
else WORK_LOG_HEADING
|
||
)
|
||
sections.append(f"{heading}\n\n{text}")
|
||
if not sections:
|
||
return title, body
|
||
prose = (body or "").strip()
|
||
joined = "\n\n".join(sections)
|
||
return title, f"{prose}\n\n{joined}" if prose else joined
|
||
|
||
|
||
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 (
|
||
trigger_title(name, 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.
|
||
"""
|
||
from scribe.models.rulebook import Rule # runtime import: see TYPE_CHECKING above
|
||
|
||
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:
|
||
if not await _claim_parent_row(session, Rule.id, rule_id, "rule"):
|
||
return
|
||
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,
|
||
kind: str | None = None,
|
||
report: dict | None = None,
|
||
*,
|
||
project_id: int | None = None,
|
||
everywhere: bool = False,
|
||
) -> list[tuple[float, "Rule"]]:
|
||
"""Return up to *limit* (score, rule) pairs most relevant to *query*.
|
||
|
||
Pass `report` (an empty dict) to learn what the threshold turned away:
|
||
the function sets `report["best_available_score"]` to the highest score
|
||
anything reached, or None when the corpus offered nothing at all. It is
|
||
the only figure that survives a call returning nothing, and therefore the
|
||
only one a bar can be judged from (#3670).
|
||
|
||
It also sets `report["searched"]`: False before anything can return, True
|
||
only where a real result set exists. So an empty query, an unavailable
|
||
embedder and a failed database query all leave it FALSE, and a caller can
|
||
tell a search that found nothing from one that never ran. A caller logging
|
||
telemetry must check it — recording a failed search as a zero-result call
|
||
reports a decline the ranker never made (#3765). ABSENT means no search
|
||
touched the dict at all, which is a stand-in in a test, not a real call.
|
||
|
||
SCOPED, and the scope is a rule's home (milestone 414). A rule lives in a
|
||
rulebook topic — GLOBAL, it applies wherever its owner works — or on one
|
||
project, where it applies to that project and nowhere else:
|
||
|
||
- default (`project_id=None`): global rules only. A hook with no bound
|
||
project gets these, and so does any caller that forgets to say; the
|
||
safe failure is surfacing less, not another project's rules.
|
||
- `project_id=N`: global rules plus project N's own, and N's only when
|
||
the caller can read that project (access.can_read_project, so a shared
|
||
project's rules reach its collaborators too).
|
||
- `everywhere=True`: every rule the caller owns, in any home. Only for an
|
||
explicit whole-rulebook question — `search(content_type="rule")` with no
|
||
project — where "is there a rule about this" is asked across everything.
|
||
|
||
This used to be scoped by OWNERSHIP alone, on the argument that "is there
|
||
a rule about this" wants the whole rulebook. That is still right for the
|
||
explicit ask. It was wrong for the hooks, which inject unasked: every
|
||
project's rules surfaced in every other project's sessions — one repo's
|
||
template conventions arriving while editing an unrelated one — and a
|
||
project rule meant nothing a session could feel.
|
||
|
||
THERE IS NO TIER TO NARROW BY ANY MORE (milestone 394). This carried a
|
||
`tier` parameter, and the arms deliberately passed nothing: filtering on it
|
||
made a whole class of rules permanently ineligible for the one mechanism
|
||
that surfaces a rule AT the moment it applies. The tier is now gone
|
||
entirely, so every rule is eligible for every arm and relevance is the
|
||
threshold's job alone — see the block above RULEHINT_LIMIT in
|
||
services/plugin_context.py for what those scores are read against.
|
||
|
||
`kind` narrows to `rule` or `preference`, and NONE is likewise the ordinary
|
||
case: a caller asking "what governs this" wants both, because the reader
|
||
needs to know what binds AND how the operator wants it done. The one place
|
||
it is passed is a RESERVED SLOT — a query that may only return a
|
||
preference, so the slot cannot be spent on something else. That is the
|
||
same reason `note_type` exists on the sibling search, and the same failure
|
||
it prevents: a slot silently filled by the wrong kind is worse than no
|
||
slot, because the line is indistinguishable from one that earned its place
|
||
on score.
|
||
|
||
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
|
||
|
||
# See the sibling search: stamped before anything can return (#3765).
|
||
if report is not None:
|
||
report["searched"] = False
|
||
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 []
|
||
|
||
distance = RuleEmbedding.embedding.cosine_distance(query_vec)
|
||
|
||
try:
|
||
# topic_id XOR project_id (migration 0059), so a rule matches exactly
|
||
# one arm of whichever clause applies. Inside the try: the access
|
||
# check reads the database too, and this function fails open.
|
||
global_rule = Rulebook.owner_user_id == user_id
|
||
if everywhere:
|
||
home = or_(global_rule, Project.user_id == user_id)
|
||
elif project_id and await can_read_project(user_id, project_id):
|
||
home = or_(global_rule, Rule.project_id == project_id)
|
||
else:
|
||
home = global_rule
|
||
|
||
async with async_session() as session:
|
||
rows = (await session.execute(
|
||
select(
|
||
Rule,
|
||
distance.label("distance"),
|
||
RuleEmbedding.chunk_index,
|
||
RuleEmbedding.chunk_text,
|
||
)
|
||
.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),
|
||
# No threshold predicate — see the note above
|
||
# semantic_search_notes. Applied below, after the collapse.
|
||
home,
|
||
*( [Rule.kind == kind] if kind else [] ),
|
||
)
|
||
# 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]] = {}
|
||
# Which chunk won, kept beside the score it won with — a rule's `why` and
|
||
# `how_to_apply` are long, and a caller shown only the head cannot see the
|
||
# clause that actually matched (#4243).
|
||
won: dict[int, dict] = {}
|
||
for rule, dist, chunk_index, chunk_text in rows:
|
||
score = 1.0 - float(dist)
|
||
if rule.id not in best or score > best[rule.id][0]:
|
||
best[rule.id] = (score, rule)
|
||
won[int(rule.id)] = {
|
||
"index": int(chunk_index), "text": chunk_text or "",
|
||
}
|
||
ranked = sorted(best.values(), key=lambda pair: pair[0], reverse=True)
|
||
kept = [pair for pair in ranked if pair[0] >= threshold][:limit]
|
||
record_best_chunk(report, {
|
||
int(r.id): won[int(r.id)] for _s, r in kept if int(r.id) in won
|
||
})
|
||
if report is not None:
|
||
# See the sibling search: absent means the search never ran (#3765).
|
||
report["searched"] = True
|
||
# One unpack — see the sibling search (#3807).
|
||
best = ranked[0] if ranked else None
|
||
report["best_available_score"] = best[0] if best else None
|
||
report["best_available_id"] = int(best[1].id) if best else None
|
||
return kept
|
||
|
||
|
||
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
|
||
)
|
||
# IDS ONLY — the text is re-read per rule below (#4262).
|
||
by_version = {
|
||
row[0] for row in (await session.execute(
|
||
select(Rule.id)
|
||
.where(Rule.deleted_at.is_(None), Rule.id.notin_(current))
|
||
)).fetchall()
|
||
}
|
||
by_time = await _vectors_older_than_their_record(
|
||
session, RuleEmbedding.rule_id, RuleEmbedding.updated_at,
|
||
Rule.id, Rule.updated_at,
|
||
)
|
||
stale = sorted(by_version | by_time)
|
||
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) (%d with stale vectors)",
|
||
len(stale), len(by_time),
|
||
)
|
||
for rule_id in stale:
|
||
row = await _current_row(
|
||
(Rule.title, Rule.statement, Rule.when_to_apply), Rule.id, rule_id
|
||
)
|
||
if row is None:
|
||
continue
|
||
await upsert_rule_embedding(rule_id, *row)
|
||
|
||
|
||
# ── 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,
|
||
report: dict | None = None,
|
||
) -> 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"),
|
||
MilestoneEmbedding.chunk_index,
|
||
MilestoneEmbedding.chunk_text,
|
||
)
|
||
.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]] = {}
|
||
# A milestone's `body` IS the plan, and search results show its short
|
||
# `description` — so a match on the design was previewed by a sentence that
|
||
# need not mention it. The winning chunk is what the caller should see.
|
||
won: dict[int, dict] = {}
|
||
for milestone, dist, chunk_index, chunk_text in rows:
|
||
score = 1.0 - float(dist)
|
||
if milestone.id not in best or score > best[milestone.id][0]:
|
||
best[milestone.id] = (score, milestone)
|
||
won[int(milestone.id)] = {
|
||
"index": int(chunk_index), "text": chunk_text or "",
|
||
}
|
||
ranked = sorted(best.values(), key=lambda pair: pair[0], reverse=True)
|
||
kept = [pair for pair in ranked if pair[0] >= threshold][:limit]
|
||
record_best_chunk(report, {
|
||
int(m.id): won[int(m.id)] for _s, m in kept if int(m.id) in won
|
||
})
|
||
return kept
|
||
|
||
|
||
# --- Systems: the charter as an ANSWER, not a filter (#4251) -----------------
|
||
|
||
|
||
def system_document(
|
||
name: str | None, description: str | None
|
||
) -> tuple[str | None, str | None]:
|
||
"""The (title, body) a System is EMBEDDED as — its name and its charter.
|
||
|
||
The plainest of the four document shapes, and deliberately so. A System's
|
||
`description` is already written as the thing this search has to match: it
|
||
says what belongs in the area and what does not, which is the answer to
|
||
"where does this go?" in the words someone asking would use. There is no
|
||
trigger to synthesise, as `rule_document` must, and no separate record to
|
||
gather in, as `task_document` must — the stored charter IS the sharp
|
||
document, the way a snippet's is.
|
||
|
||
`color`, `status` and `order_index` are left out. They are presentation and
|
||
bookkeeping; a vector that carried them would be answering a question
|
||
nobody asks of a charter.
|
||
|
||
A System with no charter yet degrades to its name. It still embeds, just
|
||
weakly — a bare name is exactly what the model docstring says is never
|
||
enough, and this is an argument for writing the charter, not for padding
|
||
the document with whatever is to hand.
|
||
"""
|
||
return (name or "").strip() or None, (description or "").strip() or None
|
||
|
||
|
||
async def upsert_system_embedding(
|
||
system_id: int, name: str | None, description: str | None
|
||
) -> None:
|
||
"""Chunk, embed and persist a System's vectors. Safe to fire-and-forget.
|
||
|
||
The third sibling's contract exactly: one document definition shared by the
|
||
write path and the backfill, and an atomic per-System replacement guarded
|
||
by the parent-row claim (#3262), so a System deleted mid-refresh wins.
|
||
"""
|
||
from scribe.models.embedding import SystemEmbedding
|
||
from scribe.models.system import System
|
||
|
||
doc_title, doc_body = system_document(name, description)
|
||
chunks = chunk_document(doc_title, doc_body)
|
||
try:
|
||
if not chunks:
|
||
async with async_session() as session:
|
||
await session.execute(
|
||
delete(SystemEmbedding).where(SystemEmbedding.system_id == system_id)
|
||
)
|
||
await session.commit()
|
||
return
|
||
except Exception:
|
||
logger.warning("Failed to clear embedding for system %d", system_id, exc_info=True)
|
||
return
|
||
|
||
try:
|
||
vectors = await get_embeddings(chunks)
|
||
except Exception:
|
||
logger.debug("Skipping embedding for system %d — embedder unavailable", system_id)
|
||
return
|
||
|
||
try:
|
||
async with async_session() as session:
|
||
if not await _claim_parent_row(session, System.id, system_id, "system"):
|
||
return
|
||
await session.execute(
|
||
delete(SystemEmbedding).where(SystemEmbedding.system_id == system_id)
|
||
)
|
||
for index, (chunk, vector) in enumerate(zip(chunks, vectors)):
|
||
session.add(SystemEmbedding(
|
||
system_id=system_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 system %d", system_id, exc_info=True)
|
||
|
||
|
||
async def semantic_search_systems(
|
||
user_id: int,
|
||
query: str,
|
||
*,
|
||
project_id: int | None = None,
|
||
limit: int = 5,
|
||
threshold: float = _SIMILARITY_THRESHOLD,
|
||
report: dict | None = None,
|
||
) -> list[tuple[float, "System"]]:
|
||
"""Return up to *limit* (score, system) pairs most like *query*.
|
||
|
||
Answers "where does this belong?" — the question asked before filing a
|
||
record or opening a file, and the one that had no tool: `list_systems`
|
||
enumerates and `system_id` filters, so a charter could narrow a search and
|
||
could never be the answer to one.
|
||
|
||
ITS OWN SEARCH rather than a note kind, for what note 3163 says about
|
||
milestones: a charter competing with the whole note corpus for one top-k
|
||
would be outranked by the records filed under it, and the right answer
|
||
would be crowded out by its own contents. The questions are different too —
|
||
"where does this belong?" is not "what prior art is there?" — and a caller
|
||
asking one should not have to read past answers to the other.
|
||
|
||
SCOPE. With `project_id`, that project's Systems, provided the caller can
|
||
read the project (access.can_read_project, rule 78) — a collaborator on a
|
||
shared project sees its areas, which is the point of a charter. Without
|
||
one, the Systems the caller owns across their projects. Archived Systems
|
||
are excluded: an archived area is one the operator has said is no longer
|
||
where things go, and that is exactly the question being asked.
|
||
|
||
Collapses to best-chunk-per-System and publishes the winning chunk, like
|
||
the sibling searches — from the start rather than retrofitted (#4243).
|
||
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 SystemEmbedding
|
||
from scribe.models.system import System
|
||
|
||
if not query or not query.strip():
|
||
return []
|
||
try:
|
||
query_vec = await get_embedding(query)
|
||
except Exception:
|
||
logger.debug("System search skipped — embedder unavailable")
|
||
return []
|
||
|
||
distance = SystemEmbedding.embedding.cosine_distance(query_vec)
|
||
try:
|
||
if project_id:
|
||
if not await can_read_project(user_id, project_id):
|
||
return []
|
||
scope = System.project_id == project_id
|
||
else:
|
||
scope = System.user_id == user_id
|
||
async with async_session() as session:
|
||
rows = (await session.execute(
|
||
select(
|
||
System,
|
||
distance.label("distance"),
|
||
SystemEmbedding.chunk_index,
|
||
SystemEmbedding.chunk_text,
|
||
)
|
||
.select_from(SystemEmbedding)
|
||
.join(System, SystemEmbedding.system_id == System.id)
|
||
.where(
|
||
scope,
|
||
System.deleted_at.is_(None),
|
||
System.status != "archived",
|
||
)
|
||
.order_by(distance)
|
||
.limit(limit * _CHUNK_OVERFETCH)
|
||
)).all()
|
||
except Exception:
|
||
logger.warning("System semantic search failed", exc_info=True)
|
||
return []
|
||
|
||
best: dict[int, tuple[float, object]] = {}
|
||
# A charter runs to several hundred words and a result shows its NAME — so
|
||
# a match on the paragraph that actually decides where a record belongs
|
||
# would be previewed by two words that cannot. The winning chunk is what
|
||
# the caller should see (#4243).
|
||
won: dict[int, dict] = {}
|
||
for system, dist, chunk_index, chunk_text in rows:
|
||
score = 1.0 - float(dist)
|
||
if system.id not in best or score > best[system.id][0]:
|
||
best[system.id] = (score, system)
|
||
won[int(system.id)] = {
|
||
"index": int(chunk_index), "text": chunk_text or "",
|
||
}
|
||
ranked = sorted(best.values(), key=lambda pair: pair[0], reverse=True)
|
||
kept = [pair for pair in ranked if pair[0] >= threshold][:limit]
|
||
record_best_chunk(report, {
|
||
int(sys_.id): won[int(sys_.id)] for _s, sys_ in kept if int(sys_.id) in won
|
||
})
|
||
return kept
|
||
|
||
|
||
async def backfill_system_embeddings() -> None:
|
||
"""Embed Systems that have no current vectors. Runs at startup beside the
|
||
note, rule and milestone backfills; a CHUNKER_VERSION bump re-embeds.
|
||
|
||
This is also the pass that makes every existing charter findable at all —
|
||
Systems got vectors in #4251, so before it runs there are none."""
|
||
from scribe.models.embedding import SystemEmbedding
|
||
from scribe.models.system import System
|
||
|
||
try:
|
||
async with async_session() as session:
|
||
current = select(SystemEmbedding.system_id).where(
|
||
SystemEmbedding.chunker_version == CHUNKER_VERSION
|
||
)
|
||
# IDS ONLY — the charter is re-read per System below (#4262).
|
||
by_version = {
|
||
row[0] for row in (await session.execute(
|
||
select(System.id)
|
||
.where(System.deleted_at.is_(None), System.id.notin_(current))
|
||
)).fetchall()
|
||
}
|
||
by_time = await _vectors_older_than_their_record(
|
||
session, SystemEmbedding.system_id, SystemEmbedding.updated_at,
|
||
System.id, System.updated_at,
|
||
)
|
||
stale = sorted(by_version | by_time)
|
||
except Exception:
|
||
logger.warning("System embedding backfill: failed to query systems", exc_info=True)
|
||
return
|
||
|
||
if not stale:
|
||
logger.info("System embedding backfill: all systems current at chunker v%d", CHUNKER_VERSION)
|
||
return
|
||
logger.info(
|
||
"System embedding backfill: embedding %d system(s) (%d with stale vectors)",
|
||
len(stale), len(by_time),
|
||
)
|
||
for system_id in stale:
|
||
row = await _current_row(
|
||
(System.name, System.description), System.id, system_id
|
||
)
|
||
if row is None:
|
||
continue
|
||
await upsert_system_embedding(system_id, *row)
|
||
|
||
|
||
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
|
||
)
|
||
# IDS ONLY — the plan is re-read per milestone below (#4262).
|
||
by_version = {
|
||
row[0] for row in (await session.execute(
|
||
select(Milestone.id)
|
||
.where(Milestone.deleted_at.is_(None), Milestone.id.notin_(current))
|
||
)).fetchall()
|
||
}
|
||
by_time = await _vectors_older_than_their_record(
|
||
session, MilestoneEmbedding.milestone_id, MilestoneEmbedding.updated_at,
|
||
Milestone.id, Milestone.updated_at,
|
||
)
|
||
stale = sorted(by_version | by_time)
|
||
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) (%d with stale vectors)",
|
||
len(stale), len(by_time),
|
||
)
|
||
for milestone_id in stale:
|
||
row = await _current_row(
|
||
(Milestone.title, Milestone.description, Milestone.body),
|
||
Milestone.id, milestone_id,
|
||
)
|
||
if row is None:
|
||
continue
|
||
await upsert_milestone_embedding(milestone_id, *row)
|