brain-of-reese/app/api/docs.py

"""GET /api/docs — the indexed document list (feeds the Sources page).

GET /api/documents/content — one indexed document's full content (feeds the
clickable document viewer, phase 10). DB-only by design: the (source, path)
pair is looked up as a row, so there is no filesystem access and no
path-traversal surface — ``../``-style values simply aren't rows (→ 404).

PATCH /api/documents/summary — the admin summary editor (phase 57):
update or clear ``documents.summary`` and re-embed the ``is_summary``
chunk (embed first, mutate second — a failed LLM call leaves the row and
chunk untouched; the content chunks are never re-embedded, D4).

PATCH /api/documents/date — the admin document-date editor (phase 106,
D7): set the owner's corrected ``documents.created_at`` (normalized
through ``app.rag.doc_dates.normalize_doc_date`` — a manually set future
date folds to today, D3) + flag it manual, or clear the manual flag
(null date — the stored date stands until the next sync). A pure DB
write: NO LLM/embedding call — a date is never embedded (the
deliberate contrast with the phase-57 ``is_summary`` re-embed).

GET /api/docs/tree — the admin's full recursive KB tree in one fetch
(phase 97, task 02): the same drill-down tree the agent's ``ls``
walks, with the file metadata the RAG view's rows and stat cards need
(the view drills client-side; ``GET /api/docs`` is untouched).

PATCH /api/folders/summary — the admin folder-description editor
(phase 97, task 03): update / create / clear a stored
``folder_summaries`` row, marking every non-empty save
``manually_edited`` (from this point on the sync-time generator skips
the row and never prunes it — task 01). A pure DB write: NO
LLM/embedding call — a folder description is never embedded (no
chunk, no retrieval role beyond the ``ls`` line), the deliberate
contrast with the phase-57 ``is_summary`` re-embed above.
"""
from __future__ import annotations

from collections.abc import Mapping, Sequence
from datetime import UTC, datetime
from pathlib import Path

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.api.sync import _sanitize_error
from app.config import get_settings
from app.core.auth import require_admin, require_user
from app.db import get_db
from app.models import Chunk, Document, FolderSummary
from app.rag.agent import list_source_names
from app.rag.doc_dates import normalize_doc_date
from app.rag.folder_summaries import MIN_DOCS_PER_FOLDER, folder_of
from app.rag.importer import match_extension
from app.rag.llm import EmbeddingError, LLMClient
from app.schemas import (
    DateResult,
    DateUpdate,
    DocContent,
    DocList,
    DocSummary,
    FolderSummaryResult,
    FolderSummaryUpdate,
    KbTree,
    KbTreeFile,
    KbTreeFolder,
    KbTreeSource,
    SummaryResult,
    SummaryUpdate,
)

router = APIRouter(tags=["kb"])


def doc_format(path: str, extensions: frozenset[str] = frozenset()) -> str:
    """Lowercased path suffix without its dot (``kubernetes.md`` → ``md``,
    ``notes/deep.Markdown`` → ``markdown``) — returned **unconditionally**
    for a non-empty suffix (display never depends on the import list — an
    out-of-scope ``readme.rst`` still badges ``rst``); ``text`` when the
    path has no suffix **and its name is not a configured token** (phase
    102: a suffix-less ``Dockerfile`` badges ``dockerfile`` when its
    lowercased full filename is one of *extensions* — the importer's
    :func:`app.rag.importer.match_extension` rule). The value shown in
    the viewer's format badge; the default empty *extensions* keeps the
    pre-phase-102 suffix-only result for every path (byte-identical).
    """
    p = Path(path)
    suffix = p.suffix.lower().lstrip(".")
    if suffix:
        return suffix
    matched = match_extension(p, extensions)
    return matched if matched is not None else "text"


