brain-of-reese/app/rag/verbatim.py

"""Phase 182 (task 02): the verbatim locator grammar, the slicer, the resolver.

The contract in one line: **the model names a location, this module supplies
the bytes** (PLAN §6's tool-surface extension, phase 182 D1–D7). An answer
never carries a model-retyped copy of a source file; it carries a ``bor``
locator fence (````` ```bor homelab/runbooks/nginx.md#12-40 `````), and the
bytes shown to the human are read out of the indexed ``documents.content``
row by :func:`block_for` / :func:`resolve_answer_blocks`. Because the
displayed text is never generated, it cannot be paraphrased, "corrected",
or hallucinated — the phase's whole point — and the same resolver serves
both entry points (the ``show_text`` tool's own block, task 03, and the
``done``-frame pass over the finished answer, task 05).

The module is deliberately split in two halves:

* the **pure** half — :func:`canonical_locator`, :func:`fences_in`,
  :func:`slice_lines`, :func:`cap_slice` — no DB, no settings, no I/O;
* the **resolver** half — :func:`document_by_identity`, :func:`block_for`,
  :func:`resolve_answer_blocks` — one bare ``SELECT`` per locator and the
  kill-switch gate.

**Import rule (one-way, checkable with
``grep -n "import" app/rag/verbatim.py``):** this module must NEVER import
the agent module (:mod:`app.rag.agent`) — ``agent.py`` already depends on
the pure half of this module for the ``show_text`` tool branch, so a
reverse dependency cycles. Allowed here: ``sqlalchemy``,
:class:`app.models.Document`,
:data:`app.rag.retriever.TRUNCATION_MARKER`, :class:`app.config.Settings`,
stdlib.

Nothing calls this module yet (tasks 03 and 05 wire it in), so landing it
changes no behavior anywhere.
"""

from __future__ import annotations

import re
from collections.abc import Sequence
from dataclasses import dataclass

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.config import Settings
from app.models import Document
from app.rag.retriever import TRUNCATION_MARKER

__all__ = [
    "LOCATOR_LANG",
    "VerbatimBlock",
    "SliceResult",
    "block_for",
    "cap_slice",
    "canonical_locator",
    "document_by_identity",
    "fences_in",
    "resolve_answer_blocks",
    "slice_lines",
]

#: The fence info-string token that marks a locator fence (D2). The
#: markdown renderer (task 06) and the copy-path builder (task 07) key on
#: this SAME constant — one grammar, one spelling, no drifting copy.
LOCATOR_LANG = "bor"

#: The range separators a locator may use between the identity and the
#: line range: a space, a colon, a bare ``#``, or ``#L`` (D7 — the grammar
#: is TOLERANT on input because the string is model-written).
_SEP = r"[ \t:#]+"

#: A line range: ``12-40``, ``12–40`` (en dash), ``L12-L40``, or a bare
#: ``12`` (one line). An optional ``L`` prefixes either bound; whitespace
#: around the dash is tolerated. Anchored at the end of the string by the
#: caller, so a range-shaped tail followed by more text is junk, not a
#: range.
_RANGE = r"(?P<range>L?[0-9]+(?:[ \t]*[-\u2013][ \t]*L?[0-9]+)?)"

#: The trailing range of a locator string (see :data:`_SEP` /
#: :data:`_RANGE`). Case-insensitive only for the ``L`` prefix.
_TRAILING_RANGE_RE = re.compile(f"{_SEP}{_RANGE}$", re.IGNORECASE)

#: The opening line of a ``bor`` locator fence: the language token,
#: whitespace, then a NON-EMPTY remainder (the locator). A bare
#: `` ```bor `` carries no locator, so it never matches and stays an
#: ordinary code block.
_FENCE_OPEN_RE = re.compile(
    rf"^```{re.escape(LOCATOR_LANG)}[ \t]+(?P<locator>.*[^ \t])[ \t]*$"
)

#: The two characters a bound may use as its line-number prefix.
_LINE_PREFIX = "Ll"


