"""Unit: the phase-182 verbatim resolver (``app/rag/verbatim.py``, task 02). The module is the single source of truth for "which exact lines of which indexed document, and what are their bytes", so the suite covers its three layers: the pure grammar (:func:`canonical_locator`, :func:`fences_in`), the pure slicing (:func:`slice_lines`, :func:`cap_slice`), and the resolver (:func:`document_by_identity`, :func:`block_for`, :func:`resolve_answer_blocks`). Session-fixture style follows the agent suite (``tests/unit/test_agent.py``): DETACHED :class:`~app.models.Document` rows plus a stand-in session for the one bare ``(source, path)`` lookup — the resolver's SQL is a single ``WHERE`` on the table's unique pair, so a database adds nothing the contract needs (and the suite runs without the stack up). """ from __future__ import annotations import ast import inspect import re import uuid from typing import Any, cast from unittest.mock import MagicMock import pytest from sqlalchemy import Select from sqlalchemy.orm import Session import app.rag.verbatim as verbatim_module from app.config import Settings from app.models import Document from app.rag.retriever import TRUNCATION_MARKER from app.rag.verbatim import ( LOCATOR_LANG, SliceResult, VerbatimBlock, block_for, canonical_locator, cap_slice, document_by_identity, fences_in, resolve_answer_blocks, slice_lines, ) def _settings(**kwargs: Any) -> Settings: kwargs.setdefault("_env_file", None) return Settings(**kwargs) # pyright: ignore[reportCallIssue] def _doc( source: str = "homelab", path: str = "a.md", title: str = "The Title", content: str = "one\ntwo\nthree", *, is_image: bool = False, ) -> Document: """A DETACHED row — the resolver reads ``source``/``path``/``title``/ ``content``/``is_image`` and nothing else, so no session is needed.""" return Document( id=uuid.uuid4(), source=source, path=path, full_path=f"/srv/sources/{source}/{path}", title=title, content=content, content_hash="0" * 64, is_image=is_image, ) class _FakeDB: """Stand-in session for :func:`document_by_identity`'s single lookup. Records every ``(source, path)`` pair it was asked for, so a test can assert the FIRST-slash split reaches the ``WHERE`` clause verbatim — including the traversal-shaped identity, which is simply a pair no row has. """ def __init__(self, docs: list[Document]) -> None: self.docs = {(d.source, d.path): d for d in docs} self.lookups: list[tuple[str, str]] = [] def scalar(self, stmt: Select[tuple[Document]]) -> Document | None: params = stmt.compile().params pair = (str(params["source_1"]), str(params["path_1"])) self.lookups.append(pair) return self.docs.get(pair) def _session(docs: list[Document]) -> _FakeDB: return _FakeDB(docs) def _real(session: _FakeDB) -> Session: return cast("Session", session) # -------------------------------------------------------------------------- # canonical_locator — the tolerant grammar in, ONE canonical string out # -------------------------------------------------------------------------- def test_whole_document_locator_is_itself() -> None: assert canonical_locator("homelab/a.md") == "homelab/a.md" @pytest.mark.parametrize( "raw", [ "homelab/a.md 12-40", "homelab/a.md:12-40", "homelab/a.md#12-40", "homelab/a.md#L12-L40", "homelab/a.md L12-L40", "homelab/a.md 12\u201340", # en dash, the grep-style range "homelab/a.md 12 - 40 ", # stray spaces around the dash ], ) def test_every_range_spelling_canonicalizes(raw: str) -> None: assert canonical_locator(raw) == "homelab/a.md#12-40" def test_bare_line_number_is_one_line_block() -> None: assert canonical_locator("homelab/a.md 12") == "homelab/a.md#12-12" def test_identity_with_slashes_and_case_survives_untouched() -> None: """Paths are case-sensitive row values; only the FIRST slash splits.""" assert canonical_locator("Homelab/Notes/Deep/a.md") == "Homelab/Notes/Deep/a.md" @pytest.mark.parametrize( "raw", [ "a.md", # no source "", " ", "homelab/a.md 0-5", # non-positive bound "homelab/a.md 40-12", # start > end "homelab/a.md x-y", # non-numeric bound "homelab/a.md 12-40 extra", # trailing junk after the range "bor homelab/a.md", # the caller left the language token in "bor", "homelab/a.md#", # dangling separator, no range "homelab/a.md:", ], ) def test_non_locators_are_refused(raw: str) -> None: assert canonical_locator(raw) is None def test_whitespace_bearing_path_with_a_range_is_refused() -> None: """The documented limit of the tolerant grammar: the fence's info string is ONE whitespace-free identity plus the range.""" assert canonical_locator("homelab/my notes/a.md 12-40") is None # -------------------------------------------------------------------------- # fences_in — the raw locator strings, in answer order # -------------------------------------------------------------------------- def test_single_fence_yields_its_raw_locator() -> None: answer = "Before.\n```bor homelab/a.md#12-40\nanything the model typed\n```\nAfter." assert fences_in(answer) == ["homelab/a.md#12-40"] def test_fences_come_back_in_answer_order_and_deduped_first() -> None: answer = ( "```bor homelab/b.md\nb\n```\n" "words\n" "```bor homelab/a.md\na\n```\n" "```bor homelab/b.md\nb again\n```\n" ) assert fences_in(answer) == ["homelab/b.md", "homelab/a.md"] def test_bare_language_fence_is_an_ordinary_code_block() -> None: """A `` ```bor `` with no path is NOT a locator — and it must not swallow the real fence that follows it.""" answer = "```bor\nplain code\n```\n```bor homelab/a.md\na\n```" assert fences_in(answer) == ["homelab/a.md"] def test_unclosed_fence_contributes_nothing() -> None: """Defensive: a half-streamed answer never resolves a partial locator.""" assert fences_in("intro\n```bor homelab/a.md#12-40\nhalf streamed") == [] def test_non_bor_fences_are_ignored() -> None: answer = "```python\nx = 1\n```\nplain text, no locator" assert fences_in(answer) == [] def test_nested_backticks_end_the_fence_first() -> None: """Documented limitation, the renderer's rule: the fence closes at the FIRST closer, so text after it is only prose unless it opens a fence.""" answer = "```bor homelab/a.md\n```\nor ```bor homelab/b.md``` inline\n```" assert fences_in(answer) == ["homelab/a.md"] def test_indented_fence_opens() -> None: assert fences_in(" ```bor homelab/a.md\na\n ```") == ["homelab/a.md"] # -------------------------------------------------------------------------- # slice_lines — the one shared coordinate system # -------------------------------------------------------------------------- _TWELVE_LINES = "\n".join(f"line {n}" for n in range(1, 13)) def test_slice_whole_document() -> None: result = slice_lines(_TWELVE_LINES) assert result == SliceResult( text=_TWELVE_LINES, first_line=1, last_line=12, total_lines=12, past_end=False ) def test_slice_offset_only_reads_to_the_end() -> None: result = slice_lines(_TWELVE_LINES, offset=10) assert result.text == "line 10\nline 11\nline 12" assert (result.first_line, result.last_line, result.total_lines) == (10, 12, 12) assert result.past_end is False def test_slice_limit_only_starts_at_line_one() -> None: result = slice_lines(_TWELVE_LINES, limit=2) assert result.text == "line 1\nline 2" assert (result.first_line, result.last_line) == (1, 2) def test_slice_offset_and_limit() -> None: result = slice_lines(_TWELVE_LINES, offset=12, limit=1) assert result.text == "line 12" assert (result.first_line, result.last_line) == (12, 12) def test_overshooting_limit_clamps_to_the_total() -> None: result = slice_lines(_TWELVE_LINES, offset=11, limit=100) assert result.text == "line 11\nline 12" assert (result.first_line, result.last_line, result.past_end) == (11, 12, False) def test_offset_past_the_end_refuses_with_the_true_total() -> None: result = slice_lines(_TWELVE_LINES, offset=13, limit=5) assert result.past_end is True assert result.text == "" assert result.total_lines == 12 def test_empty_document_is_past_end() -> None: result = slice_lines("") assert result == SliceResult( text="", first_line=1, last_line=0, total_lines=0, past_end=True ) def test_limit_below_one_is_an_empty_past_end_slice() -> None: """A contract violation from the caller: refuse loudly-by-shape, never raise (the refusal text belongs to the caller).""" result = slice_lines(_TWELVE_LINES, offset=3, limit=0) assert result.past_end is True assert result.text == "" assert result.total_lines == 12 def test_line_numbers_agree_with_splitlines() -> None: """The grep → read(offset) coordinate system: the line ``grep`` calls ``N`` is what ``offset=N, limit=1`` returns, byte for byte.""" content = "# header\n\ntext one\n indented\nlast" lines = content.splitlines() for number, expected in enumerate(lines, start=1): result = slice_lines(content, offset=number, limit=1) assert result.text == expected assert (result.first_line, result.last_line) == (number, number) # -------------------------------------------------------------------------- # cap_slice — the character budget at a LINE boundary # -------------------------------------------------------------------------- def test_cap_under_budget_returns_the_text_untouched() -> None: text, last, truncated = cap_slice("one\ntwo\nthree", 1, 100) assert (text, last, truncated) == ("one\ntwo\nthree", 3, False) def test_cap_under_budget_offsets_from_the_slices_first_line() -> None: _text, last, truncated = cap_slice("ten\neleven", 10, 100) assert (last, truncated) == (11, False) def test_cap_empty_text_reports_the_first_line() -> None: assert cap_slice("", 7, 100) == ("", 7, False) def test_cap_counts_a_slice_that_ends_on_a_blank_line() -> None: """Phase 182 (task 05) defect pin: a slice whose LAST line is blank legitimately ends with ``"\\n"`` (that empty line IS part of the text shown), so the extent counts it. ``splitlines()`` drops the trailing empty element — which made the caption claim one line fewer than the block displays AND made the canonical locator disagree with the same range resolved through any other entry point (the tool branch), so the same block showed twice.""" text, last, truncated = cap_slice("one\ntwo\n", 4, 100) assert (text, last, truncated) == ("one\ntwo\n", 6, False) def test_cap_cuts_on_a_line_boundary_and_appends_the_shared_marker() -> None: text, last, truncated = cap_slice("one\ntwo\nthree", 1, 9) # line 3 is cut assert truncated is True assert text == f"one\ntwo\n{TRUNCATION_MARKER}" # The marker is not content: the range stops at the last WHOLE line. assert last == 2 def test_cap_landing_exactly_on_a_line_end_keeps_that_line() -> None: """The budget's edge IS the line's terminator — the line fits.""" text, last, truncated = cap_slice("one\ntwo\nthree", 1, 7) # exactly "one\ntwo" assert truncated is True assert text == f"one\ntwo\n{TRUNCATION_MARKER}" assert last == 2 def test_cap_never_claims_a_line_whose_terminator_falls_outside_the_budget() -> None: text, last, truncated = cap_slice("1\n22\n333\n4444", 1, 7) # "1\n22\n33" assert truncated is True assert text == f"1\n22\n{TRUNCATION_MARKER}" assert last == 2 def test_cap_shorter_than_the_first_line_cuts_it_mid_way() -> None: """The one documented sub-line cut — and it still ends with the marker.""" text, last, truncated = cap_slice("abcdefghij\nsecond", 4, 4) assert truncated is True assert text == f"abcd\n{TRUNCATION_MARKER}" assert last == 4 @pytest.mark.parametrize("cap", [0, -1]) def test_cap_zero_or_negative_is_a_loud_mis_wiring(cap: int) -> None: with pytest.raises(ValueError, match="cap_slice"): cap_slice("anything", 1, cap) # -------------------------------------------------------------------------- # document_by_identity — the one bare lookup # -------------------------------------------------------------------------- def test_identity_lookup_hits() -> None: doc = _doc() session = _session([doc]) assert document_by_identity(_real(session), "homelab/a.md") is doc assert session.lookups == [("homelab", "a.md")] def test_identity_lookup_misses() -> None: session = _session([_doc()]) assert document_by_identity(_real(session), "homelab/missing.md") is None def test_traversal_shaped_identity_is_simply_a_row_that_does_not_exist() -> None: session = _session([_doc()]) assert document_by_identity(_real(session), "../../etc/passwd") is None assert session.lookups == [("..", "../etc/passwd")] def test_first_slash_splits_identity_when_the_path_has_slashes() -> None: doc = _doc(path="notes/deep/a.md") session = _session([doc]) assert document_by_identity(_real(session), "homelab/notes/deep/a.md") is doc assert session.lookups == [("homelab", "notes/deep/a.md")] def test_identity_without_a_slash_never_touches_the_database() -> None: """A bare source name is never a document (the agent's contract).""" session = _session([_doc()]) assert document_by_identity(_real(session), "homelab") is None assert session.lookups == [] # -------------------------------------------------------------------------- # block_for — the block a locator names # -------------------------------------------------------------------------- def test_block_whole_document() -> None: doc = _doc(content="one\ntwo\nthree") block = block_for(_real(_session([doc])), "homelab/a.md", cap=1000) assert block is not None assert (block.source, block.path, block.title) == ("homelab", "a.md", "The Title") assert (block.start_line, block.end_line, block.total_lines) == (1, 3, 3) assert block.text == "one\ntwo\nthree" assert block.truncated is False assert block.locator == "homelab/a.md" def test_block_with_a_named_range_carries_it_in_the_locator() -> None: doc = _doc(content="\n".join(f"line {n}" for n in range(1, 13))) block = block_for( _real(_session([doc])), "homelab/a.md", offset=12, limit=2, cap=1000 ) assert block is not None assert (block.start_line, block.end_line, block.total_lines) == (12, 12, 12) assert block.text == "line 12" assert block.locator == "homelab/a.md#12-12" def test_block_of_an_image_document_is_refused() -> None: """Image display belongs to ``show_image`` (LOCKED A6).""" doc = _doc(is_image=True) assert block_for(_real(_session([doc])), "homelab/a.md", cap=1000) is None def test_block_of_an_unknown_path_is_refused() -> None: assert block_for(_real(_session([_doc()])), "homelab/nope.md", cap=1000) is None def test_block_of_a_past_end_range_is_refused() -> None: doc = _doc(content="one\ntwo") assert block_for(_real(_session([doc])), "homelab/a.md", offset=9, cap=1000) is None def test_capped_blocks_range_describes_the_lines_actually_shown() -> None: doc = _doc(content="1\n22\n333\n4444") block = block_for(_real(_session([doc])), "homelab/a.md", cap=6) # cuts line 3 assert block is not None assert block.truncated is True assert block.text == f"1\n22\n{TRUNCATION_MARKER}" assert (block.start_line, block.end_line, block.total_lines) == (1, 2, 4) assert block.locator == "homelab/a.md#1-2" def test_block_reuses_a_callers_resolved_document() -> None: """The ``show_text`` branch already resolved the row (with suggestions); the block build must not look it up twice.""" db = MagicMock() block = block_for(cast("Session", db), "homelab/a.md", cap=1000, doc=_doc()) assert block is not None assert block.locator == "homelab/a.md" db.scalar.assert_not_called() # -------------------------------------------------------------------------- # resolve_answer_blocks — the done-time fusion # -------------------------------------------------------------------------- _TEN = "\n".join(f"line {n}" for n in range(1, 11)) def _answer(*locators: str) -> str: return "".join(f"```bor {loc}\nmodel text\n```\n" for loc in locators) def _shown(locator: str, *, first: int = 1, last: int = 3) -> VerbatimBlock: return VerbatimBlock( source="homelab", path="a.md", title="The Title", start_line=first, end_line=last, total_lines=10, text="one\ntwo\nthree", locator=locator, ) def test_no_fences_and_no_shown_blocks_is_empty() -> None: assert resolve_answer_blocks( _real(_session([])), "a plain answer", [], _settings() ) == [] def test_a_fence_resolves_without_any_tool_call() -> None: doc = _doc(content=_TEN) blocks = resolve_answer_blocks( _real(_session([doc])), _answer("homelab/a.md#4-5"), [], _settings(), ) assert len(blocks) == 1 assert blocks[0].locator == "homelab/a.md#4-5" assert blocks[0].text == "line 4\nline 5" assert (blocks[0].start_line, blocks[0].end_line) == (4, 5) def test_shown_blocks_are_appended_when_the_answer_locates_nothing() -> None: blocks = resolve_answer_blocks( _real(_session([])), "no fence here", [_shown("homelab/a.md"), _shown("homelab/a.md#4-6", first=4, last=6)], _settings(), ) assert [b.locator for b in blocks] == ["homelab/a.md", "homelab/a.md#4-6"] def test_fence_wins_over_the_same_locator_from_shown() -> None: """ONE block per canonical locator, placed at the FENCE (the fence controls placement).""" doc = _doc(content=_TEN) shown = _shown("homelab/a.md#2-3", first=2, last=3) blocks = resolve_answer_blocks( _real(_session([doc])), "intro\n" + _answer("homelab/a.md#2-3") + "outro", [shown, _shown("homelab/a.md#2-3", first=2, last=3)], _settings(), ) assert len(blocks) == 1 assert blocks[0].text == "line 2\nline 3" def test_capped_fence_dedupes_against_the_same_capped_shown_block() -> None: """Phase 182 (task 05) defect pin: a char-capped fence carries a NARROWER locator than the one its fence wrote (the cut extent) — and that is exactly the key the ``show_text`` tool's same-range block carries, so the pair is ONE block, placed at the fence.""" doc = _doc(content=_TEN) settings = _settings(show_text_max_chars=20) fenced = resolve_answer_blocks( _real(_session([doc])), _answer("homelab/a.md"), [], settings ) assert len(fenced) == 1 and fenced[0].truncated assert fenced[0].locator == f"homelab/a.md#1-{fenced[0].end_line}" shown = _shown(fenced[0].locator, last=fenced[0].end_line) fused = resolve_answer_blocks( _real(_session([doc])), _answer("homelab/a.md"), [shown], settings ) assert [b.locator for b in fused] == [fenced[0].locator] def test_two_fences_keep_answer_order_then_the_unlocated_shown() -> None: second = _doc(path="b.md", title="B", content="bee\nbee two") first = _doc(path="a.md", content=_TEN) shown_a = _shown("homelab/a.md") shown_c = _shown("homelab/c.md") blocks = resolve_answer_blocks( _real(_session([first, second])), _answer("homelab/b.md", "homelab/a.md#9-10"), [shown_a, shown_c], _settings(), ) # fences first, in answer order — then the shown blocks the answer # never located, in their original call order (the whole-document # ``homelab/a.md`` block is a DIFFERENT canonical locator from the # fenced ``homelab/a.md#9-10``, so it stays unlocated). assert [b.locator for b in blocks] == [ "homelab/b.md", "homelab/a.md#9-10", "homelab/a.md", "homelab/c.md", ] def test_block_budget_truncates_the_list() -> None: doc = _doc(content=_TEN) blocks = resolve_answer_blocks( _real(_session([doc])), _answer("homelab/a.md#1-2", "homelab/a.md#4-5"), [_shown("homelab/a.md#7-8", first=7, last=8)], _settings(show_text_max_blocks=1), ) assert [b.locator for b in blocks] == ["homelab/a.md#1-2"] @pytest.mark.parametrize( "overrides", [ {"answer_text": False}, {"show_text_max_blocks": 0}, {"show_text_max_chars": 0}, {"answer_text": False, "show_text_max_blocks": 0}, {"answer_text": False, "show_text_max_chars": 0}, {"show_text_max_blocks": 0, "show_text_max_chars": 0}, ], ) def test_every_kill_switch_combination_is_inert(overrides: dict[str, Any]) -> None: doc = _doc(content=_TEN) assert ( resolve_answer_blocks( _real(_session([doc])), _answer("homelab/a.md#4-5"), [_shown("homelab/a.md#7-8", first=7, last=8)], _settings(**overrides), ) == [] ) def test_unresolvable_fence_is_simply_absent() -> None: blocks = resolve_answer_blocks( _real(_session([_doc(content=_TEN)])), _answer("elsewhere/nope.md"), [], _settings(), ) assert blocks == [] def test_image_document_fence_is_absent() -> None: blocks = resolve_answer_blocks( _real(_session([_doc(content=_TEN, is_image=True)])), _answer("homelab/a.md#2-3"), [], _settings(), ) assert blocks == [] def test_a_duplicated_fence_appears_once() -> None: doc = _doc(content=_TEN) blocks = resolve_answer_blocks( _real(_session([doc])), _answer("homelab/a.md#2-3") + "words\n" + _answer("homelab/a.md 2-3"), [], _settings(), ) assert [b.locator for b in blocks] == ["homelab/a.md#2-3"] def test_malformed_fence_is_ignored_without_reserving_its_text() -> None: """A locator the grammar refuses never reaches the resolver at all.""" blocks = resolve_answer_blocks( _real(_session([_doc(content=_TEN)])), _answer("homelab/a.md 40-12"), [], _settings(), ) assert blocks == [] def test_a_loosened_grammar_still_refuses_non_numeric_bounds( monkeypatch: pytest.MonkeyPatch, ) -> None: """The defensive branch: the range regex above guarantees digit bounds, but if it is ever loosened, :func:`canonical_locator` still refuses a non-numeric bound instead of raising out of ``int()``.""" monkeypatch.setattr( verbatim_module, "_TRAILING_RANGE_RE", re.compile(r"[ :#](?P[^ ]*)$"), # the separator, then ANY tail ) assert canonical_locator("homelab/a.md:x-y") is None def test_the_character_cap_applies_to_fenced_blocks() -> None: doc = _doc(content=_TEN) blocks = resolve_answer_blocks( _real(_session([doc])), _answer("homelab/a.md"), [], _settings(show_text_max_chars=20), ) assert len(blocks) == 1 assert blocks[0].truncated is True assert blocks[0].text.endswith(TRUNCATION_MARKER) # 20 characters land exactly on the terminator of "line 3", so lines # 1-3 are whole and the caption stops at 3 (line 4 is cut). assert blocks[0].text == f"line 1\nline 2\nline 3\n{TRUNCATION_MARKER}" assert blocks[0].end_line == 3 assert blocks[0].locator == "homelab/a.md#1-3" def test_locator_language_constant_is_shared() -> None: assert LOCATOR_LANG == "bor" def test_module_never_imports_the_agent() -> None: """The one-way import rule, read off the AST (the docstring names the agent on purpose to explain the rule, so a substring check would prove nothing): ``agent.py`` imports this module's pure half; a reverse import would cycle.""" tree = ast.parse(inspect.getsource(verbatim_module)) imports: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.Import): imports.extend(alias.name for alias in node.names) elif isinstance(node, ast.ImportFrom): imports.extend(f"{node.module or ''}.{alias.name}" for alias in node.names) offending = [name for name in imports if name.startswith("app.rag.agent")] assert offending == [] def test_the_fence_scan_stops_once_the_block_budget_is_spent() -> None: """A fence-stuffed answer costs at most ``show_text_max_blocks`` successful lookups: once the display budget is in, the scan stops — the truncation to the same number makes every later block unreachable, and the resolver runs on the request's event loop.""" session = _session([_doc(content=_TEN), _doc(path="b.md", title="B", content="bee\ntwo")]) blocks = resolve_answer_blocks( _real(session), _answer("homelab/a.md", "homelab/b.md", "homelab/a.md#2-3"), [], _settings(show_text_max_blocks=2), ) assert [b.locator for b in blocks] == ["homelab/a.md", "homelab/b.md"] assert len(session.lookups) == 2