@router.get("/docs", response_model=DocList)
def list_indexed_documents(
    db: Session = Depends(get_db),  # noqa: B008
    _admin: None = Depends(require_admin),  # noqa: B008
) -> DocList:
    """All indexed documents with per-document chunk counts.

    Admin-only (phase 16 — the catalog is what the admin sign-in gates;
    the document viewer below is user-gated since phase 79, the shared
    chats being the only anonymous surface). Anonymous callers get 403
    ``admin only`` and the Sources page renders its sign-in gate instead.
    An empty list means the knowledge base has not been imported yet —
    the Sources page renders its designed empty state in that case.
    """
    rows = db.execute(
        select(
            Document.id,
            Document.source,
            Document.path,
            Document.title,
            func.count(Chunk.id).label("chunks"),
            Document.created_at,
            Document.indexed_at,
        )
        .outerjoin(Chunk, Chunk.document_id == Document.id)
        .group_by(
            Document.id,
            Document.source,
            Document.path,
            Document.title,
            Document.created_at,
            Document.indexed_at,
        )
        .order_by(Document.source, Document.path)
    ).all()
    return DocList(
        documents=[
            DocSummary(
                id=str(row.id),
                source=row.source,
                path=row.path,
                title=row.title,
                chunks=row.chunks,
                created_at=row.created_at.isoformat(),
                indexed_at=row.indexed_at.isoformat(),
            )
            for row in rows
        ]
    )


@router.get("/documents/content", response_model=DocContent)
def get_document_content(
    source: str,
    path: str,
    db: Session = Depends(get_db),  # noqa: B008
    _user: None = Depends(require_user),  # noqa: B008  # phase 79: admin or live token
) -> DocContent:
    """Full content of one indexed document, looked up by ``(source, path)``.

    Stateless (A10) and database-only: unknown pairs — including traversal
    strings such as ``../../etc/passwd`` — are just non-existent rows and
    map to 404 ``{detail: "document not found"}``.

    User-gated (phase 79 — SUPERSEDES the phase-16 "deliberately PUBLIC
    (soft rule)" note, owner decision 2026-08-22): the viewer content is
    token-or-admin like the rest of the app surface — chat cites
    documents and a signed-in user (admin or token holder) opens a cited
    document by direct URL. The ONLY anonymous content left is the
    shared chats.
    """
    row = db.execute(
        select(Document, func.count(Chunk.id).label("chunks"))
        .outerjoin(Chunk, Chunk.document_id == Document.id)
        .where(Document.source == source, Document.path == path)
        .group_by(Document.id)
    ).first()
    if row is None:
        raise HTTPException(status_code=404, detail="document not found")
    doc, chunks = row
    return DocContent(
        source=doc.source,
        path=doc.path,
        title=doc.title,
        format=doc_format(doc.path, get_settings().import_extension_set),
        summary=doc.summary,
        created_at=doc.created_at.isoformat(),
        content=doc.content,
        indexed_at=doc.indexed_at.isoformat(),
        chunks=chunks,
    )


@router.patch("/documents/summary", response_model=SummaryResult)
async def update_document_summary(
    payload: SummaryUpdate,
    db: Session = Depends(get_db),  # noqa: B008
    _admin: None = Depends(require_admin),  # noqa: B008
) -> SummaryResult:
    """Update or clear a document's stored summary and re-embed it.

    Admin-only (phase 57, D4) — the document viewer itself stays
    PUBLIC (phase 16 owner decision); only this edit affordance is
    gated. The re-embed scope is the ``is_summary`` chunk only (D4):
    the summary is the only text that changed, so the document's
    content chunks keep their existing embeddings — the total chunk
    count is unchanged by an update.

    Fail-before-write (phase 57 locked decision): when the stripped
    text is non-empty it is embedded **before** any DB mutation — an
    embedding failure returns 503 with a sanitized ``detail`` naming
    the failure (the ``ModelUnavailableError`` handling of
    ``app/api/git_sources.py``) and leaves the row and chunk untouched.
    An empty/whitespace-only ``summary`` clears instead:
    ``documents.summary = NULL`` and the ``is_summary`` chunk (if any)
    is deleted.
    """
    doc = db.scalar(
        select(Document).where(
            Document.source == payload.source, Document.path == payload.path
        )
    )
    if doc is None:
        raise HTTPException(status_code=404, detail="document not found")

    summary_chunk = db.scalar(
        select(Chunk).where(Chunk.document_id == doc.id, Chunk.is_summary.is_(True))
    )
    text = payload.summary.strip()
    if text:
        # Embed first, mutate second — a failed LLM call must never
        # leave a half-updated row (phase 57 locked decision).
        llm = LLMClient()
        try:
            vector = (await llm.embed([text]))[0]
        except EmbeddingError as e:
            raise HTTPException(status_code=503, detail=_sanitize_error(str(e))) from None
        if summary_chunk is None:
            # Markdown doc, or a phase-30 fail-soft import that indexed
            # without a summary chunk — create the position −1 chunk.
            summary_chunk = Chunk(document_id=doc.id, position=-1, is_summary=True)
            db.add(summary_chunk)
        summary_chunk.content = text
        summary_chunk.embedding = vector
        doc.summary = text
    else:
        if summary_chunk is not None:
            db.delete(summary_chunk)
        doc.summary = None
    db.commit()
    chunks = db.scalar(
        select(func.count(Chunk.id))
        .select_from(Document)
        .outerjoin(Chunk, Chunk.document_id == Document.id)
        .where(Document.id == doc.id)
    ) or 0
    return SummaryResult(
        source=doc.source, path=doc.path, summary=doc.summary, chunks=chunks
    )


