brain-of-reese/tests/e2e/test_read_line_ranges.py

"""Phase 131 task 04 E2E (Playwright, mock-only): the read line-ranges'
dedicated story suite — the owner's stuck scenario, end to end.

The whole TODO.md L7 item (``131_read_line_ranges`` task 04 is the
story's gate): with the owner's ``BOR_READ_MAX_CHARS=24000`` in force,
the agent deterministically performs the grep → read(offset/limit)
dance on a ~100 KB document — the grep finds the marker line and
reports its line number, the range read lands exactly that part (the
part the model did NOT have), the SSE ``tool`` frames carry the REAL
``offset``/``limit`` ints, and the final answer is grounded in the
document's tail (the marker line). The second turn pins the other half
of the fix: a plain read of the same large document truncates at
24000, and the surfaced truncation carries the REWRITTEN notice that
teaches the grep → read(offset/limit) workflow (the pre-phase dead-end
wording is gone).

Run in isolation (DB must be up: ``podman compose up -d db``):

    uv run pytest tests/e2e/test_read_line_ranges.py -v --no-cov

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the real
``turbo`` does whatever it does with the tools, while this story's
gate is the deterministic SCRIPTED flows in
``tests/e2e/mock_llm.py`` (the mock is the only E2E lens on the
LLM's context):

* the range turn — ``READ_LINES_TRIGGER``: a user message containing
  ``read the lines`` **and** the system prompt carrying the HIGH
  prompt's ``<tools>`` section. Request 1 greps for
  ``READ_LINES_PATTERN``; request 2 (a ``tool``-role grep result with
  a ``source/path:line:`` match line) reads the match line's document
  with ``offset`` = the parsed line number and ``limit`` =
  ``READ_LINES_LIMIT`` (30); request 3 (the read result present)
  answers ``Found it at line N. <first 80 chars of the read slice>``
  — N the offset the model passed (the phase-94 read-echo
  convention: the RANGE reached the context).
* the plain-read turn — the mock's EXISTING ``READ_CAP_TRIGGER`` flow
  (``read the capped document: read source/path``, the
  ``test_read_truncation_cap.py`` convention) reused on THIS suite's
  large document: the mock ECHOES the ENTIRE read tool result into
  its final answer (the house scripted-turn way of asserting on tool
  results), so the truncated shape — first cap chars +
  ``[…truncated…]`` + the REWRITTEN notice — lands on the rendered
  answer and the suite asserts the notice's presence AND the old
  dead-end's absence through it.

The app under test boots with the OWNER'S cap (the
env-override-for-the-app-under-test pattern — this suite's module app,
as in ``test_read_truncation_cap.py``): ``BOR_READ_MAX_CHARS=24000``
(the owner's own setting, TODO.md L7) — small enough that the ~99 KB
fixture document's plain read truncates deterministically at 24000,
big enough that the 30-line range slice (~1 300 chars) never does.

KB fixture — ONE large deterministic markdown document
(``linedoc/big.md``, ~99 KB): this module generates it — 2 000
numbered filler lines (``Line <i>: …`` — the line numbers are exact)
under 200 ``##`` headings (one heading per 10 lines), with the marker
line ``LINE_MARKER the trap components are documented here`` at the
1 500th numbered line (= line 1 652 of the document — a known line
near ~1500, FAR beyond the 24 000-char cap: the document's first
24 000 chars hold NO marker). The marker's true line is computed from
the very string written to the fixture (``BIG_DOC.splitlines().index(
…) + 1``) — the assertions use that value, never a literal, so the
suite stays true if the fixture is regenerated.

The app under test is mock-calibrated (conftest pattern):
``BOR_RELEVANCE_THRESHOLD=0.30`` + ``BOR_LEXICAL_SUPPORT_FLOOR=0.15``
+ ``BOR_SOURCE_USEFULNESS_FLOOR=0.15`` — both questions' best-chunk
cosine (~0.23 on the mock's token-overlap distribution, the marker
chunk wins) clears the floor with FTS hits (``trap`` / ``components``
in the marker chunk), so every scripted turn stays GROUNDED (the
``<tools>`` section is present), and the read document clears the
usefulness bar into ``done.sources``.

Test → observable mapping (Playwright Mapping Rule; the story is the
owner's TODO item — one Playwright file per story, A16):
1. ``test_range_read_flow_over_sse`` — a question containing
   ``read the lines`` → the SSE ``tool`` frames, IN ORDER: the
   ``grep`` call (arg ``pattern="LINE_MARKER"``), then the ``read``
   call (asserted on the FRAME's args, not a substring: ``path`` =
   the doc's combined ``source/path``, ``offset`` an int == the
   marker's true line, ``limit`` == 30 — the wire's assertion
   surface for the owner's dance); the ``done`` frame is
   non-deflected with the read document cited (phase 119 A1); the
   final answer is the GROUNDED echo — ``Found it at line <N>``
   (N == the marker's true line) + the marker text itself (the range
   reached the model's context); NO truncation-notice dead-end
   (neither the ``[…truncated…]`` marker nor the notice is in the
   answer — the range read did not truncate, so no ``tool_result``
   frame either).
2. ``test_plain_read_truncation_notice_teaches_the_workflow`` — a
   question carrying the mock's existing plain-read trigger (``read
   the capped document: read linedoc/big.md`` — the
   ``READ_CAP_TRIGGER`` flow on this suite's large document) → the
   plain read truncates at 24 000: the SSE ``tool_result`` frame
   (``truncated: true``, ``chars_shown: 24000``, ``chars_total:`` the
   document's true length) lands after the ``tool`` frame and before
   the answer's first delta (the pinned order); the surfaced
   truncation — the mock's echo of the read result, the LLM-side
   surface the notice rides on — carries the REWRITTEN notice
   naming ``offset`` and ``limit`` (the pinned substrings — the
   pre-phase-131 dead-end wording "to locate what you need — grep
   searches the whole document" is gone).
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import time
from collections.abc import Iterator
from pathlib import Path
from typing import Any

import httpx
import pytest
from playwright.sync_api import Locator, Page, expect
from sqlalchemy import select, text

from app.config import Settings as _Settings
from app.db import SessionLocal
from app.models import Document
from app.rag.agent import READ_TRUNCATION_NOTICE
from app.rag.retriever import TRUNCATION_MARKER
from e2e.auth_helpers import login
from e2e.conftest import (
    ADMIN_PASSWORD,
    SESSION_SECRET,
    USE_REAL_LLM,
    _wait_http,
)
from e2e.mock_llm import READ_LINES_LIMIT, READ_LINES_PATTERN

REPO = Path(__file__).resolve().parents[2]

# Phase 79 (task 04, full inventory): the conftest session app owns
# its port in a combined run — this module app binds its own port
# instead (a same-port second uvicorn dies on bind and would drive
# the wrong server). Env-overridable.
APP_PORT = int(os.environ.get("E2E_APP_PORT_READLINES", "8148"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"

#: The owner's read cap (TODO.md L7 — the env under which the owner
#: hit the stuck scenario): the app under test boots with it.
READ_CAP = 24000

SOURCE = "linedoc"  # the local directory's basename = the source name
BIG_REL = "big.md"
BIG_SP = f"{SOURCE}/{BIG_REL}"  # the combined read identity

#: The marker line the flow greps for: the mock's canned pattern
#: (``READ_LINES_PATTERN`` — imported so the mock can never drift
#: from the fixture) + the owner's question wording. The fixture
#: document carries exactly one line containing it — the range
#: read's target; the echo assertion asserts on this very string.
MARKER_LINE_TEXT = f"{READ_LINES_PATTERN} the trap components are documented here"

# The fixture's layout (deterministic — see the module docstring):
# 2 000 numbered lines under 200 headings (10 lines each), the
# marker replacing the 1 500th numbered line.
NUMBERED_LINES = 2000
LINES_PER_BLOCK = 10
MARKER_NUMBERED = 1500

_FILLER_WORDS = [
    "quiet", "mirror", "vault", "raid", "pool", "chain", "slot", "cycle",
    "guard", "probe", "audit", "manifest", "retention", "rotation",
    "label", "window", "image", "snapshot", "archive", "copy", "verify",
    "steady", "order", "exact", "clean", "whole", "night", "weekly",
    "plan", "track", "mark", "note", "file", "page", "entry", "index",
    "column", "row", "cell",
]


def _big_doc() -> str:
    """The ~99 KB fixture document (deterministic — module docstring).

    The H1, then 200 ``##`` headings (one per 10 numbered lines), each
    followed by its 10 numbered lines (``Line <i>: …``) — the marker
    line REPLACES the ``MARKER_NUMBERED``-th numbered line. The line
    numbers are exact: the marker's true line is computed from the
    returned string (``splitlines().index(…) + 1``), never a literal.
    """
    lines: list[str] = ["# Lined document", ""]
    numbered = 0
    wi = 0
    for block in range(1, NUMBERED_LINES // LINES_PER_BLOCK + 1):
        lines.append(f"## Lines of the big lined read {block:04d}")
        for _ in range(LINES_PER_BLOCK):
            numbered += 1
            if numbered == MARKER_NUMBERED:
                lines.append(MARKER_LINE_TEXT)
            else:
                words = _FILLER_WORDS
                lines.append(
                    f"Line {numbered}: "
                    + " ".join(words[wi % len(words) : wi % len(words) + 6])
                )
                wi += 6
    return "\n".join(lines)


BIG_DOC = _big_doc()
BIG_DOC_TOTAL = len(BIG_DOC)
#: The marker's true (1-based) line — computed from the very string
#: written to the fixture (the determinism pin: the assertions use
#: this, never a literal).
MARKER_LINE_NO = BIG_DOC.splitlines().index(MARKER_LINE_TEXT) + 1

# The suite's length contracts (the task's pinned-count assumptions —
# the assertions use ``len(BIG_DOC)`` of the very string written to
# the fixture, and ``synced_kb`` pins the stored content
# byte-identical to it): the document is OVER the owner's cap (the
# plain read truncates), the marker is FAR past the cap (the first
# cap chars hold NO marker — the stuck premise), the marker is unique
# (the grep has exactly one match line), and the range slice is UNDER
# the cap (the range read does not truncate — no ``tool_result``
# frame on the range turn).
assert BIG_DOC_TOTAL > READ_CAP, BIG_DOC_TOTAL
assert MARKER_LINE_TEXT not in BIG_DOC[:READ_CAP], (
    f"the first {READ_CAP} chars must hold no marker"
)
assert BIG_DOC.count(MARKER_LINE_TEXT) == 1
_slice = "\n".join(
    BIG_DOC.splitlines()[MARKER_LINE_NO - 1 : MARKER_LINE_NO - 1 + READ_LINES_LIMIT]
)
assert len(_slice) < READ_CAP, len(_slice)

# The REWRITTEN notice for this fixture's counts (the app.rag.agent
# constant). The bubble renders the raw LLM-side text through
# renderMarkdown (escape-first, markdown.js) — the notice's inline-
# code backticks (`` `offset` `` / `` `limit` ``) become <code>
# elements, so the DOM text_content() carries the notice with the
# backticks stripped (the test_read_truncation_cap convention).
NOTICE = READ_TRUNCATION_NOTICE.format(shown=READ_CAP, total=BIG_DOC_TOTAL)
NOTICE_RENDERED = NOTICE.replace("`", "")
# The pinned substrings (task 04): the notice names BOTH range
# arguments — the workflow teaching.
assert "offset" in NOTICE and "limit" in NOTICE
# The pre-phase-131 dead-end wording (must be GONE from the
# surfaced truncation).
OLD_DEAD_END_1 = "to locate what you need"
OLD_DEAD_END_2 = "grep searches the whole document"

# The scripted turns. Q_RANGE carries the mock's ``READ_LINES_TRIGGER``
# (the range dance); Q_PLAIN carries the mock's existing
# ``READ_CAP_TRIGGER`` (``read the capped document: read
# source/path`` — the plain-read flow's scripted call, the
# test_read_truncation_cap convention) on this suite's large
# document. The shared ``trap components`` words keep both turns
# GROUNDED on the single-document KB (the mock-calibrated gate — see
# the module docstring).
Q_RANGE = f"Read the lines of {BIG_SP} — what are the trap components?"
Q_PLAIN = f"Read the capped document: read {BIG_SP} — the trap components, plain read check"


# --------------------------------------------------------------------------
# Fixtures
# --------------------------------------------------------------------------


@pytest.fixture(scope="module")
def linedoc_dirs(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The story's local-directory source: one host temp dir (the app
    server runs on the same host, so the path is visible to it)
    holding the ONE generated large document. The directory's
    basename is the source name (``kind=local``, phase 38)."""
    root = tmp_path_factory.mktemp("bor_read_lines") / SOURCE
    root.mkdir()
    (root / BIG_REL).write_text(BIG_DOC, encoding="utf-8")
    assert (root / BIG_REL).read_text(encoding="utf-8") == BIG_DOC
    assert not (root / ".git").exists()  # the local-kind story: NOT git
    return root


