FabledScribe/src/scribe/services/knowledge.py

"""Knowledge service — unified query across notes, tasks, plans, and processes.

ACL (rules #47/#78, decision note 2094): these queries were owner-only until
2026-07-25, which meant a record shared with you could be opened by id but never
*found*. They now honour shares — at two different widths:

  - **searching** (a `q` the caller typed) uses the full read scope, so a record
    shared directly with you is findable when you go looking for it. BOTH halves
    of the hybrid search — keyword and semantic — see equally, or a record would
    be findable by wording and invisible by meaning;
  - **browsing** (no `q`) and the facet counts beside it use the narrower browse
    scope: your own records plus anything in a project you can reach.

The asymmetry is the point. A record that appears unasked reads as one you
endorsed, so a one-off direct share has to be searched for rather than arriving
in your ambient lists.
"""
import json
import logging

from sqlalchemy import and_, func, or_, select

from scribe.models import async_session
from scribe.models.note import Note
from scribe.models.base import iso
from scribe.services.access import browsable_notes_clause, readable_notes_clause

logger = logging.getLogger(__name__)

_SNIPPET_LEN = 200


# --- the location filter (reverse lookup, #2083) ------------------------------
#
# ONE predicate in two dialects: "this record carries a location matching every
# part asked for." The SQL form runs in the browse arm and the keyword arm,
# before the count and the page slice, so totals stay honest; the Python form
# runs in the semantic arm, which post-filters candidates it already holds in
# memory. THE TWO MUST CHANGE TOGETHER — a filter only one arm applies makes a
# snippet findable by wording and invisible by location, which is exactly the
# split tests/test_retrieval_scopes.py exists to prevent.
#
# Semantics, both dialects:
#   - parts are ANDed WITHIN a single location entry, never across the list. A
#     snippet whose first location is in repo A and whose second is at path B
#     does not answer repo=A + path=B — it was never in that place.
#   - `path` matches exactly OR as a directory prefix: "frontend/src" finds
#     "frontend/src/lib/x.ts". Prefix is the one part a GIN containment lookup
#     can't serve (migration 0070), which is why this is a jsonpath `@?` rather
#     than `@>` — jsonpath `starts with` is index-served by the same GIN index.
#
# The filter reads `notes.data`, so it only sees rows carrying that mirror.
# Rows written before migration 0070 are populated once at startup by
# snippets.backfill_snippet_data — without that, this query would answer "no
# snippets here" for an old snippet and the caller would write the helper again.

LOCATION_KEYS = ("repo", "path", "symbol")


def location_parts(repo: str = "", path: str = "", symbol: str = "") -> dict[str, str]:
    """The non-empty, stripped location parts asked for; {} when none were."""
    given = {"repo": repo, "path": path, "symbol": symbol}
    return {k: (v or "").strip() for k, v in given.items() if (v or "").strip()}


def _path_matches(have: str, want: str) -> bool:
    """Exact, or `have` sits somewhere under the `want` directory."""
    return have == want or have.startswith(want.rstrip("/") + "/")


def location_matches(data: dict | None, parts: dict[str, str]) -> bool:
    """Python dialect of the location predicate. Keep in step with _location_clause."""
    if not parts:
        return True
    for loc in (data or {}).get("locations") or []:
        if all(
            _path_matches((loc.get(key) or "").strip(), want)
            if key == "path"
            else (loc.get(key) or "").strip() == want
            for key, want in parts.items()
        ):
            return True
    return False


def location_jsonpath(parts: dict[str, str]) -> str:
    """The jsonpath behind the SQL dialect — one `locations` entry matching all parts.

    Values are embedded as JSON string literals (jsonpath uses JSON quoting), so
    a repo or path carrying a quote can't break out of the expression. The keys
    are our own fixed set, never caller input.
    """
    filters = []
    for key in LOCATION_KEYS:
        if key not in parts:
            continue
        want = parts[key]
        literal = json.dumps(want)
        if key == "path":
            prefix = json.dumps(want.rstrip("/") + "/")
            filters.append(f"(@.path == {literal} || @.path starts with {prefix})")
        else:
            filters.append(f"@.{key} == {literal}")
    return f"$.locations[*] ? ({' && '.join(filters)})"