@router.patch("/documents/date", response_model=DateResult)
def update_document_date(
    payload: DateUpdate,
    db: Session = Depends(get_db),  # noqa: B008
    _admin: None = Depends(require_admin),  # noqa: B008
) -> DateResult:
    """Set or clear the owner's correction of a document's creation date.

    Admin-only (phase 106, D7 — the phase-57 split): the document
    viewer itself stays user-gated (``/documents/content`` — admin OR
    token holder); only this edit affordance is admin-gated. DB-only
    (the ``/documents/content`` row-lookup rule): the ``(source, path)``
    pair is looked up as a row — an unknown pair, including traversal
    strings such as ``../../etc/passwd``, is simply not a row (→ 404
    ``document not found``), and there is no filesystem access. NO
    LLM/embedding call — a date is never embedded (no chunk, no
    retrieval role) — the deliberate contrast with the phase-57
    ``is_summary`` re-embed in :func:`update_document_summary`.

    * **Set** (``date`` present) — ``datetime.fromisoformat`` accepts a
      bare ``YYYY-MM-DD`` (midnight) and full ISO datetimes; a
      MALFORMED value 422s here (the model field is an unconstrained
      ``str | None`` on purpose, so the detail can name the field). The
      parse goes through
      :func:`app.rag.doc_dates.normalize_doc_date` (D3 — the single
      choke point: naive → UTC, aware → converted, a manually set
      FUTURE date also folds to today — consistency with the sourced
      path) and stores ``created_at`` + ``created_at_manual = True``
      (the owner's correction — the sync-time importer then SKIPS the
      refresh on this row, D1/D4).
    * **Clear** (``date`` null/absent — the "revert to sync"
      operation) — ``created_at_manual = False`` ONLY: the stored date
      stands until the next sync refreshes it (the API cannot
      re-read the source, D7).

    The response echoes the stored state — ``created_at`` (ISO-8601)
    + ``created_at_manual`` — the viewer re-renders its Created badge
    from it (no second fetch).
    """
    doc = db.scalar(
        select(Document).where(
            Document.source == payload.source, Document.path == payload.path
        )
    )
    if doc is None:
        raise HTTPException(status_code=404, detail="document not found")
    if payload.date:
        try:
            parsed = datetime.fromisoformat(payload.date)
        except ValueError:
            raise HTTPException(
                status_code=422,
                detail="date must be an ISO date or datetime (e.g. 2024-06-15)",
            ) from None
        doc.created_at = normalize_doc_date(parsed)
        doc.created_at_manual = True
    else:
        # The CLEAR (D7): drop the manual flag only — the stored date
        # stands until the next sync refreshes it.
        doc.created_at_manual = False
    db.commit()
    return DateResult(
        source=doc.source,
        path=doc.path,
        created_at=doc.created_at.isoformat(),
        created_at_manual=doc.created_at_manual,
    )