@dataclass(frozen=True)
class VerbatimBlock:
    """One displayed verbatim block — the bytes plus the truth about them.

    ``start_line`` / ``end_line`` are **1-based and INCLUSIVE** and describe
    EXACTLY the lines :attr:`text` contains *after* any character cap — a
    capped block's ``end_line`` is the last line actually shown, never the
    line the locator asked for (the caption must not claim lines nobody
    sees). ``total_lines`` is the whole document's line count, so the
    caption can say ``lines 12-40 of 512``. :attr:`text` never carries a
    trailing newline (lines are joined with ``"\\n"``), and
    :attr:`truncated` says whether :data:`~app.rag.retriever.TRUNCATION_MARKER`
    was appended on its own last line.

    :attr:`locator` is the CANONICAL form (D7) built by the factory
    (:func:`block_for`), never passed in by a caller — it is the dedupe key
    of :func:`resolve_answer_blocks` and the matching key the copy-path
    builder uses (task 07).
    """

    source: str
    path: str
    title: str
    start_line: int
    end_line: int
    total_lines: int
    text: str
    truncated: bool = False
    locator: str = ""


@dataclass(frozen=True)
class SliceResult:
    """The outcome of one line-range slice (see :func:`slice_lines`).

    ``first_line`` / ``last_line`` are 1-based and inclusive; ``past_end``
    is the caller's refusal signal and always travels with the true
    ``total_lines`` (the teaching refusal names the real line count).
    """

    text: str
    first_line: int
    last_line: int
    total_lines: int
    past_end: bool


def canonical_locator(raw: str) -> str | None:
    """Normalize one model-written locator string, or refuse it (D7).

    Accepted forms — ``source/path`` (the ``read``/``grep`` identity
    contract, split at the FIRST ``'/'``) followed by an optional separator
    and line range:

    ==========================  ==========================
    input                       canonical output
    ==========================  ==========================
    ``homelab/a.md``            ``homelab/a.md``
    ``homelab/a.md 12-40``      ``homelab/a.md#12-40``
    ``homelab/a.md:12-40``      ``homelab/a.md#12-40``
    ``homelab/a.md#12-40``      ``homelab/a.md#12-40``
    ``homelab/a.md#L12-L40``    ``homelab/a.md#12-40``
    ``homelab/a.md 12\u201340``     ``homelab/a.md#12-40``
    ``homelab/a.md 12``         ``homelab/a.md#12-12``
    ==========================  ==========================

    The canonical form is always ``source/path`` + ``#start-end`` with
    plain digits — never the en dash, never the ``L``, so the fence, the
    tool block and the UI share ONE string (D7).

    Returns ``None`` (NOT a locator) for: empty/whitespace input, a value
    with no ``'/'`` in the path part, a non-numeric or non-positive bound,
    ``start > end``, trailing junk after the range, or a leading ``bor``
    token the caller left in. No DB access, no case folding (paths are
    case-sensitive row values). A locator carrying a range also refuses an
    identity containing whitespace: the fence's info string is ONE
    whitespace-free identity plus the range, so a path that contains a
    space is expressed as a whole-document locator (no range) or not at
    all — the documented limit of the tolerant grammar.
    """
    text = raw.strip()
    if not text:
        return None
    # A caller that pasted the fence's language token in has already lost
    # the identity — refuse rather than guess which token was the path.
    if text.split(" ", 1)[0] == LOCATOR_LANG:
        return None

    match = _TRAILING_RANGE_RE.search(text)
    if match is None:
        identity, range_part = text, None
    else:
        # ``match.start()`` is where the separator begins — the identity
        # is everything before it (a multi-character separator, e.g. two
        # spaces, is consumed whole).
        identity, range_part = text[: match.start()], match.group("range")

    # The identity is one whitespace-free ``source/path`` token, and it
    # never ends in a dangling separator (``a.md#`` with no range is a
    # typo, not a whole document).
    if not identity or any(c.isspace() for c in identity):
        return None
    if identity.endswith(("#", ":", "-")):
        return None
    if "/" not in identity:
        return None
    if range_part is None:
        return identity

    bounds = [b.strip().lstrip(_LINE_PREFIX) for b in re.split(r"[-\u2013]", range_part)]
    # Defensive: :data:`_RANGE` already guarantees digit bounds, so this
    # only fires if the grammar above is ever loosened — it keeps
    # ``canonical_locator`` total (never an ``int()`` ValueError to a
    # caller that hands it model-written text).
    if any(not b.isdigit() for b in bounds):
        return None
    start = int(bounds[0])
    end = int(bounds[1]) if len(bounds) == 2 else start
    if start < 1 or end < 1 or start > end:
        return None
    return f"{identity}#{start}-{end}"