def _location_clause(parts: dict[str, str]):
    """SQL dialect of the location predicate. Keep in step with location_matches."""
    return Note.data.path_exists(location_jsonpath(parts))


# --- drift-check filter (#2086) ----------------------------------------------
# `verification` selects on the drift-check verdict stored in `data.verification`
# (see services/snippets.py). Statuses are the service's own constants; the two
# composite values are what the operator actually asks for.
#
# The interesting one is `attention`. A verdict describes the code it was checked
# against, so an OK verdict on code that has since been edited is not an OK
# record — nobody has checked what's actually there. That is expressible in SQL
# only because `data.code_sha` mirrors the current code's fingerprint alongside
# the verdict's: jsonpath compares the two fields within the row, so this stays
# one index-served predicate rather than a post-filter that would make the
# pagination total a lie.
#
# Same two-dialect discipline as the location filter above, and the same hazard:
# THE TWO MUST CHANGE TOGETHER. tests/test_snippet_drift_check.py is the guard —
# it walks both dialects over the same cases, including the one that motivated
# `attention` existing (an ok verdict whose code_sha has gone stale, which is
# neither `drifted` nor `unverified` yet plainly needs looking at).

def verification_matches(data: dict | None, value: str) -> bool:
    """Python dialect of the verification predicate.

    Keep in step with _verification_clause — the semantic arm's candidates are
    already fetched, so there is no query left to narrow and the same rule has
    to be expressible twice. Same arrangement as location_matches.
    """
    want = (value or "").strip().lower()
    if not want:
        return True
    verdict = (data or {}).get("verification") or {}
    status = verdict.get("status") or ""
    if not status:
        return want == "unverified"
    expired = verdict.get("code_sha") != (data or {}).get("code_sha")
    # A push touched the recorded location since the verdict (#2691): the repo
    # moved under it, so it needs a look even though it hasn't failed.
    invalidated = bool(verdict.get("invalidated_by"))
    if want == "unverified":
        return False
    if want == "drifted":
        return status != "ok"
    if want == "attention":
        return status != "ok" or expired or invalidated
    if want == "ok":
        return status == "ok" and not expired and not invalidated
    return status == want


_VERIFY_DRIFTED_JSONPATH = '$.verification ? (@.status != "ok")'
_VERIFY_EXPIRED_JSONPATH = "$ ? (@.verification.code_sha != @.code_sha)"
_VERIFY_ANY_JSONPATH = "$.verification"
# A push touched the recorded location since the verdict (#2691).
_VERIFY_INVALIDATED_JSONPATH = "$.verification.invalidated_by"


def _verification_clause(value: str):
    """SQL predicate for one `verification` filter value, or None for no filter."""
    want = (value or "").strip().lower()
    if not want:
        return None
    has_verdict = Note.data.path_exists(_VERIFY_ANY_JSONPATH)
    if want == "unverified":
        # Never checked at all. Rows predating migration 0070 have no `data`
        # whatsoever and land here correctly — which is right, they haven't been.
        return ~has_verdict
    if want == "drifted":
        return Note.data.path_exists(_VERIFY_DRIFTED_JSONPATH)
    if want == "attention":
        # Everything worth looking at: a failing verdict, an expired one, OR
        # one whose recorded location a push has since touched (#2691).
        return or_(
            Note.data.path_exists(_VERIFY_DRIFTED_JSONPATH),
            and_(has_verdict, Note.data.path_exists(_VERIFY_EXPIRED_JSONPATH)),
            Note.data.path_exists(_VERIFY_INVALIDATED_JSONPATH),
        )
    if want == "ok":
        # A clean bill of health that still describes the current code. The
        # `~expired` half matters: without it this would quietly include records
        # whose blessing has lapsed, which is the exact failure the feature is
        # meant to catch. Same for push-invalidation — "ok" must mean the repo
        # hasn't moved under the verdict either.
        return and_(
            Note.data.path_exists('$.verification ? (@.status == "ok")'),
            ~Note.data.path_exists(_VERIFY_EXPIRED_JSONPATH),
            ~Note.data.path_exists(_VERIFY_INVALIDATED_JSONPATH),
        )
    # A specific status: 'missing' | 'moved' | 'changed'.
    return Note.data.path_exists(
        f"$.verification ? (@.status == {json.dumps(want)})"
    )