@router.patch("/folders/summary", response_model=FolderSummaryResult)
def update_folder_summary(
    payload: FolderSummaryUpdate,
    db: Session = Depends(get_db),  # noqa: B008
    _admin: None = Depends(require_admin),  # noqa: B008
) -> FolderSummaryResult:
    """Update / create / clear a folder's stored description.

    Admin-only: the catalog is admin-only (phase 16) and this gate is
    the API-level defense in depth (the RAG view never renders for
    anonymous, the endpoint must not lean on that). ``folder_path =
    ""`` is the SOURCE ROOT (the phase-94 ``folder_summaries``
    convention — the top-level source summary).

    Checks, in order — DB-only (the ``/documents/content`` rule: no
    filesystem access at all, a traversal string such as ``../../etc``
    is simply not a prefix of any indexed path):

    * **Source** — registered (``list_source_names``) OR has indexed
      documents → else 404 ``{"detail": "source not found"}``.
    * **Folder** — ``""`` is valid for an allowed source; otherwise the
      phase-94 existence rule over the source's indexed paths (some
      ``Document.path`` starts with ``folder_path + "/"`` — a file
      merely sharing the folder's name is NOT a folder) → else 404
      ``{"detail": "folder not found"}``.

    Writes: a non-empty (after ``strip()``) ``summary`` upserts the row
    with ``summary = stripped text``, ``manually_edited = True``, and a
    fresh UTC ``updated_at`` — a manual description can be CREATED
    where no row exists (a < 2-document folder, or the generator's
    fail-soft miss), and from this save on the task-01 keep/keep-out
    rules apply (the generator skips the row and never prunes it).
    An empty/whitespace-only ``summary`` CLEARS instead: the row is
    ``db.delete``'d when present (the phase-57 analog — the row may be
    AI-written or manual, either way it is gone; the next KB-changing
    sync regenerates an AI row — the reset path). A clear with no row
    is a 200 no-op. The response echoes the stored text — ``summary``
    null after a clear.

    No LLM/embedding call on this path: a folder description is never
    embedded (no chunk, no retrieval role beyond the ``ls`` line) —
    the deliberate contrast with the phase-57 ``is_summary`` re-embed
    in :func:`update_document_summary`. The no-LLM contract is
    source-pinned in the test suite (the handler's source never names
    the LLM client).
    """
    source = payload.source
    folder = payload.folder_path
    paths = list(db.scalars(select(Document.path).where(Document.source == source)))
    if source not in list_source_names(db) and not paths:
        raise HTTPException(status_code=404, detail="source not found")
    if folder and not any(p.startswith(folder + "/") for p in paths):
        raise HTTPException(status_code=404, detail="folder not found")
    row = db.scalar(
        select(FolderSummary).where(
            FolderSummary.source == source, FolderSummary.folder_path == folder
        )
    )
    text = payload.summary.strip()
    if text:
        if row is None:
            row = FolderSummary(source=source, folder_path=folder)
            db.add(row)
        row.summary = text
        row.manually_edited = True
        row.updated_at = datetime.now(UTC)
    else:
        if row is not None:
            db.delete(row)
    db.commit()
    return FolderSummaryResult(source=source, folder_path=folder, summary=text or None)


#: One catalogue row the tree builder consumes:
#: ``(source, path, title, chunks, indexed_at, created_at)`` — the
#: ``GET /api/docs`` query's columns minus the document ``id`` (the
#: tree has no document ids), in the same ``(source, path)`` order;
#: ``indexed_at`` / ``created_at`` are the ISO-8601 strings the
#: endpoint converts (the builder stays pure over plain types —
#: unit-testable without a DB).
TreeDocRow = tuple[str, str, str, int, str, str]

#: One of a source's file rows, already source-scoped:
#: ``(path, title, chunks, indexed_at, created_at)``.
TreeFileRow = tuple[str, str, int, str, str]