@pytest.fixture(scope="module")
def app_server(mock_llm: int, linedoc_dirs: Path) -> Iterator[str]:
    """The real app under test — per-module env (the conftest
    pattern, cf. ``test_read_truncation_cap.py``): NO
    ``BOR_GIT_SOURCES`` (the env fallback is git-only — the source
    here is a DB-registered local directory), the mock LLM, the
    mock-calibrated gate (threshold 0.30 + lexical floor 0.15 — the
    conftest pair; the code-default floor 0.35 would refuse boot
    against the 0.30 threshold, phase 112) — and the SUBJECT: the
    owner's cap ``BOR_READ_MAX_CHARS=24000`` (TODO.md L7). The
    session app is never started in this isolated run, so no port
    clash."""
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    env["BOR_ENVIRONMENT"] = "e2e"
    env["BOR_STATIC_DIR"] = str(REPO / "frontend")
    env["BOR_LLM_BASE_URL"] = (
        "https://aipi.reeseapps.com/v1"
        if USE_REAL_LLM
        else f"http://127.0.0.1:{mock_llm}/v1"
    )
    # Mock-calibrated gate (conftest pattern): both scripted turns
    # stay GROUNDED (the questions' best-chunk cosine ~0.23 clears
    # the 0.15 floor with FTS hits — the marker chunk wins).
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    env["BOR_LEXICAL_SUPPORT_FLOOR"] = "0.15"
    # The read document must clear the citation bar into
    # done.sources (phase 119 A1 — the range turn asserts it).
    env["BOR_SOURCE_USEFULNESS_FLOOR"] = "0.15"
    # Phase 67: instant retry waits + the code-default budget (the
    # conftest leak-guard pattern).
    env["BOR_LLM_RETRY_DELAY"] = "0"
    env["BOR_LLM_RETRIES"] = str(_Settings.model_fields["llm_retries"].default)
    env.setdefault(
        "BOR_DATABASE_URL",
        "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese",
    )
    # Phase 16: admin auth must be set or create_app() refuses to boot.
    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
    env["BOR_SESSION_SECRET"] = SESSION_SECRET
    # The repo's .env file carries the owner's settings (the app reads
    # it from cwd) — override the source list with an EMPTY value (the
    # env var beats the .env file): the registry must hold EXACTLY the
    # local directory this suite registers (a leftover env git list
    # would pollute the KB the scripted turns run against).
    env["BOR_GIT_SOURCES"] = ""
    # Leak guards (conftest pattern): an operator's local (gitignored)
    # .env cannot leak corpus-specific settings into the app under
    # test.
    env["BOR_DOCS_REPO"] = ""
    env["BOR_SUGGESTIONS"] = json.dumps(
        _Settings.model_fields["suggestions"].default
    )
    env["BOR_INPUT_PLACEHOLDER"] = _Settings.model_fields["input_placeholder"].default
    env["BOR_FOOTER_TEXT"] = _Settings.model_fields["footer_text"].default
    # THE SUBJECT: the owner's cap (TODO.md L7 — the env under which
    # the owner hit the stuck scenario; the production default of
    # 128 000 stays pinned in tests/unit/test_config.py).
    env["BOR_READ_MAX_CHARS"] = str(READ_CAP)
    proc = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "app.main:app",
         "--host", "127.0.0.1", "--port", str(APP_PORT), "--log-level", "warning"],
        cwd=REPO,
        env=env,
    )
    try:
        _wait_http(f"{APP_URL}/api/health")
        yield APP_URL
    finally:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