def _note_to_item(note: Note) -> dict:
    item: dict = {
        "id": note.id,
        "note_type": note.note_type or "note",
        "title": note.title,
        "snippet": (note.body or "")[:_SNIPPET_LEN],
        "tags": note.tags or [],
        "project_id": note.project_id,
        # These lists now include records shared with the caller, so the client
        # needs the owner to tell "mine" from "someone else's" in a mixed list.
        "user_id": note.user_id,
        "created_at": iso(note.created_at),
        "updated_at": iso(note.updated_at),
    }
    # Drift verdict (#2086), when one has been recorded. Included here rather
    # than decorated on by the snippet layer because `current` is derivable from
    # `data` alone — the verdict's code_sha against the row's — so this needs no
    # body parsing and stays a plain projection of the column. Omitted entirely
    # when unchecked, so "no key" and "never verified" don't become two states
    # the client has to tell apart.
    # Snippet language, same reasoning as the verdict below: a plain projection of
    # the `data` mirror, no body parsing. Needed because a prior-art hit in a
    # DIFFERENT language than the file being written is useful as the shape of a
    # solution but must not be mistaken for code to paste (#2244) — and the caller
    # can only say "different" if the language is on the item.
    language = (note.data or {}).get("language") if note.data else None
    if language:
        item["language"] = language

    verdict = (note.data or {}).get("verification") if note.data else None
    if verdict and verdict.get("status"):
        item["verification"] = {
            "status": verdict["status"],
            "current": verdict.get("code_sha") == (note.data or {}).get("code_sha"),
            "checked_at": verdict.get("checked_at"),
            "detail": verdict.get("detail"),
            "path": verdict.get("path"),
        }
        # Present only when a push has touched the recorded location since the
        # verdict (#2691) — the "recheck me" marker, cleared by re-verifying.
        if verdict.get("invalidated_by"):
            item["verification"]["invalidated_by"] = verdict["invalidated_by"]

    # Task fields — override note_type and add status/priority/due_date
    if note.is_task:
        item["note_type"] = "task"
        item["task_kind"] = note.task_kind
        item["status"] = note.status
        item["priority"] = note.priority
        item["due_date"] = iso(note.due_date)

    return item


def _apply_type_filter(stmt, note_type: str | None):
    """Apply the type facet to a Note select.

    'task' = any task (status not null); 'plan' = a task with task_kind='plan';
    any other non-empty type = a non-task note of that note_type; None = all.

    Trashed rows (deleted_at set) are always excluded.
    """
    stmt = stmt.where(Note.deleted_at.is_(None))
    if note_type == "task":
        return stmt.where(Note.status.isnot(None))
    if note_type == "plan":
        return stmt.where(Note.status.isnot(None)).where(Note.task_kind == "plan")
    if note_type:
        return stmt.where(Note.note_type == note_type).where(Note.status.is_(None))
    return stmt