def _folder_counts(
    rows: Sequence[TreeFileRow],
) -> tuple[set[str], dict[str, int]]:
    """One source's folders (existence rule) + recursive counts (pure).

    The phase-94 rules, reused verbatim from
    :func:`app.rag.agent.group_folder_listing` (ONE concept end to end —
    the UI tree is the ``ls`` tree plus file metadata): a folder exists
    ⟺ some indexed path starts with ``folder + "/"`` (a slash-boundary
    prefix of an indexed path — a document's OWN path is never a
    folder; the folder's parent is :func:`folder_of`, the shared
    notion, never re-derived). The count of a folder is its recursive
    subtree — every path EQUAL to the folder (the file sharing its
    name counts, the ``path == folder`` arm) or starting with
    ``folder + "/"`` — exactly the set the sync-time folder summary
    describes.

    Returns ``(folders, counts)`` — the folder-name set and the
    per-folder count map (every folder counts ≥ 1 by construction: its
    own descendants, or the file wearing its name, exist).
    """
    folders: set[str] = set()
    for path, _title, _chunks, _indexed_at, _created_at in rows:
        folder = folder_of(path)
        while folder:
            folders.add(folder)
            folder = folder_of(folder)
    counts: dict[str, int] = {folder: 0 for folder in folders}
    for path, _title, _chunks, _indexed_at, _created_at in rows:
        if path in folders:
            counts[path] += 1
        folder = folder_of(path)
        while folder:
            counts[folder] += 1
            folder = folder_of(folder)
    return folders, counts


def _level_children(
    source: str,
    folder: str,
    folders: set[str],
    counts: dict[str, int],
    rows: Sequence[TreeFileRow],
    summaries: Mapping[tuple[str, str], str],
) -> list[KbTreeFolder | KbTreeFile]:
    """One level's children (pure): subfolders in path order, then the
    direct files in input (catalog) order.

    *folder* is source-relative (``""`` = the source root). A folder
    *sub* appears here iff ``folder_of(sub) == folder`` (a DIRECT
    subfolder) and it exists (the :func:`_folder_counts` set — the
    phase-94 existence rule); a file appears iff
    ``folder_of(path) == folder`` (a DIRECT file). The subfolder
    order is the sorted (path) order and the file order is the input
    (catalog — ``GET /api/docs``) order, both matching
    :func:`app.rag.agent.group_folder_listing` level-for-level; the
    file list is NOT capped (the ``ls`` 50-line cap is a model-context
    budget — the UI is for humans). Recurses one level per call.

    Each folder node also carries the phase-98 D3 ``summary_pending``
    flag: the recursive count ≥ :data:`MIN_DOCS_PER_FOLDER` AND no
    stored ``folder_summaries`` row for ``(source, sub)`` — the same
    rule the source node applies (see :func:`build_kb_tree`).

    And, since phase 106 (D9), each folder node carries ``updated_at``
    = the subtree's MAX document ``created_at``: the max over this
    folder's direct files' dates and its subfolder children's (already
    recursive) ``updated_at`` values, via :func:`_subtree_max`.
    """
    children: list[KbTreeFolder | KbTreeFile] = []
    for sub in sorted(g for g in folders if folder_of(g) == folder):
        sub_children = _level_children(source, sub, folders, counts, rows, summaries)
        children.append(
            KbTreeFolder(
                path=sub,
                documents=counts[sub],
                updated_at=_subtree_max(sub_children),
                summary=summaries.get((source, sub)),
                summary_pending=counts[sub] >= MIN_DOCS_PER_FOLDER
                and (source, sub) not in summaries,
                children=sub_children,
            )
        )
    for path, title, chunks, indexed_at, created_at in rows:
        if folder_of(path) == folder:
            children.append(
                KbTreeFile(
                    path=path,
                    title=title,
                    chunks=chunks,
                    created_at=created_at,
                    indexed_at=indexed_at,
                )
            )
    return children


def _subtree_max(children: Sequence[KbTreeFolder | KbTreeFile]) -> str | None:
    """A node's ``updated_at`` (phase 106, D9): the subtree's MAX
    document ``created_at``, computed from the node's direct children —
    files contribute their ``created_at``, subfolder nodes contribute
    their (already recursive) ``updated_at``.

    All values share the same UTC ``isoformat()`` shape (the endpoint
    converts every ``created_at`` before the builder runs), so the
    LEXICOGRAPHIC max is the chronological max — ISO-8601 strings of
    one offset order by instant. ``None`` when no child carries a date
    (a node with no documents at all — the 0-document source).
    """
    dates = [
        child.created_at if child.kind == "file" else child.updated_at
        for child in children
    ]
    dates = [d for d in dates if d is not None]
    return max(dates) if dates else None