def fences_in(answer: str) -> list[str]:
    """Every ``bor`` locator fence's RAW locator string, in answer order.

    Rules (the same shape ``markdown.js`` applies, task 06):

    * the opening line is a fence whose info string is exactly
      `` ```bor `` + whitespace + a NON-empty remainder — a bare
      `` ```bor `` with no path is an ordinary code block, not a locator,
      and is not swallowed;
    * the fence closes at the next line that is nothing but backticks;
    * an **unclosed** fence contributes nothing (defensive: a half-streamed
      answer must not resolve a locator the model never finished) and ends
      the scan, since its body runs to the end of the answer;
    * the body is read and DISCARDED — this module never looks at what the
      model typed inside the fence (the bytes come from the row, D1);
    * nested `` ``` `` inside a body is NOT supported: the fence ends at
      the first closer (documented limitation, same rule as the renderer).

    Duplicates collapse keeping the FIRST occurrence.
    """
    lines = answer.splitlines()
    locators: list[str] = []
    index = 0
    while index < len(lines):
        match = _FENCE_OPEN_RE.match(lines[index].strip())
        if match is None:
            index += 1
            continue
        closer = _find_fence_close(lines, index + 1)
        if closer is None:
            # Unclosed: the body runs to the end of the answer, so there
            # is nothing further to scan and this fence yields nothing.
            break
        locator = match.group("locator").strip()
        if locator and locator not in locators:
            locators.append(locator)
        index = closer + 1
    return locators


def _find_fence_close(lines: Sequence[str], start: int) -> int | None:
    """Index of the fence's closing line at or after ``start``, ``None`` if unclosed."""
    for offset in range(start, len(lines)):
        stripped = lines[offset].strip()
        if stripped.startswith("```") and not stripped.strip("`"):
            return offset
    return None


def slice_lines(
    content: str, offset: int | None = None, limit: int | None = None
) -> SliceResult:
    """Cut ``content`` to a 1-based line range with ``read``'s exact semantics.

    Same indexing (``content.splitlines()``) and the same range arithmetic
    as the ``read`` tool (``app/rag/agent.py``) and ``grep``, so a line
    number the model got from ``grep`` names the same line here — the one
    shared coordinate system the whole phase depends on:

    * ``offset`` omitted → 1; ``limit`` omitted → read to the end;
    * ``end = min(offset + limit - 1, total_lines)`` — an overshooting
      ``limit`` CLAMPS (``last_line`` is the document's last line);
    * ``offset`` past the end, an empty document, or a ``limit`` below 1 →
      ``past_end=True`` with empty text and the TRUE ``total_lines`` (the
      caller builds the teaching refusal).

    Never raises for a range problem: the refusal is the caller's, the
    total always travels with the result.
    """
    lines = content.splitlines()
    total_lines = len(lines)
    first = 1 if offset is None else offset
    end = total_lines if limit is None else min(first + limit - 1, total_lines)
    if total_lines == 0 or first > total_lines or end < first:
        return SliceResult(
            text="",
            first_line=first,
            last_line=0,
            total_lines=total_lines,
            past_end=True,
        )
    return SliceResult(
        text="\n".join(lines[first - 1 : end]),
        first_line=first,
        last_line=end,
        total_lines=total_lines,
        past_end=False,
    )


