FabledScribe/src/scribe/services/knowledge.py

"""Knowledge service — one query across every record kind Scribe holds.

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 re
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"
            # Repo names are recorded free-form ("Scribe" / "FabledScribe" /
            # "fabledscribe") — case is never the distinguishing thing (#2874).
            else (loc.get(key) or "").strip().lower() == want.lower()
            if key == "repo"
            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})")
        elif key == "repo":
            # Case-insensitive, anchored, regex-escaped (#2874) — mirrors the
            # Python dialect's .lower() compare.
            pattern = json.dumps("^" + re.escape(want) + "$")
            filters.append(f'(@.repo like_regex {pattern} flag "i")')
        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


# What each type facet MEANS, once, for every arm that has to know.
#
# The vocabulary spans BOTH typing axes — `note_type` for non-task records and
# `task_kind` for tasks — so a facet cannot be a filter on one column, which is
# why this is a table rather than a chain of ifs. Each entry is
# (is_task, the value pinned on that axis); None pins nothing, i.e. every task.
#
# It is a table because the alternative had already gone wrong. The predicate
# was written three times — a SQL if-chain, a Python if-chain over semantic
# candidates, and a ternary computing the `is_task` pre-filter — and the three
# only agreed by luck. Adding `issue` to the SQL arm alone (the obvious edit,
# and the one #3128 was about to make) would have set the pre-filter to
# is_task=False, handed the Python arm a candidate set containing no tasks at
# all, and returned an empty semantic half for the Issues facet forever, with
# nothing red anywhere. A new facet is now one row here.
#
# `plan` is retired (0066) but kept: 90 legacy plan-tasks exist and a facet
# they answer to costs one line. It simply has no chip in the UI any more.
_FACETS: dict[str, tuple[bool, str | None]] = {
    "task": (True, None),
    "work": (True, "work"),
    "issue": (True, "issue"),
    "spike": (True, "spike"),
    "plan": (True, "plan"),
    "note": (False, "note"),
    "process": (False, "process"),
    "snippet": (False, "snippet"),
}

# The non-task record types, for the counts query. Derived so it cannot drift
# from the table above.
NON_TASK_FACETS = tuple(
    value for _is_task, value in _FACETS.values() if not _is_task and value
)

# The whole vocabulary, for the door's request validation — public so the route
# validates against the same table the query reads instead of a hand-kept copy.
FACET_TYPES = frozenset(_FACETS)


# An unrecognised facet resolves to "a non-task note whose note_type is that
# string" — which matches nothing, since no row stores an unknown type. That is
# the behaviour the old if-chain had by falling through, and it is the right
# one: a typo should return an empty list, never the whole corpus.
def _facet(note_type: str) -> tuple[bool, str | None]:
    return _FACETS.get(note_type, (False, note_type))


def facet_is_task(note_type: str | None) -> bool | None:
    """The `is_task` pre-filter a facet implies — None when it spans both.

    Used to narrow the semantic candidate set before it is fetched. Reads the
    same table `_apply_type_filter` and `matches_facet` read, so the pre-filter
    can no longer disagree with the predicate it is meant to anticipate.
    """
    if not note_type:
        return None
    return _facet(note_type)[0]


def matches_facet(note, note_type: str | None) -> bool:
    """The Python dialect of `_apply_type_filter`, for candidates the vector
    search has already fetched — there is no query left to narrow.

    Generated from the same table, so this is a translation rather than a
    second implementation. Note the `not note.is_task` arm: the hand-written
    version omitted it and was saved only by the upstream pre-filter.
    """
    if not note_type:
        return True
    is_task, value = _facet(note_type)
    if is_task:
        return note.is_task and (value is None or note.task_kind == value)
    return not note.is_task and note.note_type == value


def _apply_type_filter(stmt, note_type: str | None):
    """Apply the type facet to a Note select. Trashed rows are always excluded."""
    stmt = stmt.where(Note.deleted_at.is_(None))
    if not note_type:
        return stmt
    is_task, value = _facet(note_type)
    if is_task:
        stmt = stmt.where(Note.status.isnot(None))
        if value is not None:
            stmt = stmt.where(Note.task_kind == value)
        return stmt
    return stmt.where(Note.status.is_(None)).where(Note.note_type == value)


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 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 = facet_is_task(note_type)
        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 not matches_facet(note, 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:
        def _scoped(stmt):
            stmt = stmt.where(visible).where(Note.deleted_at.is_(None))
            for tag in tags or []:
                stmt = stmt.where(Note.tags.contains([tag]))
            return stmt

        # One grouped query per typing axis. The task half used to be a count
        # for 'task' plus a second count for 'plan', which is why 'issue' —
        # 17% of every task here — had no number to show: each kind needed its
        # own query and nobody added one. Grouping by task_kind counts every
        # kind, including ones added later, for the same two round-trips.
        non_task = _scoped(
            select(Note.note_type, func.count(Note.id))
            .where(Note.status.is_(None))
            .where(Note.note_type.in_(NON_TASK_FACETS))
        ).group_by(Note.note_type)
        counts = {t: n for t, n in (await session.execute(non_task)).all()}

        by_kind = _scoped(
            select(Note.task_kind, func.count(Note.id))
            .where(Note.status.isnot(None))
        ).group_by(Note.task_kind)
        kind_counts = {k: n for k, n in (await session.execute(by_kind)).all()}

    # Kinds are SUBSETS of 'task' and are deliberately left out of the total —
    # adding them would count every task twice.
    counts["task"] = sum(kind_counts.values())
    for kind, value in _FACETS.items():
        if value[0] and value[1] is not None:
            counts[kind] = kind_counts.get(kind, 0)

    for t in NON_TASK_FACETS:
        counts.setdefault(t, 0)
    counts["total"] = counts["task"] + sum(counts[t] for t in NON_TASK_FACETS)
    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]