def build_kb_tree(
    names: Sequence[str],
    doc_rows: Sequence[TreeDocRow],
    summaries: Mapping[tuple[str, str], str],
) -> list[KbTreeSource]:
    """The pure tree builder behind ``GET /api/docs/tree`` (phase 97,
    task 02) — module-level and DB-free so unit tests drive it
    directly (the house pattern; the endpoint composes the fetches).

    *names* — the registry source names in order (``app.rag.agent.
    list_source_names`` — deduped, registry order). *doc_rows* — the
    catalogue ``(source, path, title, chunks, indexed_at, created_at)``
    tuples (both stamps ISO-8601) in the ``GET /api/docs`` query order
    (``source, path``). *summaries* —
    ``{(source, folder_path): summary}`` over the stored
    ``folder_summaries`` rows (``folder_path = ""`` = the source root;
    rows for sources the tree does not list are simply never
    referenced).

    Shape, per the phase-97 ``00_phase.md`` "The tree endpoint":

    * **Sources** — the registry names first (each ALWAYS present — a
      registered 0-document source lists with ``documents: 0`` and no
      children), then the distinct indexed sources not in *names*
      (alphabetical — the superset rule: the catalog has never hidden
      an indexed document, while the agent's ``ls`` keeps listing
      registry sources only — unchanged). Every doc source is listed by
      construction; the registry order still leads.
    * **Folder nodes** — ``path`` source-relative (never ``""`` — the
      source node IS the root); direct subfolders only, in path
      (sorted) order, a folder existing only under the phase-94
      existence rule; ``documents`` = the recursive subtree count;
      ``updated_at`` (phase 106, D9) = the subtree's MAX document
      ``created_at`` — DERIVED as the builder recurses (the max of the
      direct files' dates and the children's ``updated_at`` values via
      :func:`_subtree_max`; the ISO-8601 strings share one
      ``isoformat()`` shape, so the lexicographic max is the
      chronological one), never stored — ``null`` when the node has no
      documents at all; ``summary`` = the stored row (AI OR manual —
      any row) or null; ``children`` = the folder's own subfolders +
      direct files, same shape.
    * **File nodes** — direct files only, in input (catalog) order;
      ``path`` source-relative; ``title`` / ``chunks`` / ``created_at``
      (phase 106) / ``indexed_at`` verbatim from the catalogue row.
      File nodes carry NO pending flag (the file table has no
      description column) and no ``updated_at`` (a file's date IS its
      ``created_at``).
    * **Pending** — ``summary_pending`` on the SOURCE and every FOLDER
      node (phase 98, decision D3 — ONE concept): true iff the node's
      recursive ``documents`` count ≥
      :data:`app.rag.folder_summaries.MIN_DOCS_PER_FOLDER` (2) AND it
      has NO stored ``folder_summaries`` row (AI or manual — any row;
      the builder sees stored rows only). That is EXACTLY
      :func:`app.rag.folder_summaries.missing_folder_summaries`'s
      candidate set (phase 96's gap-fill regenerates precisely those
      keys on the next sync — the marker is honest: "waiting to
      generate", and the integration cross-check pins the tree's
      pending set to that function so the marker can never drift from
      the gap-fill). A < 2-document folder is NEVER pending (it never
      gets a summary — its one file line IS its description), and a
      registered 0-document source never is.

    ONE concept end to end: the builder reuses
    :func:`app.rag.folder_summaries.folder_of` and the phase-94
    existence / count rules, so — for a single-source dataset — its
    level equals :func:`app.rag.agent.group_folder_listing`'s output
    (same subfolder ``(path, count, summary)`` triples in order, same
    file ``(source, path, title, date)`` 4-tuples in order — phase
    106, D5: the agent's file lines carry the appended ``date`` field
    and the tree's file nodes carry ``created_at``; the "UI shows
    what the agent sees" cross-check, unit-pinned at the root and a
    nested level, compares the extended shapes).

    D9 (phase 106): ``updated_at`` on every SOURCE and FOLDER node is
    the subtree's MAX document ``created_at`` — derived, never stored;
    ``None`` for a node with no documents at all (the registered
    0-document source).
    """
    by_source: dict[str, list[TreeFileRow]] = {}
    for source, path, title, chunks, indexed_at, created_at in doc_rows:
        by_source.setdefault(source, []).append((path, title, chunks, indexed_at, created_at))
    tree: list[KbTreeSource] = []
    listed: set[str] = set()
    for name in names:
        if name in listed:  # defensive: list_source_names dedupes
            continue
        listed.add(name)
        tree.append(_source_node(name, by_source.get(name, ()), summaries))
    for source in sorted(by_source):
        if source not in listed:
            tree.append(_source_node(source, by_source[source], summaries))
    return tree