def cap_slice(text: str, first_line: int, cap: int) -> tuple[str, int, bool]:
    """Apply the character budget to one slice AT A LINE boundary.

    Returns ``(text_to_show, last_line_shown, truncated)``. The whole
    point of cutting at a line boundary is that the caption's line range
    stays true: ``last_line_shown`` is the last line the reader actually
    sees, so a capped block never claims lines it did not show.

    * ``cap <= 0`` raises :class:`ValueError` — a mis-wired caller must be
      loud. The ``show_text_max_chars == 0`` kill switch is checked by
      :func:`resolve_answer_blocks`, never by reaching here;
    * ``len(text) <= cap`` → the text untouched, its own last line,
      ``truncated=False``;
    * otherwise cut to the last WHOLE line that fits and append the shared
      :data:`~app.rag.retriever.TRUNCATION_MARKER` on its own line — the
      marker is not content and is not counted in ``last_line_shown`` (a
      line counts as fitting only when its terminator is inside the
      budget, or sits exactly at the budget's edge);
    * a first line LONGER than ``cap`` cuts that one line mid-way (the only
      sub-line cut, documented) and still ends with the marker.
    """
    if cap <= 0:
        raise ValueError(f"cap_slice needs a positive cap, got {cap!r}")
    if len(text) <= cap:
        if not text:
            return text, first_line, False
        # ``split("\\n")``, NOT ``splitlines()``: a slice that ends on a
        # BLANK line legitimately ends with ``"\\n"`` (line 5 of
        # ``"a\nb\n\n"`` is empty and present), and ``splitlines()`` drops
        # that final empty element — which would report an extent one
        # line short of the text actually shown (a caption that lies, and
        # a canonical locator that no longer matches the same range
        # resolved by another entry point).
        return text, first_line + len(text.split("\n")) - 1, False
    cut = text[:cap]
    if text[cap] == "\n":
        # The budget lands exactly on a line end — every line of the cut
        # is whole, nothing to trim.
        return (
            f"{cut}\n{TRUNCATION_MARKER}",
            first_line + len(cut.splitlines()) - 1,
            True,
        )
    boundary = cut.rfind("\n")
    if boundary <= 0:
        # No whole line fits — the single documented mid-line cut.
        return f"{cut}\n{TRUNCATION_MARKER}", first_line, True
    whole = cut[:boundary]
    return (
        f"{whole}\n{TRUNCATION_MARKER}",
        first_line + len(whole.splitlines()) - 1,
        True,
    )


def document_by_identity(db: Session, combined: str) -> Document | None:
    """The indexed row at the combined ``source/path`` identity, or ``None``.

    Split at the FIRST ``'/'`` — source names are directory basenames and
    can never contain one, so the part before is the source and everything
    after is the path (the same identity contract as
    ``app.rag.agent._resolve_path``).

    This is that contract **reimplemented as a bare lookup**, deliberately
    NOT a call into ``agent.py``: the tool branch already has
    ``_resolve_path`` (with its did-you-mean suggestions), the fence
    resolver only needs yes/no, and importing the agent from here would
    cycle (``agent.py`` imports this module). A traversal string
    (``../../etc/passwd``) is simply a row that does not exist — there is
    no file access here to guard, only a ``WHERE`` clause.
    """
    if "/" not in combined:
        return None
    source, _, path = combined.partition("/")
    return db.scalar(
        select(Document).where(Document.source == source, Document.path == path)
    )


def block_for(
    db: Session,
    combined: str,
    *,
    offset: int | None = None,
    limit: int | None = None,
    cap: int,
    doc: Document | None = None,
) -> VerbatimBlock | None:
    """Build the one block a locator names, or ``None`` (nothing displays).

    Resolves ``combined`` (or takes the caller's already-resolved ``doc``)
    and refuses ``None`` for: no such document (including every
    traversal-shaped identity), an ``is_image`` document (image display
    belongs to ``show_image`` — LOCKED A6), and an empty or past-the-end
    slice. Otherwise ``slice_lines`` → ``cap_slice`` → the block, with the
    canonical :attr:`~VerbatimBlock.locator` built here (``source/path``,
    plus ``#first-last`` when the request named a range **or** the cap cut
    the slice — the form task 07 matches on).
    """
    if doc is None:
        doc = document_by_identity(db, combined)
    if doc is None or doc.is_image:
        return None
    sliced = slice_lines(doc.content, offset, limit)
    if sliced.past_end or not sliced.text:
        return None
    text, last_line, truncated = cap_slice(sliced.text, sliced.first_line, cap)
    locator = f"{doc.source}/{doc.path}"
    if offset is not None or limit is not None or truncated:
        locator = f"{locator}#{sliced.first_line}-{last_line}"
    return VerbatimBlock(
        source=doc.source,
        path=doc.path,
        title=doc.title,
        start_line=sliced.first_line,
        end_line=last_line,
        total_lines=sliced.total_lines,
        text=text,
        truncated=truncated,
        locator=locator,
    )


