FabledScribe/src/scribe/services/dedup.py

"""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.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


@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"


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' or 'task' (drives the
    update_<kind> hint)."""
    return {
        "duplicate": True,
        "existing_id": dup.id,
        "existing_title": dup.title,
        "similarity": dup.similarity,
        "match": dup.reason,
        "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."
        ),
    }


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",
) -> DuplicateMatch | None:
    """Best near-duplicate of (title, body) within the same owner + project +
    kind, or None. Title match first (cheap, exact), 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)."""
    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: semantic similarity (only with a substantial body) ---
    if body and len(body.strip()) >= _MIN_BODY_FOR_SEMANTIC:
        query = f"{title}\n{body}".strip()
        # 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=_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",
        )
        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 A LOWER THRESHOLD THAN THE GATE. The gate BLOCKS a write at 0.90 and has to
# be unforgiving of noise. This report only makes a suggestion the operator
# reviews, so it can afford to be looser and catch the pairs the gate lets
# through — which are precisely the ones that accumulated. It 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_KEY = "kb_duplicate_threshold"
DUPLICATE_DEFAULT_THRESHOLD = 0.82
# 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) -> float:
    """The user's near-duplicate similarity floor, clamped to [0, 1]."""
    from scribe.services.settings import get_setting

    try:
        value = float(await get_setting(
            user_id, DUPLICATE_THRESHOLD_KEY, str(DUPLICATE_DEFAULT_THRESHOLD)
        ))
    except (TypeError, ValueError):
        value = DUPLICATE_DEFAULT_THRESHOLD
    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]))


async def find_duplicate_snippets(
    user_id: int, *, threshold: float | None = None, limit: int = _MAX_DUPLICATE_PAIRS
) -> dict:
    """Near-duplicate snippets already in the record, grouped into merge 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.

    Returns {"groups": [{"note_ids": [...], "snippets": [...], "top_score": f}],
    "pairs": [...], "threshold": f}. Fail-open (an empty report) like the rest of
    this module — a suggestion feature must not be able to break the page it
    decorates.
    """
    floor = await get_duplicate_threshold(user_id) 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)

    pairs: list[tuple[int, int, float]] = []
    try:
        async with async_session() as session:
            stmt = (
                select(left.note_id, right.note_id, distance.label("distance"))
                .select_from(left)
                # `<` not `!=`: each unordered pair exactly once, and it drops
                # the self-pair (distance 0) that would otherwise dominate.
                .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(
                    left_note.note_type == SNIPPET_NOTE_TYPE,
                    right_note.note_type == SNIPPET_NOTE_TYPE,
                    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,
                    distance <= max_distance,
                )
                .order_by(distance.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 snippet scan failed", exc_info=True)
        return {"groups": [], "pairs": [], "threshold": floor}

    if not pairs:
        return {"groups": [], "pairs": [], "threshold": floor}

    best: dict[tuple[int, int], float] = {(a, b): s for a, b, s in pairs}
    grouped = group_pairs(pairs)

    # Titles for presentation. One fetch for every id in the report.
    ids = sorted({n for g in grouped for n in g})
    titles: dict[int, str] = {}
    try:
        async with async_session() as session:
            rows = (await session.execute(
                select(Note.id, Note.title).where(Note.id.in_(ids))
            )).all()
            titles = {int(i): t for i, t in rows}
    except Exception:
        logger.debug("duplicate report titles unavailable", 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
        ]
        groups.append({
            "note_ids": members,
            "snippets": [
                {"id": nid, "title": titles.get(nid, "")} for nid in 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,
        })
    groups.sort(key=lambda g: (-g["top_score"], g["note_ids"][0]))
    return {"groups": groups, "pairs": pairs, "threshold": floor}


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