async def query_knowledge(
    user_id: int,
    note_type: str | None,
    tags: list[str],
    sort: str,
    q: str | None,
    limit: int,
    offset: int,
    project_id: int | None = None,
    locations: dict[str, str] | None = None,
    verification: str = "",
) -> tuple[list[dict], int]:
    """Query knowledge objects (non-task notes) with filters.

    `project_id` narrows to one project (None = every project).

    `locations` narrows to records whose `data.locations` holds an entry matching
    every part given — build it with `location_parts(repo=…, path=…, symbol=…)`.
    Today only snippets carry locations, but the column is general, so the filter
    lives here with the query rather than in one type's service.

    `verification` narrows on the drift-check verdict: 'ok', 'drifted',
    'unverified', 'attention', or one specific failure ('missing' | 'moved' |
    'changed'). Empty means no filter.

    Returns (items, total_count).
    """
    # Semantic search path — scores take priority over sort
    if q:
        return await _semantic_knowledge_search(
            user_id, q, note_type=note_type, tags=tags, limit=limit,
            offset=offset, project_id=project_id, locations=locations,
            verification=verification,
        )

    # No query = browsing. Narrower scope: a record shared directly with the
    # caller is search-only and must not appear in an ambient list.
    visible = browsable_notes_clause(user_id)
    async with async_session() as session:
        base = select(Note).where(visible)

        base = _apply_type_filter(base, note_type)
        if project_id is not None:
            base = base.where(Note.project_id == project_id)

        for tag in tags:
            base = base.where(Note.tags.contains([tag]))

        if locations:
            base = base.where(_location_clause(locations))

        verify_clause = _verification_clause(verification)
        if verify_clause is not None:
            base = base.where(verify_clause)

        # Count before pagination
        count_stmt = select(func.count()).select_from(base.subquery())
        total: int = (await session.execute(count_stmt)).scalar_one()

        # Apply sort
        if sort == "created":
            base = base.order_by(Note.created_at.desc())
        elif sort == "alpha":
            base = base.order_by(Note.title.asc())
        elif sort == "type":
            base = base.order_by(Note.note_type.asc(), Note.updated_at.desc())
        else:  # modified (default)
            base = base.order_by(Note.updated_at.desc())

        rows = list((await session.execute(base.limit(limit).offset(offset))).scalars().all())

    return [_note_to_item(n) for n in rows], total