@pytest.fixture(scope="module")
def app_url(app_server: str) -> str:
    return app_server


def _truncate_all() -> None:
    """Fresh registry + KB (the E2E isolation pattern): the E2E
    suites share one Postgres, so a leftover git_sources row or
    document would pollute the retrieval the scripted turns run
    against (the single-document KB's gate margins are pinned
    against EXACTLY this one document)."""
    with SessionLocal() as db:
        db.execute(
            text(
                "TRUNCATE chunks, documents, query_log, steering_notes, "
                "kb_overview, git_sources, folder_summaries"
            )
        )
        db.commit()


def _wait_sync_done_http(client: httpx.Client, timeout_s: float = 180.0) -> dict[str, Any]:
    """Poll the (cookie-authenticated) status endpoint until the run
    reaches a terminal state (the test_ls_tree_drilldown pattern, over
    plain httpx — this fixture has no browser page yet)."""
    deadline = time.monotonic() + timeout_s
    body: dict[str, Any] = {}
    while time.monotonic() < deadline:
        r = client.get("/api/sync/status")
        assert r.status_code == 200, r.text
        body = r.json()
        if body["state"] in ("success", "failed"):
            return body
        time.sleep(0.5)
    raise AssertionError(f"sync did not reach a terminal state: {body}")