def _source_node(
    source: str,
    rows: Sequence[TreeFileRow],
    summaries: Mapping[tuple[str, str], str],
) -> KbTreeSource:
    """One source node (pure): whole-source count + the source-root
    summary + the root level's children (direct subfolders + direct
    files).

    ``documents`` is ``len(rows)`` — the source's WHOLE recursive
    count (every one of its documents, the set its stored
    ``(source, "")`` summary describes). A source with no rows lists
    ``documents: 0`` and no children (the registered 0-document source
    — the phase-70/72 invariant, extended by the superset rule) and
    is never ``summary_pending`` (0 < the minimum).

    ``summary_pending`` (phase 98, D3): the whole-source count ≥
    :data:`MIN_DOCS_PER_FOLDER` AND no stored ``(source, "")`` row —
    the source-root arm of the rule :func:`build_kb_tree` documents.

    ``updated_at`` (phase 106, D9): the source's subtree max —
    :func:`_subtree_max` over the root level's children; ``None`` for a
    0-document source (no children, no dates).
    """
    folders, counts = _folder_counts(rows)
    children = _level_children(source, "", folders, counts, rows, summaries)
    return KbTreeSource(
        name=source,
        documents=len(rows),
        updated_at=_subtree_max(children),
        summary=summaries.get((source, "")),
        summary_pending=len(rows) >= MIN_DOCS_PER_FOLDER and (source, "") not in summaries,
        children=children,
    )


@router.get("/docs/tree", response_model=KbTree)
def list_kb_tree(
    db: Session = Depends(get_db),  # noqa: B008
    _admin: None = Depends(require_admin),  # noqa: B008
) -> KbTree:
    """The full recursive KB tree in ONE fetch (phase 97, task 02).

    Admin-only, like ``GET /api/docs`` — anonymous callers get 403
    ``admin only`` (the RAG view's anonymous gate never fetches the
    tree). The RAG view drills CLIENT-side: this is the view's single
    fetch, zero per-level requests (the ``00_phase.md`` "The tree
    endpoint" contract).

    Composition: the registry source names (``list_source_names`` —
    the superset rule's registry half, imported from ``app.rag.agent``
    exactly as ``app/api/chat.py`` does) + the SAME outerjoin/grouped
    catalogue query ``GET /api/docs`` runs (the document ``id``
    excluded — the tree has no document ids) + ALL stored
    ``folder_summaries`` rows (a bounded select — one row per ≥ 2-doc
    folder; rows for sources the tree does not list are never
    referenced by the builder) — through the pure
    :func:`build_kb_tree`. ``GET /api/docs`` itself is untouched.
    """
    names = list_source_names(db)
    rows = db.execute(
        select(
            Document.source,
            Document.path,
            Document.title,
            func.count(Chunk.id).label("chunks"),
            Document.created_at,
            Document.indexed_at,
        )
        .outerjoin(Chunk, Chunk.document_id == Document.id)
        .group_by(
            Document.id,
            Document.source,
            Document.path,
            Document.title,
            Document.created_at,
            Document.indexed_at,
        )
        .order_by(Document.source, Document.path)
    ).all()
    doc_rows: list[TreeDocRow] = [
        (source, path, title, chunks, indexed_at.isoformat(), created_at.isoformat())
        for source, path, title, chunks, created_at, indexed_at in rows
    ]
    summaries: dict[tuple[str, str], str] = {
        (source, folder_path): summary
        for source, folder_path, summary in db.execute(
            select(FolderSummary.source, FolderSummary.folder_path, FolderSummary.summary)
        ).all()
    }
    return KbTree(sources=build_kb_tree(names, doc_rows, summaries))