async def _semantic_knowledge_search(
    user_id: int,
    q: str,
    note_type: str | None,
    tags: list[str],
    limit: int,
    offset: int,
    project_id: int | None = None,
    locations: dict[str, str] | None = None,
    verification: str = "",
) -> tuple[list[dict], int]:
    """Hybrid search: keyword matches first (title/body ILIKE), then semantic results.

    Exact keyword matches always rank above semantic-only matches so that
    searching for a name like "Weston" surfaces the note with that title
    before conceptually related notes.

    BEST-EFFORT TOP-N, not exhaustive pagination: the ranked candidate set is
    capped (keyword limit*2 + up to ~200 semantic), so `total` is the size of
    that window, NOT the true match count, and matches beyond the cap are not
    reachable by paging. Each page also recomputes the full merge (O(corpus)
    per page). Acceptable for an interactive "best results" feed; a cached
    ranked-id list or pgvector ORDER BY/LIMIT is the fix if exhaustive,
    cheap pagination is ever needed.
    """
    # 1. Keyword search — title and body ILIKE
    keyword_notes: list[Note] = []
    try:
        # A typed query is an explicit act, so it reaches the caller's full read
        # scope — including records shared directly with them.
        visible = readable_notes_clause(user_id)
        async with async_session() as session:
            pattern = f"%{q}%"
            base = (
                select(Note)
                .where(visible)
                .where(Note.title.ilike(pattern) | Note.body.ilike(pattern))
            )
            base = _apply_type_filter(base, note_type)
            if project_id is not None:
                base = base.where(Note.project_id == project_id)
            for tag in tags:
                base = base.where(Note.tags.contains([tag]))
            if locations:
                base = base.where(_location_clause(locations))
            verify_clause = _verification_clause(verification)
            if verify_clause is not None:
                base = base.where(verify_clause)
            # Title matches first, then body-only matches, newest first within each
            base = base.order_by(
                Note.title.ilike(pattern).desc(),
                Note.updated_at.desc(),
            ).limit(limit * 2)
            keyword_notes = list((await session.execute(base)).scalars().all())
    except Exception:
        logger.warning("Keyword search failed", exc_info=True)

    # 2. Semantic search — conceptual similarity, at the SAME scope as the
    # keyword half above. Both halves of one search must see equally, or a shared
    # record would be findable by wording and invisible by meaning — which is the
    # case a semantic search exists to serve.
    semantic_notes: list[Note] = []
    try:
        from scribe.services.embeddings import (
            INTERACTIVE_SEARCH_THRESHOLD,
            semantic_search_notes,
        )
        is_task_filter = True if note_type in ("task", "plan") else (False if note_type else None)
        import time as _time
        _t0 = _time.perf_counter()
        candidates = await semantic_search_notes(
            user_id=user_id,
            scope="read",
            query=q,
            limit=min(200, limit * 4),
            # The shared interactive floor — this was a bare `0.3` while
            # routes/search.py had the same number as a commented constant, the
            # exact pair where one moves and the other doesn't (#2463).
            threshold=INTERACTIVE_SEARCH_THRESHOLD,
            is_task=is_task_filter,
            project_id=project_id,
        )
        # The human's MAIN search surface, and it logged nothing — so
        # retrieval_logs claimed the web's search was /api/search, a narrower
        # path with (measured) zero frontend consumers. Distinct source, so
        # threshold tuning can include or exclude human queries deliberately
        # rather than by accident (#2463).
        from scribe.services.retrieval_telemetry import record_retrieval
        record_retrieval(
            user_id=user_id, source="browse_search", query=q,
            threshold=INTERACTIVE_SEARCH_THRESHOLD, limit=min(200, limit * 4),
            project_id=project_id, is_task=is_task_filter, results=candidates,
            duration_ms=(_time.perf_counter() - _t0) * 1000.0,
        )
        for _score, note in candidates:
            if note.deleted_at is not None:
                continue
            if note_type == "task" and not note.is_task:
                continue
            elif note_type == "plan" and (not note.is_task or note.task_kind != "plan"):
                continue
            elif note_type and note_type not in ("task", "plan") and note.note_type != note_type:
                continue
            if tags and not all(t in (note.tags or []) for t in tags):
                continue
            # The Python dialect of the same predicate the SQL arms apply above —
            # these candidates arrive already fetched, so there's no query to
            # narrow. See the comment on location_matches.
            if locations and not location_matches(note.data, locations):
                continue
            if verification and not verification_matches(note.data, verification):
                continue
            semantic_notes.append(note)
    except Exception:
        logger.warning("Semantic search unavailable, using keyword results only", exc_info=True)

    # 3. Merge — keyword matches first, then semantic (deduplicated)
    seen_ids: set[int] = set()
    merged: list[Note] = []
    for note in keyword_notes:
        if note.id not in seen_ids:
            seen_ids.add(note.id)
            merged.append(note)
    for note in semantic_notes:
        if note.id not in seen_ids:
            seen_ids.add(note.id)
            merged.append(note)

    total = len(merged)
    page_items = merged[offset: offset + limit]
    return [_note_to_item(n) for n in page_items], total


async def get_knowledge_tags(user_id: int, note_type: str | None = None) -> list[str]:
    """Distinct tags across what this user can BROWSE.

    Follows the browse list rather than the read scope: a facet is itself a
    passive surface, and offering a tag that only a search-only record carries
    would filter the visible list down to nothing."""
    visible = browsable_notes_clause(user_id)
    async with async_session() as session:
        base = (
            select(func.unnest(Note.tags).label("tag"))
            .where(visible)
        )
        base = _apply_type_filter(base, note_type)
        stmt = base.distinct().order_by("tag")
        rows = list((await session.execute(stmt)).scalars().all())
    return [r for r in rows if r]


