bvandeusenandClaude Opus 5
21a701c572
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 15s
CI & Build / TypeScript typecheck (push) Successful in 54s
CI & Build / integration (push) Successful in 55s
CI & Build / Python tests (push) Successful in 1m38s
CI & Build / Build & push image (push) Successful in 36s
Measured live after deploy: three plans reworded from existing active milestones scored 0.83-0.87 against the milestone they restated, and the nearest distinct plans 0.72-0.77. At the inherited 0.90 the semantic arm matched nothing, so only an identical title was caught. 0.80 sits in the gap. The Settings form's default moves with it (test_settings_defaults_agree). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy
903 lines
40 KiB
Python
903 lines
40 KiB
Python
"""Write-time near-duplicate detection — the update-over-create gate.
|
|
|
|
Goal: stop a second near-identical row from being created when an existing one
|
|
should be UPDATED instead. Duplicates bloat the store and, worse, get surfaced
|
|
by semantic search (RAG) later as competing/stale copies that then have to be
|
|
reconciled. This is the enforcement half of the instruction-level "prefer
|
|
updating over creating" reflex.
|
|
|
|
OPT-IN by design: the interactive create paths (MCP create tools + REST create
|
|
routes) run this gate; internal/programmatic creates do NOT (e.g. a recurring
|
|
task spawning its next instance, or a bulk import — those legitimately repeat a
|
|
title and must not be blocked). Callers that want the gate call find_duplicate_*
|
|
themselves and act on a hit; nothing here mutates.
|
|
|
|
Two signals, both scoped to the same owner + project + kind:
|
|
1. Normalized-title exact match — cheap, always checked.
|
|
2. Semantic similarity (cosine ≥ _SEMANTIC_THRESHOLD) — only when the incoming
|
|
body is substantial. Short/title-only embeddings sit in a tight neighborhood
|
|
and false-positive (the pre-pivot lesson: "Lore: Shell 0" vs
|
|
"Lore: Reinitialization 0" matched at 0.91 with no body), so we gate it on
|
|
a minimum body length.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import aliased
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.embedding import NoteEmbedding
|
|
from scribe.models.note import Note
|
|
from scribe.models.rulebook import Rule
|
|
from scribe.models.base import iso
|
|
from scribe.services.access import can_read_project
|
|
from scribe.services import embeddings as embeddings_svc
|
|
# Imported rather than redeclared: no service imports this module (the create
|
|
# gate is called from the routes/tools layer), so there is no cycle to dodge,
|
|
# and a second copy of the constant is a thing to drift.
|
|
from scribe.services.snippets import SNIPPET_NOTE_TYPE
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Run the semantic check only when the incoming body has at least this many
|
|
# characters — below it, embeddings are dominated by the title and false-positive.
|
|
_MIN_BODY_FOR_SEMANTIC = 200
|
|
# Cosine threshold for "this is the same thing, reworded." Deliberately high to
|
|
# keep false positives rare (a hard block with a force-override is unforgiving of
|
|
# noise). Matches the 0.90 the pre-pivot dedup settled on.
|
|
_SEMANTIC_THRESHOLD = 0.90
|
|
|
|
# SNIPPETS ARE MEASURED DIFFERENTLY, and #2518 is why. A snippet's embedded
|
|
# document is mostly PROSE ABOUT the code — name, when-to-reach-for-it,
|
|
# signature, the comments explaining the choice — with the artefact itself a
|
|
# minority of the text. Two measurements on the same corpus:
|
|
#
|
|
# .btn-danger vs .btn-danger-outline 0.92 siblings, blocked (false positive)
|
|
# .btn-primary re-recorded verbatim
|
|
# under a different name <0.90 a literal copy, ALLOWED THROUGH
|
|
#
|
|
# The second is the one that settles it. Identical code at an identical
|
|
# repo·path·symbol sailed past the gate because the description differed, while
|
|
# two deliberately-parallel variants were refused because theirs did not. The
|
|
# arm is not mis-tuned; it is reading the wrong field, and no threshold fixes
|
|
# that — lowering it blocks more siblings, raising it allows more copies.
|
|
#
|
|
# So: STRUCTURE decides, and the semantic arm becomes a backstop set above the
|
|
# band where legitimate variants live (0.92 observed). It still catches a
|
|
# genuine reword that shares neither location nor code, which is the case the
|
|
# structural signals cannot see.
|
|
_SNIPPET_SEMANTIC_THRESHOLD = 0.96
|
|
|
|
# The gate queries per CHUNK of the candidate (#280) — this caps how many
|
|
# searches one save may cost. Eight chunks ≈ five thousand words of candidate;
|
|
# a duplicate hiding past that is the duplicate report's job to find, not a
|
|
# reason to stall the write path.
|
|
_GATE_MAX_CHUNKS = 8
|
|
|
|
|
|
@dataclass
|
|
class DuplicateMatch:
|
|
"""An existing record judged a near-duplicate of an incoming create."""
|
|
id: int
|
|
title: str
|
|
similarity: float # 1.0 for an exact normalized-title match
|
|
reason: str # "title" | "semantic"
|
|
|
|
|
|
# How each signal describes itself when it blocks a write. The structural ones
|
|
# are CERTAIN, so they say what was matched instead of hedging with "similar" —
|
|
# and they point at merge, not update, because two records of one artefact is
|
|
# what merge exists to fold back together.
|
|
_REASON_PHRASING = {
|
|
"location": (
|
|
"is already recorded at that exact repo · path · symbol",
|
|
"Update it (update_{kind}), or if you meant to record a second call "
|
|
"site, use merge_snippets so one record carries both locations.",
|
|
),
|
|
"code": (
|
|
"already holds identical code",
|
|
"Update it (update_{kind}) rather than keeping two copies that must "
|
|
"then be kept in step by hand.",
|
|
),
|
|
}
|
|
|
|
|
|
def duplicate_response(dup: "DuplicateMatch", kind: str) -> dict:
|
|
"""Standard 'blocked — update instead' payload returned by a create tool
|
|
when the gate finds a near-duplicate. `kind` is 'note', 'task' or 'snippet'
|
|
(drives the update_<kind> hint)."""
|
|
phrasing = _REASON_PHRASING.get(dup.reason)
|
|
if phrasing:
|
|
claim, advice = phrasing
|
|
message = (
|
|
f'An existing {kind} (id {dup.id}: "{dup.title}") {claim}. '
|
|
f"{advice.format(kind=kind)} If this really is a distinct {kind}, "
|
|
f"retry with force=true."
|
|
)
|
|
else:
|
|
message = (
|
|
f'A {dup.reason}-similar {kind} already exists (id {dup.id}: '
|
|
f'"{dup.title}"). Prefer UPDATING it (update_{kind}) over creating a '
|
|
f"near-duplicate. If this really is a distinct {kind}, retry with "
|
|
f"force=true."
|
|
)
|
|
return {
|
|
"duplicate": True,
|
|
"existing_id": dup.id,
|
|
"existing_title": dup.title,
|
|
"similarity": dup.similarity,
|
|
"match": dup.reason,
|
|
"message": message,
|
|
}
|
|
|
|
|
|
async def _find_snippet_by_structure(
|
|
user_id: int,
|
|
code: str,
|
|
locations: list[dict] | None,
|
|
project_id: int | None,
|
|
) -> DuplicateMatch | None:
|
|
"""Exact-identity duplicate of an incoming snippet, or None.
|
|
|
|
Two signals, both index-served off `notes.data` and both CERTAIN rather than
|
|
probabilistic — which is what the semantic arm could not be (#2518):
|
|
|
|
location the same named thing in the same file. Requires BOTH path and
|
|
symbol: a path alone is a directory of many artefacts, and
|
|
matching on it would refuse every second snippet from one file.
|
|
code byte-identical code, wherever it lives. Uses the same
|
|
fingerprint the drift check uses, so "identical" means the same
|
|
thing in both places.
|
|
|
|
Fail-open like the rest of this module: a failed lookup lets the write
|
|
through rather than blocking on an infrastructure problem.
|
|
"""
|
|
from scribe.services.knowledge import location_jsonpath
|
|
from scribe.services.snippets import code_sha
|
|
|
|
identifying = [
|
|
loc for loc in (locations or [])
|
|
if (loc.get("path") or "").strip() and (loc.get("symbol") or "").strip()
|
|
]
|
|
if not identifying and not (code or "").strip():
|
|
return None # nothing to match on — don't open a session for it
|
|
|
|
def _scoped(stmt):
|
|
stmt = stmt.where(
|
|
Note.user_id == user_id,
|
|
Note.deleted_at.is_(None),
|
|
Note.note_type == SNIPPET_NOTE_TYPE,
|
|
)
|
|
# Same scoping rule as the title and semantic arms: a project's records
|
|
# compare only within that project, orphans only to orphans.
|
|
return (stmt.where(Note.project_id == project_id) if project_id is not None
|
|
else stmt.where(Note.project_id.is_(None)))
|
|
|
|
try:
|
|
async with async_session() as session:
|
|
for loc in identifying:
|
|
parts = {
|
|
"path": (loc["path"]).strip(),
|
|
"symbol": (loc["symbol"]).strip(),
|
|
}
|
|
repo = (loc.get("repo") or "").strip()
|
|
if repo:
|
|
parts["repo"] = repo
|
|
stmt = _scoped(select(Note)).where(
|
|
Note.data.path_exists(location_jsonpath(parts))
|
|
)
|
|
hit = (await session.execute(stmt.limit(1))).scalars().first()
|
|
if hit is not None:
|
|
return DuplicateMatch(hit.id, hit.title, 1.0, "location")
|
|
|
|
if (code or "").strip():
|
|
stmt = _scoped(select(Note)).where(
|
|
Note.data["code_sha"].astext == code_sha(code)
|
|
)
|
|
hit = (await session.execute(stmt.limit(1))).scalars().first()
|
|
if hit is not None:
|
|
return DuplicateMatch(hit.id, hit.title, 1.0, "code")
|
|
except Exception:
|
|
logger.debug("snippet structural dedup skipped — query failed", exc_info=True)
|
|
return None
|
|
|
|
|
|
async def find_duplicate_note(
|
|
user_id: int,
|
|
title: str,
|
|
body: str = "",
|
|
project_id: int | None = None,
|
|
is_task: bool | None = None,
|
|
note_type: str = "note",
|
|
code: str = "",
|
|
locations: list[dict] | None = None,
|
|
) -> DuplicateMatch | None:
|
|
"""Best near-duplicate of (title, body) within the same owner + project +
|
|
kind, or None. Title match first (cheap, exact), then — for snippets — the
|
|
structural signals, then semantic when the body is long enough to be
|
|
meaningful. Never raises — embedder failure degrades to title-only (callers
|
|
should still be able to create).
|
|
|
|
`code` and `locations` are the snippet's structured fields. They are ignored
|
|
for every other kind, and passing them is what lets the gate compare
|
|
ARTEFACTS rather than descriptions of artefacts (#2518).
|
|
"""
|
|
norm = " ".join((title or "").split()).lower()
|
|
|
|
# --- Signal 1: normalized-title exact match (same scope) ---
|
|
# Fail-open: a dedup-check failure (DB down, etc.) must never block a
|
|
# legitimate create — degrade to "no duplicate found" and let it through.
|
|
if norm:
|
|
try:
|
|
async with async_session() as session:
|
|
stmt = select(Note).where(
|
|
Note.user_id == user_id,
|
|
Note.deleted_at.is_(None),
|
|
Note.note_type == note_type,
|
|
func.lower(func.trim(Note.title)) == norm,
|
|
)
|
|
if project_id is not None:
|
|
stmt = stmt.where(Note.project_id == project_id)
|
|
else:
|
|
stmt = stmt.where(Note.project_id.is_(None))
|
|
if is_task is True:
|
|
stmt = stmt.where(Note.status.isnot(None))
|
|
elif is_task is False:
|
|
stmt = stmt.where(Note.status.is_(None))
|
|
existing = (await session.execute(stmt.limit(1))).scalars().first()
|
|
if existing is not None:
|
|
return DuplicateMatch(existing.id, existing.title, 1.0, "title")
|
|
except Exception:
|
|
logger.debug("dedup title check skipped — query failed", exc_info=True)
|
|
return None
|
|
|
|
# --- Signal 2: structural identity (snippets only) ---
|
|
# Ahead of the semantic arm because it is exact: when it fires there is
|
|
# nothing to weigh, and its verdict is the one worth showing.
|
|
if note_type == SNIPPET_NOTE_TYPE:
|
|
structural = await _find_snippet_by_structure(
|
|
user_id, code, locations, project_id
|
|
)
|
|
if structural is not None:
|
|
return structural
|
|
|
|
# --- Signal 3: semantic similarity (only with a substantial body) ---
|
|
if body and len(body.strip()) >= _MIN_BODY_FOR_SEMANTIC:
|
|
# Query with the SAME chunker the corpus was embedded with (#280). This
|
|
# was the copy that mattered most and was easiest to miss: these are
|
|
# QUERY documents, compared against embedded ones — shaped differently
|
|
# from the corpus, the gate degrades silently. Chunking also makes the
|
|
# gate see what the whole-document query diluted: a long candidate that
|
|
# duplicates an existing record IN ONE SECTION now matches on that
|
|
# section. Capped so one pathological paste can't turn a save into
|
|
# dozens of searches — a duplicate past the cap is the duplicate
|
|
# report's job, not the gate's.
|
|
for query in embeddings_svc.chunk_document(title, body)[:_GATE_MAX_CHUNKS]:
|
|
# Scope the semantic check the same way as the title check: a record
|
|
# in project P compares only to P; a project-less (orphan) record
|
|
# compares only to other orphans (orphan_only), NOT across every
|
|
# project — without this, semantic_search_notes applies no project
|
|
# filter when project_id is None and would match an orphan note
|
|
# against any project's notes.
|
|
hits = await embeddings_svc.semantic_search_notes(
|
|
user_id, query, project_id=project_id, is_task=is_task,
|
|
orphan_only=(project_id is None),
|
|
limit=3,
|
|
threshold=(_SNIPPET_SEMANTIC_THRESHOLD
|
|
if note_type == SNIPPET_NOTE_TYPE else _SEMANTIC_THRESHOLD),
|
|
# Owner-only, deliberately: this gate BLOCKS a create and tells
|
|
# the caller to update the match instead. Matching someone
|
|
# else's record would refuse their write and point them at
|
|
# something they may not be able to edit.
|
|
scope="own",
|
|
# NOT demoted by supersession (#278). A superseded record is
|
|
# still a duplicate of what you are about to write — the claim
|
|
# is that it is no longer CURRENT, not that it is gone. Demoting
|
|
# it here would let the same note be recorded a second time, and
|
|
# the second copy would be the one nothing warns about.
|
|
demote_superseded=False,
|
|
)
|
|
for score, note in hits:
|
|
# semantic_search_notes doesn't filter note_type — enforce it
|
|
# here so a note doesn't shadow a task of the same wording, etc.
|
|
if note.note_type == note_type:
|
|
return DuplicateMatch(note.id, note.title, round(score, 3), "semantic")
|
|
|
|
return None
|
|
|
|
|
|
# --- corpus-wide near-duplicate report (#2088) -------------------------------
|
|
# The gate above PREVENTS a new duplicate; merge_snippets CURES one you point it
|
|
# at. Neither FINDS the duplicates already sitting in the record — someone had to
|
|
# notice them by hand, which is the exact failure the Drafter exists to remove.
|
|
#
|
|
# WHY OWN SNIPPETS ONLY. merge_snippets requires every record to share the
|
|
# target's owner (cross-owner merge is out of scope), so a report that surfaced
|
|
# someone else's snippet would propose a merge that cannot be performed. The
|
|
# scope here is set by what the operator can actually act on, not by what they
|
|
# can see.
|
|
#
|
|
# WHY THE FLOOR IS PER-KIND. Chunked embeddings (#280) changed what a pair
|
|
# score MEANS for multi-chunk records: a note-pair's similarity is its closest
|
|
# chunk pair, so any family of related long records — a dev-log run, a research
|
|
# fan-out — clears a floor that whole-document vectors used to dilute below it.
|
|
# Measured on the live corpus (2026-08-09): at 0.82 the note/task reports
|
|
# saturate the pair cap with related-but-distinct families, while 0.93+ returns
|
|
# the genuinely-alike records. Snippets are single-chunk (short by nature), so
|
|
# their similarity scale never shifted and they keep the old floor.
|
|
#
|
|
# Snippets sit BELOW the 0.90 write gate — the report catches what the gate
|
|
# lets through. Notes/tasks sit ABOVE it, and that is not a contradiction: the
|
|
# gate compares a new record against best-matching chunks too, but it blocks a
|
|
# WRITE and must stay forgiving, while the report proposes a REVIEW and at
|
|
# chunk grain 0.90 would still drown it in families. Each is a setting rather
|
|
# than a constant (rule #25) because the right value depends on how uniform a
|
|
# corpus is, and nobody can guess that from here.
|
|
|
|
DUPLICATE_THRESHOLD_KEYS = {
|
|
"snippet": "kb_duplicate_threshold_snippet",
|
|
"note": "kb_duplicate_threshold_note",
|
|
"task": "kb_duplicate_threshold_task",
|
|
}
|
|
DUPLICATE_DEFAULT_THRESHOLDS = {"snippet": 0.82, "note": 0.93, "task": 0.93}
|
|
# Hard cap on returned pairs. A pathologically uniform corpus is O(n²) pairs, and
|
|
# a report nobody can read is not a report.
|
|
_MAX_DUPLICATE_PAIRS = 200
|
|
|
|
|
|
async def get_duplicate_threshold(user_id: int, kind: str = "snippet") -> float:
|
|
"""The user's near-duplicate similarity floor for `kind`, clamped to [0, 1]."""
|
|
from scribe.services.settings import get_setting
|
|
|
|
default = DUPLICATE_DEFAULT_THRESHOLDS[kind]
|
|
try:
|
|
value = float(await get_setting(
|
|
user_id, DUPLICATE_THRESHOLD_KEYS[kind], str(default)
|
|
))
|
|
except (TypeError, ValueError):
|
|
value = default
|
|
return min(1.0, max(0.0, value))
|
|
|
|
|
|
def group_pairs(pairs: list[tuple[int, int, float]]) -> list[list[int]]:
|
|
"""Collapse similar-pairs into candidate merge SETS (connected components).
|
|
|
|
Pure and synchronous so the grouping rule is testable without a database.
|
|
|
|
Transitive on purpose: if A~B and B~C, all three land in one set even when
|
|
A and C fall below the threshold. That matches what merge does — it folds
|
|
every source into one survivor — and it avoids handing the operator three
|
|
overlapping pairs to reconcile by hand, which is the chore being removed.
|
|
The cost is that a chain of mild resemblances can rope in a pair that isn't
|
|
really alike; the operator sees the members and picks, so a set is a
|
|
proposal, never an action.
|
|
"""
|
|
parent: dict[int, int] = {}
|
|
|
|
def find(x: int) -> int:
|
|
parent.setdefault(x, x)
|
|
while parent[x] != x:
|
|
parent[x] = parent[parent[x]]
|
|
x = parent[x]
|
|
return x
|
|
|
|
def union(a: int, b: int) -> None:
|
|
ra, rb = find(a), find(b)
|
|
if ra != rb:
|
|
parent[rb] = ra
|
|
|
|
for left, right, _score in pairs:
|
|
union(left, right)
|
|
|
|
groups: dict[int, list[int]] = {}
|
|
for node in parent:
|
|
groups.setdefault(find(node), []).append(node)
|
|
# Biggest clusters first — the most tangled thing is the most worth fixing.
|
|
# Ids ascending within a set so the output is stable across runs.
|
|
return sorted((sorted(g) for g in groups.values() if len(g) > 1),
|
|
key=lambda g: (-len(g), g[0]))
|
|
|
|
|
|
def _symbols(data: dict | None) -> set[str]:
|
|
"""Every symbol a snippet claims, from its indexed location mirror."""
|
|
return {
|
|
(loc.get("symbol") or "").strip()
|
|
for loc in (data or {}).get("locations") or []
|
|
if (loc.get("symbol") or "").strip()
|
|
}
|
|
|
|
|
|
def _drop_sibling_pairs(
|
|
pairs: list[tuple[int, int, float]], records: dict[int, dict]
|
|
) -> list[tuple[int, int, float]]:
|
|
"""Remove pairs that are VARIANTS of one thing rather than copies of it.
|
|
|
|
Two snippets that both name a symbol, name DIFFERENT symbols, and hold
|
|
different code are two artefacts. The author asserted that by naming them
|
|
apart, and merging them would destroy a distinction someone made on purpose.
|
|
|
|
Without this, a design system's button family reports as a single merge set:
|
|
eight recipes, every direct pair over the floor, top score 0.92 (#2518).
|
|
They resemble each other because variants of one component are SUPPOSED to —
|
|
same selector prefix, same token families, deliberately parallel prose. The
|
|
similarity is read correctly; it just does not mean "duplicate".
|
|
|
|
THE COST, stated plainly: a helper genuinely recorded twice under two names
|
|
— `debounce` and `useDebouncedRef` — is no longer reported. That is real
|
|
recall lost. It is the better trade because the report is a merge PROPOSAL:
|
|
a missed pair costs a duplicate nobody was going to notice anyway, while a
|
|
wrong set invites an operator to collapse a component family in one click.
|
|
Same-symbol and no-symbol duplicates, which is how re-recording usually
|
|
looks, still report.
|
|
"""
|
|
kept = []
|
|
for left, right, score in pairs:
|
|
left_data, right_data = records.get(left) or {}, records.get(right) or {}
|
|
left_syms, right_syms = _symbols(left_data), _symbols(right_data)
|
|
shas = (left_data.get("code_sha"), right_data.get("code_sha"))
|
|
identical_code = shas[0] is not None and shas[0] == shas[1]
|
|
# Both named, no name in common, and the code differs → siblings.
|
|
if (left_syms and right_syms and not (left_syms & right_syms)
|
|
and not identical_code):
|
|
continue
|
|
kept.append((left, right, score))
|
|
return kept
|
|
|
|
|
|
# What a duplicate group should become, per kind. Snippets merge losslessly —
|
|
# one helper, every call site folded into the survivor. Notes and tasks do NOT
|
|
# merge: folding two records destroys what each actually said, which is why
|
|
# consolidated_at was dropped rather than built (#2483). The report can detect
|
|
# the cluster and name the options; deciding which applies is the reader's job,
|
|
# because it turns on what the records SAY, not how alike they score (#2547).
|
|
_KIND_SUGGESTION = {
|
|
"snippet": (
|
|
"Same reusable thing recorded more than once → merge_snippets(target, "
|
|
"others); the survivor keeps every call site. Deliberately-parallel "
|
|
"variants (a component family) are siblings — leave them."
|
|
),
|
|
"note": (
|
|
"Read before acting — records this alike are one of three things. A "
|
|
"correction pair (one re-measures or reverses the other): declare "
|
|
"supersedes on the newer, both survive, the older is demoted and "
|
|
"labelled. State smeared across dated records: extract it into the "
|
|
"System's reference note (updated in place) and leave these as "
|
|
"history. Genuinely parallel records: leave them alone. Do NOT merge "
|
|
"notes — folding them destroys what each said."
|
|
),
|
|
"task": (
|
|
"Two tasks this alike usually mean the same work opened twice: keep "
|
|
"the one with the real history, fold anything unique into its body or "
|
|
"a work-log, and cancel the other with a pointer. If one CORRECTS the "
|
|
"other's conclusions, supersession also works for tasks."
|
|
),
|
|
}
|
|
|
|
# kind → the Note-model predicate for BOTH sides of the self-join. Tasks are
|
|
# notes with a status, not a note_type of their own — the same split every
|
|
# list surface makes.
|
|
_REPORT_KINDS = ("snippet", "note", "task")
|
|
|
|
|
|
def _kind_clauses(kind: str, note_alias):
|
|
"""The WHERE terms that make an aliased Note row one `kind` of record."""
|
|
if kind == "snippet":
|
|
return (note_alias.note_type == SNIPPET_NOTE_TYPE,)
|
|
if kind == "task":
|
|
return (note_alias.note_type == "note", note_alias.status.isnot(None))
|
|
# kind == "note": documents only — a task is a note with a status, and
|
|
# mixing them would propose folding a to-do into a write-up.
|
|
return (note_alias.note_type == "note", note_alias.status.is_(None))
|
|
|
|
|
|
async def find_duplicate_records(
|
|
user_id: int,
|
|
*,
|
|
kind: str = "snippet",
|
|
threshold: float | None = None,
|
|
limit: int = _MAX_DUPLICATE_PAIRS,
|
|
) -> dict:
|
|
"""Near-duplicate records of one kind, grouped into candidate sets.
|
|
|
|
One indexed self-join over `note_embeddings` rather than an N² Python scan:
|
|
pgvector's cosine distance is the same operator semantic search uses, so a
|
|
similarity floor is a distance ceiling and the work stays in Postgres.
|
|
|
|
`kind` is "snippet", "note", or "task". The query is the same; what differs
|
|
is the group payload and the SUGGESTION attached to it — merge is only ever
|
|
proposed for snippets (see _KIND_SUGGESTION). Non-snippet groups also carry
|
|
`members` with dates and any supersession claims that already exist inside
|
|
the group, because "is this a correction pair or a chronicle cluster?" is
|
|
answered by reading, and dates plus existing claims are the evidence.
|
|
|
|
Returns {"groups": [...], "pairs": [...], "threshold": f, "suggestion": s}.
|
|
Snippet groups keep their historical shape (`snippets` key) so the existing
|
|
UI and MCP consumers don't churn. Fail-open (an empty report) like the rest
|
|
of this module — a suggestion feature must not break the page it decorates.
|
|
"""
|
|
if kind not in _REPORT_KINDS:
|
|
raise ValueError(f"kind must be one of {_REPORT_KINDS}, not {kind!r}")
|
|
floor = (
|
|
await get_duplicate_threshold(user_id, kind)
|
|
if threshold is None else threshold
|
|
)
|
|
floor = min(1.0, max(0.0, floor))
|
|
max_distance = min(2.0, max(0.0, 1.0 - floor))
|
|
|
|
left = aliased(NoteEmbedding, name="left_emb")
|
|
right = aliased(NoteEmbedding, name="right_emb")
|
|
left_note = aliased(Note, name="left_note")
|
|
right_note = aliased(Note, name="right_note")
|
|
distance = left.embedding.cosine_distance(right.embedding)
|
|
# Chunk grain (#280): a note-pair's similarity is its closest CHUNK pair —
|
|
# two records duplicate each other where their most similar sections do,
|
|
# which is the honest definition when one section of a long note restates
|
|
# another record. GROUP BY collapses the chunk cross-product to one row
|
|
# per note pair.
|
|
best = func.min(distance)
|
|
|
|
pairs: list[tuple[int, int, float]] = []
|
|
try:
|
|
async with async_session() as session:
|
|
stmt = (
|
|
select(left.note_id, right.note_id, best.label("distance"))
|
|
.select_from(left)
|
|
# `<` not `!=`: each unordered pair exactly once, and it drops
|
|
# the self-pairs (including cross-chunk self-pairs, which would
|
|
# otherwise flag every multi-chunk note against itself).
|
|
.join(right, left.note_id < right.note_id)
|
|
.join(left_note, left_note.id == left.note_id)
|
|
.join(right_note, right_note.id == right.note_id)
|
|
.where(
|
|
*_kind_clauses(kind, left_note),
|
|
*_kind_clauses(kind, right_note),
|
|
left_note.deleted_at.is_(None),
|
|
right_note.deleted_at.is_(None),
|
|
# Owner-scoped on both sides — see the note above on why the
|
|
# report is bounded by what merge can actually act on.
|
|
left_note.user_id == user_id,
|
|
right_note.user_id == user_id,
|
|
)
|
|
.group_by(left.note_id, right.note_id)
|
|
.having(best <= max_distance)
|
|
.order_by(best.asc())
|
|
.limit(max(1, limit))
|
|
)
|
|
rows = list((await session.execute(stmt)).all())
|
|
pairs = [(int(a), int(b), round(1.0 - float(d), 4)) for a, b, d in rows]
|
|
except Exception:
|
|
logger.warning("Near-duplicate %s scan failed", kind, exc_info=True)
|
|
return _empty_report(kind, floor)
|
|
|
|
if not pairs:
|
|
return _empty_report(kind, floor)
|
|
|
|
# Titles + structural fields + dates, one fetch for every id the scan
|
|
# proposed. Dates matter for non-snippet groups: "is this a correction pair
|
|
# or a chronicle cluster?" is partly answered by when each was written.
|
|
scanned = sorted({n for pair in pairs for n in pair[:2]})
|
|
titles: dict[int, str] = {}
|
|
records: dict[int, dict] = {}
|
|
meta: dict[int, dict] = {}
|
|
try:
|
|
async with async_session() as session:
|
|
rows = (await session.execute(
|
|
select(
|
|
Note.id, Note.title, Note.data,
|
|
Note.created_at, Note.updated_at, Note.task_kind,
|
|
).where(Note.id.in_(scanned))
|
|
)).all()
|
|
for i, t, d, created, updated, task_kind in rows:
|
|
titles[int(i)] = t
|
|
records[int(i)] = d or {}
|
|
meta[int(i)] = {
|
|
"created_at": iso(created),
|
|
"updated_at": iso(updated),
|
|
"task_kind": task_kind,
|
|
}
|
|
except Exception:
|
|
logger.debug("duplicate report titles unavailable", exc_info=True)
|
|
|
|
if kind == "snippet":
|
|
# Fails OPEN, and the direction matters: with `records` empty the
|
|
# filter keeps every pair, so a lookup failure degrades to the
|
|
# unfiltered report rather than to an empty one. A report that silently
|
|
# returns nothing reads as "your corpus is clean", the wrong lie.
|
|
# Snippet-only: the filter keys on symbol/code_sha, which other kinds
|
|
# don't carry — and for them a look-alike is a finding, not a sibling.
|
|
pairs = _drop_sibling_pairs(pairs, records)
|
|
if not pairs:
|
|
return _empty_report(kind, floor)
|
|
|
|
best: dict[tuple[int, int], float] = {(a, b): s for a, b, s in pairs}
|
|
grouped = group_pairs(pairs)
|
|
|
|
# Supersession claims already declared WITHIN a group. A pair someone has
|
|
# already ruled on must not be re-proposed as an open question — and for
|
|
# the reader, an existing claim is the strongest evidence the group is a
|
|
# correction chain rather than a chronicle cluster.
|
|
claims: set[tuple[int, int]] = set()
|
|
if kind != "snippet" and grouped:
|
|
try:
|
|
from scribe.models.note_supersession import NoteSupersession
|
|
all_ids = sorted({n for g in grouped for n in g})
|
|
async with async_session() as session:
|
|
rows = (await session.execute(
|
|
select(
|
|
NoteSupersession.superseder_id,
|
|
NoteSupersession.superseded_id,
|
|
).where(
|
|
NoteSupersession.superseder_id.in_(all_ids),
|
|
NoteSupersession.superseded_id.in_(all_ids),
|
|
)
|
|
)).all()
|
|
claims = {(int(a), int(b)) for a, b in rows}
|
|
except Exception:
|
|
logger.debug("supersession lookup for report failed", exc_info=True)
|
|
|
|
groups = []
|
|
for members in grouped:
|
|
scores = [
|
|
s for (a, b), s in best.items() if a in members and b in members
|
|
]
|
|
group: dict = {
|
|
"note_ids": members,
|
|
# The strongest resemblance in the set — how confident the
|
|
# suggestion is, and what the list sorts on.
|
|
"top_score": max(scores) if scores else floor,
|
|
}
|
|
if kind == "snippet":
|
|
# Historical shape — the existing UI and MCP consumers read
|
|
# `snippets`, and churning them buys nothing.
|
|
group["snippets"] = [
|
|
{"id": nid, "title": titles.get(nid, "")} for nid in members
|
|
]
|
|
else:
|
|
group["members"] = [
|
|
{"id": nid, "title": titles.get(nid, ""), **meta.get(nid, {})}
|
|
for nid in members
|
|
]
|
|
in_group = [
|
|
{"superseder_id": a, "superseded_id": b}
|
|
for (a, b) in sorted(claims)
|
|
if a in members and b in members
|
|
]
|
|
if in_group:
|
|
group["existing_supersessions"] = in_group
|
|
groups.append(group)
|
|
groups.sort(key=lambda g: (-g["top_score"], g["note_ids"][0]))
|
|
return {
|
|
"groups": groups, "pairs": pairs, "threshold": floor,
|
|
"suggestion": _KIND_SUGGESTION[kind],
|
|
}
|
|
|
|
|
|
def _empty_report(kind: str, floor: float) -> dict:
|
|
return {
|
|
"groups": [], "pairs": [], "threshold": floor,
|
|
"suggestion": _KIND_SUGGESTION[kind],
|
|
}
|
|
|
|
|
|
async def find_duplicate_snippets(
|
|
user_id: int, *, threshold: float | None = None, limit: int = _MAX_DUPLICATE_PAIRS
|
|
) -> dict:
|
|
"""The snippet report — find_duplicate_records(kind="snippet"), kept under
|
|
its established name because both surfaces and the SnippetListView consume
|
|
it. New kinds go through the general function."""
|
|
return await find_duplicate_records(
|
|
user_id, kind="snippet", threshold=threshold, limit=limit
|
|
)
|
|
|
|
|
|
async def find_duplicate_rule(
|
|
title: str,
|
|
topic_id: int | None = None,
|
|
project_id: int | None = None,
|
|
) -> DuplicateMatch | None:
|
|
"""Title-based near-duplicate of a rule, scoped to the same topic (a rulebook
|
|
rule) or the same project (a project rule). Rules aren't a semantic-retrieval
|
|
surface, so a normalized-title match is the right (and only) signal. Fail-open
|
|
like find_duplicate_note."""
|
|
norm = " ".join((title or "").split()).lower()
|
|
if not norm or (topic_id is None and project_id is None):
|
|
return None
|
|
try:
|
|
async with async_session() as session:
|
|
stmt = select(Rule).where(
|
|
Rule.deleted_at.is_(None),
|
|
func.lower(func.trim(Rule.title)) == norm,
|
|
)
|
|
if topic_id is not None:
|
|
stmt = stmt.where(Rule.topic_id == topic_id)
|
|
else:
|
|
stmt = stmt.where(Rule.project_id == project_id)
|
|
existing = (await session.execute(stmt.limit(1))).scalars().first()
|
|
if existing is not None:
|
|
return DuplicateMatch(existing.id, existing.title, 1.0, "title")
|
|
except Exception:
|
|
logger.debug("dedup rule title check skipped — query failed", exc_info=True)
|
|
return None
|
|
|
|
|
|
# --- the plan gate (milestone 415) -------------------------------------------
|
|
# A session asked "what work is open?" that cannot see an existing plan makes a
|
|
# second one: a new milestone beside the one that already covers the work, or
|
|
# loose tasks beside it. Each copy then collects its own steps, and neither
|
|
# shows the whole. This gate asks the question before start_planning (or
|
|
# create_milestone) writes: is there an ACTIVE plan in this project for this?
|
|
#
|
|
# Active only: a done milestone is history, and planning the next round of
|
|
# the same area is legitimate work rather than a copy of it.
|
|
#
|
|
# Project-scoped, not owner-scoped like the note gate. The note gate refuses to
|
|
# point at someone else's record because they may not be able to edit it; a
|
|
# plan is different. Creating one needs write on the project, and write on the
|
|
# project is exactly what adding steps to its existing plan needs, so a caller
|
|
# who reaches this gate can act on whatever it returns.
|
|
#
|
|
# Its own threshold, as a setting (rule 25), and lower than the note gate's
|
|
# 0.90 because plan documents are shaped differently: a milestone is embedded
|
|
# as title, description and plan (embeddings.milestone_document), while the
|
|
# candidate usually has no description and carries its steps instead, so even
|
|
# a faithful rewording never scores like a copy.
|
|
#
|
|
# Measured after the first deploy (2026-09-15, bge-small-en-v1.5, #4079): three
|
|
# plans reworded from existing active milestones scored 0.83, 0.85 and 0.87
|
|
# against the milestone they restated; the nearest DIFFERENT plan for each, and
|
|
# a distinct plan on a neighbouring topic, scored 0.72-0.77. At 0.90 the
|
|
# semantic arm matched nothing, leaving only the title arm. 0.80 sits in the
|
|
# gap. A gate that blocks on noise teaches sessions to pass force=true every
|
|
# time, which is why it is not lower; four samples is thin, which is why the
|
|
# operator can move it. Retune alongside the embedder, not the corpus.
|
|
PLAN_MATCH_THRESHOLD_KEY = "kb_plan_match_threshold"
|
|
PLAN_MATCH_DEFAULT_THRESHOLD = 0.80
|
|
|
|
|
|
async def get_plan_match_threshold(user_id: int) -> float:
|
|
"""The user's plan-gate similarity floor, clamped to [0, 1]."""
|
|
from scribe.services.settings import get_setting
|
|
|
|
try:
|
|
value = float(await get_setting(
|
|
user_id, PLAN_MATCH_THRESHOLD_KEY, str(PLAN_MATCH_DEFAULT_THRESHOLD)
|
|
))
|
|
except (TypeError, ValueError):
|
|
value = PLAN_MATCH_DEFAULT_THRESHOLD
|
|
return min(1.0, max(0.0, value))
|
|
|
|
|
|
def plan_candidate_text(
|
|
description: str | None = None,
|
|
body: str | None = None,
|
|
steps: list[tuple[str | None, str | None]] | None = None,
|
|
) -> str:
|
|
"""What a plan that doesn't exist yet says about itself, for the gate.
|
|
|
|
The steps belong in it: a plan passed with steps and no design is still
|
|
recognisable by them, and what its steps say is most of what makes two
|
|
plans the same plan. Each step is (title, body), joined by embedding_text
|
|
like every other record that becomes embedded text (#2486).
|
|
"""
|
|
parts = [(description or "").strip(), (body or "").strip()]
|
|
parts += [embeddings_svc.embedding_text(t, b) for t, b in (steps or [])]
|
|
return "\n\n".join(p for p in parts if p)
|
|
|
|
|
|
async def find_matching_plan(
|
|
user_id: int,
|
|
project_id: int,
|
|
title: str,
|
|
text: str = "",
|
|
) -> DuplicateMatch | None:
|
|
"""An ACTIVE milestone in the project that already is this plan, or None.
|
|
|
|
Normalized-title match first, then semantic when `text` (from
|
|
plan_candidate_text) is long enough to mean something, the same floor the
|
|
note gate uses and for the same reason: a title-only embedding sits in a
|
|
tight neighbourhood and false-positives. Never raises; a failed check lets
|
|
the plan through, because a create must not depend on a recall aid.
|
|
"""
|
|
from scribe.models.milestone import Milestone
|
|
|
|
if not project_id:
|
|
return None
|
|
# Rule 78, before either arm: a match names a milestone, and a caller who
|
|
# cannot read the project must not learn its plans by guessing titles.
|
|
try:
|
|
if not await can_read_project(user_id, project_id):
|
|
return None
|
|
except Exception:
|
|
logger.debug("plan gate access check failed — letting the plan through", exc_info=True)
|
|
return None
|
|
norm = " ".join((title or "").split()).lower()
|
|
if norm:
|
|
try:
|
|
async with async_session() as session:
|
|
existing = (await session.execute(
|
|
select(Milestone).where(
|
|
Milestone.project_id == project_id,
|
|
Milestone.deleted_at.is_(None),
|
|
Milestone.status == "active",
|
|
func.lower(func.trim(Milestone.title)) == norm,
|
|
).limit(1)
|
|
)).scalars().first()
|
|
if existing is not None:
|
|
return DuplicateMatch(existing.id, existing.title, 1.0, "title")
|
|
except Exception:
|
|
logger.debug("plan gate title check skipped — query failed", exc_info=True)
|
|
return None
|
|
|
|
if len((text or "").strip()) < _MIN_BODY_FOR_SEMANTIC:
|
|
return None
|
|
doc_title, doc_body = embeddings_svc.milestone_document(title, None, text)
|
|
query = "\n\n".join(p for p in (doc_title, doc_body) if p)
|
|
try:
|
|
hits = await embeddings_svc.semantic_search_milestones(
|
|
user_id, query, project_id=project_id, status="active", limit=1,
|
|
threshold=await get_plan_match_threshold(user_id),
|
|
)
|
|
except Exception:
|
|
logger.debug("plan gate semantic check skipped", exc_info=True)
|
|
return None
|
|
if hits:
|
|
score, milestone = hits[0]
|
|
return DuplicateMatch(milestone.id, milestone.title, round(score, 3), "semantic")
|
|
return None
|
|
|
|
|
|
def plan_match_response(dup: DuplicateMatch, progress: dict | None = None) -> dict:
|
|
"""The payload start_planning / create_milestone return instead of a second
|
|
plan: the existing one, how far along it is, and how to add to it."""
|
|
progress = progress or {}
|
|
total, completed = progress.get("total", 0), progress.get("completed", 0)
|
|
how = "has the same title" if dup.reason == "title" else "reads as the same plan"
|
|
return {
|
|
"duplicate": True,
|
|
"existing_id": dup.id,
|
|
"existing_title": dup.title,
|
|
"existing_milestone": {
|
|
"id": dup.id,
|
|
"title": dup.title,
|
|
"description": progress.get("description") or "",
|
|
"total": total,
|
|
"completed": completed,
|
|
},
|
|
"similarity": dup.similarity,
|
|
"match": dup.reason,
|
|
"message": (
|
|
f'An active plan in this project {how}: milestone {dup.id} '
|
|
f'"{dup.title}" ({completed} of {total} steps done). Nothing was '
|
|
f"created. Add your steps to it with create_records(milestone_id="
|
|
f"{dup.id}, ...), and revise its design with update_milestone if the "
|
|
f"scope has grown. Read it first with get_milestone({dup.id}). If this "
|
|
f"really is a separate plan, retry with force=true."
|
|
),
|
|
}
|
|
|
|
|
|
async def plan_gate(
|
|
user_id: int,
|
|
project_id: int,
|
|
title: str,
|
|
text: str = "",
|
|
) -> dict | None:
|
|
"""find_matching_plan, answered: the plan_match_response to return in
|
|
place of a new plan, or None to go ahead and create it."""
|
|
from scribe.services import milestones as milestones_svc
|
|
|
|
dup = await find_matching_plan(user_id, project_id, title, text)
|
|
if dup is None:
|
|
return None
|
|
try:
|
|
rows = await milestones_svc.get_project_milestone_summary(user_id, project_id)
|
|
progress = next((r for r in rows if r.get("id") == dup.id), None)
|
|
except Exception:
|
|
# The match stands without its progress; losing the count must not
|
|
# turn a found plan into a second one.
|
|
progress = None
|
|
return plan_match_response(dup, progress)
|