@pytest.fixture(scope="module")
def synced_kb(app_server: str, linedoc_dirs: Path) -> None:
    """The story's precondition: the one-source KB synced under the
    deterministic mock. Registers the temp directory through the
    authenticated API (the ``test_local_directory_sources.py``
    pattern), runs the REAL in-process sync (``POST /api/sync`` —
    walk → chunk → embed → overview → version bump), pins the stored
    content byte-identical to the fixture string (the ``chars_total``
    every assertion below uses is ``len(BIG_DOC)`` of the very string
    written to the directory)."""
    _truncate_all()
    with httpx.Client(base_url=app_server, timeout=30.0) as client:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, r.text
        r = client.post(
            "/api/git-sources", json={"kind": "local", "path": str(linedoc_dirs)}
        )
        assert r.status_code == 201, r.text
        r = client.post("/api/sync")
        assert r.status_code == 202, r.text
        body = _wait_sync_done_http(client)
    assert body["state"] == "success", body
    detail = body["detail"]
    assert detail["added"] == 1, detail
    assert detail["pruned"] == 0, detail
    # The import stored the fixture string BYTE-IDENTICALLY — the
    # ``chars_total`` every assertion below uses is ``len()`` of the
    # very string written to the directory.
    with SessionLocal() as db:
        stored = db.scalar(
            select(Document.content).where(
                Document.source == SOURCE, Document.path == BIG_REL
            )
        )
        assert stored == BIG_DOC, "stored content drifted from the fixture"