def resolve_answer_blocks(
    db: Session,
    answer: str,
    shown: Sequence[VerbatimBlock],
    settings: Settings,
) -> list[VerbatimBlock]:
    """The blocks an answer displays, in DISPLAY ORDER (the done-time pass).

    Fuses the two ways a block enters an answer — the `` ```bor `` locator
    fences the model wrote (resolved here, from the row) and the blocks the
    ``show_text`` tool already showed during the turn (``shown``) — into
    the ONE list task 05 puts on the ``done`` frame and persists with the
    record.

    **Kill switches first** (checked HERE so no caller has to remember
    them): ``[]`` when ``settings.answer_text`` is off, or
    ``settings.show_text_max_blocks == 0``, or
    ``settings.show_text_max_chars == 0``.

    **Ordering rule** (task 07's DOM order and the E2E assertions depend on
    it): answer order first — one block per fence, in the order the fences
    appear — then the ``shown`` blocks the answer never located, in their
    original call order. Dedupe is by canonical locator, and the **fence
    wins** over a same-locator entry from ``shown`` because the fence
    controls placement. A fence that does not resolve (unknown path, image
    document, past-the-end range, malformed locator) is simply absent: the
    caller sees "not among the blocks", which is what task 07 renders as
    the unresolved state — the answer's own fence text is left alone, so
    nothing is silently deleted. The list is finally truncated to
    ``settings.show_text_max_blocks`` — and the fence scan STOPS once that
    many blocks are in, because the same truncation makes every later
    block unreachable (the bound keeps a fence-stuffed answer to at most
    ``show_text_max_blocks`` successful lookups instead of one ``SELECT``
    per fence on the request's event loop).
    """
    if not settings.answer_text:
        return []
    if settings.show_text_max_blocks == 0 or settings.show_text_max_chars == 0:
        return []

    cap = settings.show_text_max_chars
    blocks: list[VerbatimBlock] = []
    #: The locator strings already CLAIMED (canonical written forms) — a
    #: repeated fence never resolves twice, and an unresolvable fence
    #: still holds its slot against a same-locator ``shown`` block.
    claimed: set[str] = set()
    #: The locators of the blocks actually appended — a resolved block
    #: can carry a NARROWER locator than the locator its fence wrote (the
    #: char cap cut the slice), and the same range shown by the tool
    #: carries exactly that narrower key, so both keys dedupe.
    emitted: set[str] = set()

    for raw in fences_in(answer):
        canonical = canonical_locator(raw)
        if canonical is None or canonical in claimed:
            continue
        # Reserve the key BEFORE resolving: an unresolvable fence must not
        # let a same-locator ``shown`` block jump into the fence's slot.
        claimed.add(canonical)
        block = _block_for_canonical(db, canonical, cap)
        if block is None or block.locator in emitted:
            continue
        claimed.add(block.locator)
        emitted.add(block.locator)
        blocks.append(block)
        if len(blocks) >= settings.show_text_max_blocks:
            # The display budget is already spent — the truncation below
            # makes every later block unreachable, so stop LOOKING too:
            # a fence-stuffed answer costs at most
            # ``show_text_max_blocks`` successful lookups instead of one
            # ``SELECT`` per fence on the request's event loop.
            break

    for block in shown:
        if block.locator in emitted or block.locator in claimed:
            continue
        emitted.add(block.locator)
        blocks.append(block)

    return blocks[: settings.show_text_max_blocks]


def _block_for_canonical(db: Session, canonical: str, cap: int) -> VerbatimBlock | None:
    """Resolve one ALREADY-CANONICAL locator (the output of :func:`canonical_locator`)."""
    identity, _, range_part = canonical.partition("#")
    if not range_part:
        return block_for(db, identity, cap=cap)
    start_str, _, end_str = range_part.partition("-")
    start = int(start_str)
    end = int(end_str)
    return block_for(db, identity, offset=start, limit=end - start + 1, cap=cap)