async def get_knowledge_counts(user_id: int, tags: list[str] | None = None) -> dict[str, int]:
    """Per-type counts for the sidebar, over what this user can BROWSE — so the
    numbers match the list they sit beside rather than promising rows that only a
    search would surface."""
    visible = browsable_notes_clause(user_id)
    async with async_session() as session:
        # Count non-task types
        stmt = (
            select(Note.note_type, func.count(Note.id))
            .where(visible)
            .where(Note.status.is_(None))
            .where(Note.deleted_at.is_(None))
            .where(Note.note_type.in_(["note", "process"]))
            .group_by(Note.note_type)
        )
        if tags:
            for tag in tags:
                stmt = stmt.where(Note.tags.contains([tag]))
        rows = list((await session.execute(stmt)).all())
        counts = {row[0]: row[1] for row in rows}

        # Count tasks separately (is_task = status IS NOT NULL)
        task_stmt = (
            select(func.count(Note.id))
            .where(visible)
            .where(Note.status.isnot(None))
            .where(Note.deleted_at.is_(None))
        )
        if tags:
            for tag in tags:
                task_stmt = task_stmt.where(Note.tags.contains([tag]))
        task_count: int = (await session.execute(task_stmt)).scalar_one()
        counts["task"] = task_count

        # Plans are a subset of tasks (task_kind='plan'); counted for the facet
        # but NOT added to total to avoid double-counting against "task".
        plan_stmt = (
            select(func.count(Note.id))
            .where(visible)
            .where(Note.status.isnot(None))
            .where(Note.task_kind == "plan")
            .where(Note.deleted_at.is_(None))
        )
        if tags:
            for tag in tags:
                plan_stmt = plan_stmt.where(Note.tags.contains([tag]))
        counts["plan"] = (await session.execute(plan_stmt)).scalar_one()

    for t in ("note", "task", "plan", "process"):
        counts.setdefault(t, 0)
    counts["total"] = sum(counts[t] for t in ("note", "task", "process"))
    return counts


async def query_knowledge_ids(
    user_id: int,
    note_type: str | None,
    tags: list[str],
    sort: str,
    q: str | None,
    limit: int = 100,
    offset: int = 0,
) -> tuple[list[int], int]:
    """Return note IDs only — cheap query for the two-tier pagination feed."""
    if q:
        # Re-use semantic search, extract IDs in rank order
        items, total = await _semantic_knowledge_search(
            user_id, q, note_type=note_type, tags=tags,
            limit=limit, offset=offset,
        )
        return [item["id"] for item in items], total

    # Browsing (see query_knowledge) — narrower scope.
    visible = browsable_notes_clause(user_id)
    async with async_session() as session:
        base = select(Note.id).where(visible)

        base = _apply_type_filter(base, note_type)
        for tag in tags:
            base = base.where(Note.tags.contains([tag]))

        count_stmt = select(func.count()).select_from(base.subquery())
        total: int = (await session.execute(count_stmt)).scalar_one()

        if sort == "created":
            base = base.order_by(Note.created_at.desc())
        elif sort == "alpha":
            base = base.order_by(Note.title.asc())
        elif sort == "type":
            base = base.order_by(Note.note_type.asc(), Note.updated_at.desc())
        else:
            base = base.order_by(Note.updated_at.desc())

        ids = list((await session.execute(base.limit(limit).offset(offset))).scalars().all())

    return ids, total


async def get_knowledge_by_ids(user_id: int, ids: list[int]) -> list[dict]:
    """Fetch full items for the given IDs, preserving the requested order."""
    if not ids:
        return []
    # Fetching specific ids is explicit, so this takes the full read scope — the
    # ids came from either a browse or a search, and both must resolve.
    visible = readable_notes_clause(user_id)
    async with async_session() as session:
        stmt = (
            select(Note)
            .where(visible)
            .where(Note.id.in_(ids))
            .where(Note.deleted_at.is_(None))
        )
        rows = list((await session.execute(stmt)).scalars().all())
    by_id = {n.id: n for n in rows}
    return [_note_to_item(by_id[i]) for i in ids if i in by_id]