@pytest.fixture(autouse=True)
def _clean(db_ready: None) -> Iterator[None]:
    """Per-test query_log isolation (the KB itself is module-scoped —
    the scripted turns never change it, so the registry persists
    across the tests of this module)."""
    with SessionLocal() as db:
        db.execute(text("TRUNCATE query_log"))
        db.commit()
    yield
    with SessionLocal() as db:
        db.execute(text("TRUNCATE query_log"))
        db.commit()


# --------------------------------------------------------------------------
# Page helpers (the test_read_truncation_cap / test_ls_tree_drilldown
# house patterns)
# --------------------------------------------------------------------------

#: Captures the raw SSE ``data:`` payloads of the /api/chat stream
#: (a response clone read in the background) — wire-level assertions
#: for the ``tool`` / ``tool_result`` frames, independent of the UI
#: rendering.
SSE_HOOK = """
() => {
  if (window.__sseInstalled) return;
  window.__sseInstalled = true;
  window.__sseFrames = [];
  const origFetch = window.fetch;
  window.fetch = async function (...args) {
    const res = await origFetch.apply(this, args);
    try {
      const url = typeof args[0] === 'string' ? args[0] : args[0].url;
      if (url.includes('/api/chat')) {
        res.clone().text().then((bodyText) => {
          for (const block of bodyText.split('\\n\\n')) {
            const line = block.trim();
            if (line.startsWith('data: ')) {
              window.__sseFrames.push(line.slice(6));
            }
          }
        });
      }
    } catch (e) { /* non-clonable responses: ignored */ }
    return res;
  };
}
"""


def _install_page_hooks(page: Page) -> None:
    page.evaluate(SSE_HOOK)


def _frames(page: Page, deadline_s: float = 120.0) -> list[dict]:
    """The SSE frames captured since the last submit (``_submit``
    clears the buffer), once the hook's background read settles. The
    default deadline is generous: the plain-read turn's echo streams
    the whole 24 000-char truncated result (the mock's 12-char /
    0.02 s pacing ≈ 40 s)."""
    deadline = time.monotonic() + deadline_s
    while True:
        raw = page.evaluate("() => window.__sseFrames || []")
        parsed = [json.loads(line) for line in raw if line]
        if any(f.get("type") == "done" for f in parsed):
            return parsed
        if time.monotonic() > deadline:
            raise AssertionError(
                f"SSE hook captured no `done` frame (frames so far: "
                f"{len(parsed)}) — hook install failed?"
            )
        time.sleep(0.05)


def _tool_frames(frames: list[dict]) -> list[dict]:
    return [f for f in frames if f.get("type") == "tool"]


def _result_frames(frames: list[dict]) -> list[dict]:
    return [f for f in frames if f.get("type") == "tool_result"]


def _submit(page: Page, question: str) -> None:
    page.evaluate("window.__sseFrames = []")
    page.fill("#message-input", question)
    page.click("#send-btn")
    # The user bubble lands synchronously with the submit handler.
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)


def _wait_settled(page: Page, timeout_ms: int = 120_000) -> None:
    """The turn is complete: answer text in the bubble, button
    recovered (the phase-48 settle wait; the timeout is a parameter —
    the plain-read turn's 24 000-char echo streams for ~40 s)."""
    expect(page.locator(".msg.brain .bubble").last).not_to_have_text(
        "", timeout=timeout_ms
    )
    expect(page.locator("#send-btn")).to_be_enabled(timeout=timeout_ms)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=timeout_ms)


def _last_brain(page: Page) -> Locator:
    return page.locator(".msg.brain").last


# --------------------------------------------------------------------------
# 1. The range turn: the owner's grep → read(offset/limit) dance on
#    the wire, grounded in the large doc's tail
# --------------------------------------------------------------------------


def test_range_read_flow_over_sse(
    page: Page, app_url: str, synced_kb: None, db_ready: None
) -> None:
    page.set_default_timeout(60_000)
    login(page, app_url, next="/")
    _install_page_hooks(page)

    _submit(page, Q_RANGE)
    _wait_settled(page)

    frames = _frames(page)
    # The owner's dance on the wire — IN ORDER, asserted on the
    # frames' ARGS (not a substring): the grep (the canned pattern),
    # then the range read — path = the doc's combined identity,
    # offset = the marker's TRUE line (the mock parsed it from the
    # grep result's first match line — the mock does the owner's
    # exact dance), limit = READ_LINES_LIMIT. The range fields ride
    # the SAME frame (the phase-131 wire surface — absent keys on
    # every non-range call).
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "grep", "argument": READ_LINES_PATTERN},
        {
            "type": "tool",
            "name": "read",
            "argument": BIG_SP,
            "offset": MARKER_LINE_NO,
            "limit": READ_LINES_LIMIT,
        },
    ], _tool_frames(frames)
    # The range read did NOT truncate (30 lines ≈ 1 300 chars << the
    # 24 000 cap) — it streams NO tool_result frame (one frame = one
    # noteworthy event, the phase-95 additive contract).
    assert _result_frames(frames) == [], _result_frames(frames)
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is False, done
    # The read document is the turn's cited source (phase 119 A1 —
    # done.sources is the READ docs).
    assert any(
        s["path"] == BIG_REL and s["source"] == SOURCE for s in done["sources"]
    ), done["sources"]

    # The GROUNDED echo — the range reached the model's context and
    # landed in the answer: ``Found it at line <N>`` (N == the
    # marker's true line) + the marker text itself (the answer is
    # grounded in the large doc's TAIL — far past the 24 000-char
    # cap a plain read could reach)…
    text = _last_brain(page).locator(".bubble").text_content() or ""
    assert f"Found it at line {MARKER_LINE_NO}." in text, text
    assert MARKER_LINE_TEXT in text, text
    # …and NO truncation-notice dead-end (the pre-phase stuck shape —
    # "the rest is NOT shown, use grep" — is not the answer: the
    # range read landed).
    assert TRUNCATION_MARKER not in text, text
    assert "TRUNCATED —" not in text, text


# --------------------------------------------------------------------------
# 2. The plain-read turn: the truncation still fires at the owner's
#    cap, and the surfaced notice teaches the workflow (the dead-end
#    is gone)
# --------------------------------------------------------------------------


def test_plain_read_truncation_notice_teaches_the_workflow(
    page: Page, app_url: str, synced_kb: None, db_ready: None
) -> None:
    page.set_default_timeout(150_000)
    login(page, app_url, next="/")
    _install_page_hooks(page)

    _submit(page, Q_PLAIN)
    _wait_settled(page)

    frames = _frames(page)
    # The plain read executed — the tool frame carries NO range keys
    # (a plain read: the phase-131 additive fields are ABSENT, never
    # null — the pre-phase frame shape, byte-identical).
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "read", "argument": BIG_SP}
    ], _tool_frames(frames)
    # EXACTLY ONE tool_result frame — the pinned counts (the
    # owner's cap: showing the first READ_CAP of the fixture's true
    # length).
    assert _result_frames(frames) == [
        {
            "type": "tool_result",
            "name": "read",
            "argument": BIG_SP,
            "truncated": True,
            "chars_shown": READ_CAP,
            "chars_total": BIG_DOC_TOTAL,
        }
    ], _result_frames(frames)
    # Frame order: the marker lands AFTER the call is shown and
    # BEFORE the answer's first delta (the phase-37/48 tool-line
    # timing is untouched — the beat between the two is the
    # truncation).
    i_tool = next(i for i, f in enumerate(frames) if f.get("type") == "tool")
    i_result = next(
        i for i, f in enumerate(frames) if f.get("type") == "tool_result"
    )
    i_delta = next(i for i, f in enumerate(frames) if f.get("type") == "delta")
    assert i_tool < i_result < i_delta, (i_tool, i_result, i_delta)
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is False, done

    # The surfaced truncation — the mock's echo of the read result
    # (the LLM-side surface the notice rides on — the house
    # scripted-turn lens): the REWRITTEN notice landed in the model's
    # context and in the rendered answer, naming BOTH range
    # arguments (the workflow teaching: grep reports line numbers,
    # then read with `offset` + `limit`)…
    text = _last_brain(page).locator(".bubble").text_content() or ""
    assert TRUNCATION_MARKER in text, text
    assert NOTICE_RENDERED in text, text
    # …and the pre-phase-131 dead-end wording is GONE.
    assert OLD_DEAD_END_1 not in text, text
    assert OLD_DEAD_END_2 not in text, text