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

"""Phase 167 task 06 E2E (Playwright, mock-free): the Sources view loads a
SMALL catalog, drills it level by level, and revalidates instead of
re-downloading.

The dedicated story suite for ``167_api_payload_diet`` (``TODO.md:5`` —
"GET /api/docs/tree loads 2 MB in production which causes long loading
times"): the owner-visible half of the payload diet, pinned in a real
browser against the real app. Every claim here is the browser's own
observable — the request log, the parsed response body, the wire headers,
the rendered rows — never a server-side helper.

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

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

Mock-only suite (the phase-97 convention): no chat, no sync, no vision
call happens on any path below — the mock LLM server is started ONLY
because ``create_app`` wants a reachable ``BOR_LLM_BASE_URL`` at boot.
``E2E_REAL_LLM`` is therefore irrelevant here.

KB fixture — seeded STRAIGHT INTO THE DB (the ``test_prose_word_wrap.py``
pattern: the tree is database-only, so no sync and no LLM are involved),
after a TRUNCATE (the ``test_kb_tree.py`` isolation pattern — this suite
asserts exact counts, and the E2E Postgres is shared). Eight documents
across THREE sources, twelve chunks:

* ``atlas`` (5) — the deep chain the drill walks: ``index.md`` at the
  root, ``alpha/`` → ``beta/`` → ``gamma/`` with one document at every
  level, plus the image document ``alpha/pic.png`` (real PNG bytes in a
  scratch dir, so the thumbnail renders through the NODE route and not
  just the full tree);
* ``beacon`` (2) — ``top.md`` + ``notes/short.md``, with NO stored
  description anywhere: the pending-marker source;
* ``cobalt`` (1) — one flat file: the level with no folders at all.

Three stored ``folder_summaries`` rows (the descriptions are long on
purpose — the pre-phase 2 MB payload was mostly inlined description
text, and the gzip assertion below needs a body over the threshold): the
``atlas`` root, ``atlas/alpha/beta`` (a level reached ONLY through a
node fetch), and ``cobalt``.

Dates are fixed, and deliberately NOT monotone with depth: the newest
document in ``atlas`` is the ROOT-level ``index.md``, while the deepest
file (``alpha/beta/gamma/three.md``) is the newest one INSIDE ``alpha``
— so the truncated ``alpha`` row's Updated cell proves phase 106 D9 (the
recursive max-date map) survives the depth limit, and the atlas row's
proves it does not leak downward.

Test → observable mapping (Playwright Mapping Rule):
1. ``test_first_boot_is_one_small_tree_request`` — THE TODO REPRO: the
   Sources boot issues EXACTLY ONE ``/api/docs/tree`` request (and zero
   ``/api/docs/tree/node``), its query is ``depth=1``, its body carries
   ``totals`` (documents 8 / chunks 12 / a last_indexed stamp), every
   source node's ``documents`` count is RECURSIVE (atlas 5 with two
   levels of files it does not list), no node deeper than level 1
   exists, and every level-1 folder reports ``truncated: true``. The
   stat cards read ``totals`` (not a client-side walk of the partial
   tree).
2. ``test_tree_response_arrives_gzipped`` — the wire half of the fix:
   the tree response arrives ``Content-Encoding: gzip`` with
   ``Accept-Encoding`` in ``Vary`` (Chromium advertises gzip natively,
   so this is the real negotiation, not a stub), and the uncompressed
   body is over the ``BOR_COMPRESS_MIN_BYTES`` threshold the app boots
   with (the assertion could never pass vacuously).
3. ``test_drill_into_a_folder_fetches_that_level_once`` — clicking a
   source costs NOTHING (the depth-1 payload holds that level);
   clicking the truncated ``alpha`` folder costs EXACTLY ONE
   ``/api/docs/tree/node?source=atlas&path=alpha``; the level renders
   what the pre-phase full tree rendered — the recursive count and the
   recursive D9 date on the truncated ``beta`` row, the stored
   description, the file rows with their Created dates, the image
   thumbnail through the node route — and the phase-156 bounded scroll
   panel + sticky header survive the lazy render.
4. ``test_two_levels_deep_then_back_uses_the_cache`` — ``alpha`` (fetch
   1) → ``beta`` (fetch 2, the level block shows the description that
   only a node fetch can deliver) → the ``alpha`` crumb: ZERO new
   requests and the rows re-render from what the client holds → the
   ``Knowledge base`` crumb: still zero → re-entering ``alpha``: still
   two fetches in total (``levelCache``, never a refetch).
5. ``test_pending_folder_marker_survives_the_lazy_level`` — the
   description-less ``beacon`` source and its ``notes`` folder render
   the phase-98 D4 "Summary pending" marker (class + text + tooltip —
   text and color, never color alone) in a level reached by drilling.
6. ``test_reload_revalidates_to_304`` — a document reload sends
   ``If-None-Match`` and the WIRE answers ``304`` with a header-sized
   transfer while the table still renders (the browser serves the
   cached body). Chromium hands ``fetch`` the cached 200 — Playwright's
   response event never sees the 304 — so the proof rides the CDP
   Network events (the same two facts a DevTools Network panel shows).
7. ``test_view_switch_revalidates_too`` — the same cache hit through the
   SPA: History → Sources re-issues the tree request with
   ``If-None-Match`` and gets a 304 (the mount-once view's refresh hook
   still asks the server, and the server answers cheaply).
8. ``test_an_edit_busts_the_tag_so_fresh_content_lands`` — the never-stale
   half (PLAN §7.4): a folder description saved through the admin API
   changes the body, so the reload's conditional request answers 200
   with the new payload and the new text renders on the row.
9. ``test_vanished_folder_resets_to_the_top_level`` — a node stub that
   answers 404 (the folder was pruned under the open view) is the honest
   reset: the breadcrumb is gone, the source rows render, no stale
   ``alpha`` rows and no loading row left behind.
10. ``test_failed_drill_keeps_the_level_and_retry_recovers`` — a 500
    node stub: the ``role="alert"`` line names the folder and the real
    Retry button sits beside it, the level the user was ON stays
    rendered (rows + breadcrumb untouched); Retry (the stub lifted)
    loads the level.
11. ``test_in_flight_drill_marks_the_link_and_ignores_a_second_click`` —
    with the node request held by the harness: the clicked link carries
    ``aria-busy`` + ``aria-disabled``, the ONE ``role="status"``
    "Loading alpha…" line sits in the panel, a second click issues NO
    second request, and releasing the held request renders the level.
"""
from __future__ import annotations

import base64
import json
import os
import subprocess
import sys
import time
from collections.abc import Iterator
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, urlparse

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

from app.config import Settings as _Settings
from app.db import SessionLocal
from app.models import Chunk, Document, FolderSummary
from e2e.auth_helpers import login
from e2e.conftest import (
    ADMIN_PASSWORD,
    SESSION_SECRET,
    _wait_http,
)

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

# The conftest session app owns 8123 in a combined run — this module app
# binds its own port (the test_kb_tree.py pattern). Env-overridable.
APP_PORT = int(os.environ.get("E2E_APP_PORT_KBLAZY", "8185"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"

TREE_PATH = "/api/docs/tree"
NODE_PATH = "/api/docs/tree/node"

# --------------------------------------------------------------------------
# The seeded catalogue (fixed IDs-by-name, fixed dates, fixed counts)
# --------------------------------------------------------------------------

ATLAS = "atlas"
BEACON = "beacon"
COBALT = "cobalt"

ATLAS_INDEX = "index.md"
ALPHA_ONE = "alpha/one.md"
ALPHA_PIC = "alpha/pic.png"
BETA_TWO = "alpha/beta/two.md"
GAMMA_THREE = "alpha/beta/gamma/three.md"
BEACON_TOP = "top.md"
NOTES_SHORT = "notes/short.md"
COBALT_SOLO = "solo.md"

#: ``(source, path, title, created_at day-of, chunks)`` — the whole
#: catalogue in ONE table so the counts below cannot drift from the
#: seed. The dates are NOT monotone with depth (see the docstring).
SEED_DOCS: tuple[tuple[str, str, str, str, int], ...] = (
    (ATLAS, ATLAS_INDEX, "Atlas Index", "2024-07-01", 2),
    (ATLAS, ALPHA_ONE, "Atlas Alpha One", "2024-02-11", 1),
    (ATLAS, ALPHA_PIC, "Atlas Alpha Pic", "2024-02-20", 1),
    (ATLAS, BETA_TWO, "Atlas Beta Two", "2024-05-05", 3),
    (ATLAS, GAMMA_THREE, "Atlas Gamma Three", "2024-06-06", 1),
    (BEACON, BEACON_TOP, "Beacon Top", "2024-03-03", 2),
    (BEACON, NOTES_SHORT, "Beacon Notes Short", "2024-04-04", 1),
    (COBALT, COBALT_SOLO, "Cobalt Solo", "2024-01-01", 1),
)

TOTAL_DOCS = len(SEED_DOCS)
TOTAL_CHUNKS = sum(row[4] for row in SEED_DOCS)

#: ``source -> (recursive document count, the subtree MAX created_at)``.
#: ``alpha``'s max is the DEEPEST file's date — the recursive map, not
#: the level's own files; ``atlas``'s max is its root file.
SOURCE_COUNTS: dict[str, tuple[int, str]] = {
    ATLAS: (5, "2024-07-01"),
    BEACON: (2, "2024-04-04"),
    COBALT: (1, "2024-01-01"),
}

#: The stored descriptions. The length is load-bearing: the pre-phase
#: 2 MB payload was mostly inlined folder text, and test 2 needs a
#: depth-1 body over the compression threshold.
ATLAS_ROOT_SUM = (
    "Atlas is the deep fixture source: the index note sits at the root, "
    "one folder chain runs three levels down to gamma, and a picture "
    "documents the alpha level alongside its notes."
)
BETA_SUM = (
    "Beta is the second level of the atlas chain. It holds the two.md "
    "note and the gamma folder below it, and this description reaches the "
    "view through the per-level node route alone."
)
COBALT_SUM = (
    "Cobalt is the single-file source: one flat document, no folders at "
    "all, kept to pin the level that renders no folder rows."
)

#: The image document's summary (the vision description the importer
#: stores) — the thumbnail's alt text.
PIC_DESCRIPTION = (
    "A one-pixel PNG fixture that pins the image-document thumbnail "
    "through the lazily loaded level."
)

#: The D4 pending tooltip (the phase-98 copy, verbatim).
PENDING_TITLE = "No stored description yet — the next sync will generate one."

#: The owner edit that busts the ETag (test 8) — distinctive enough to
#: never match a seeded row.
ALPHA_SUMMARY_EDIT = "Owner note: alpha holds the first level of the atlas chain. (RESE-167-01)"

PNG_1X1 = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="
)


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


@pytest.fixture(scope="module")
def app_server(mock_llm: int, tmp_path_factory: pytest.TempPathFactory) -> Iterator[str]:
    """The real app under test, on its own port.

    The conftest leak-guard pattern throughout: an operator's local
    ``.env`` cannot reach the app under test (every value the suite
    asserts on is pinned from the code default or the E2E constant).
    ``compress_min_bytes`` is pinned to the CODE DEFAULT (test 2 asserts
    the depth-1 body clears it, so gzip is the production setting, not
    a harness gift).
    """
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    env["BOR_ENVIRONMENT"] = "e2e"
    env["BOR_STATIC_DIR"] = str(REPO / "frontend")
    env["BOR_LLM_BASE_URL"] = f"http://127.0.0.1:{mock_llm}/v1"
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    env["BOR_LEXICAL_SUPPORT_FLOOR"] = "0.15"
    env["BOR_SOURCE_USEFULNESS_FLOOR"] = "0.15"
    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"
    )
    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
    env["BOR_SESSION_SECRET"] = SESSION_SECRET
    env["BOR_GIT_SOURCES"] = ""
    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
    env["BOR_COMPRESS_MIN_BYTES"] = str(_Settings.model_fields["compress_min_bytes"].default)
    # The image bytes live in this module's scratch dir (the seeded
    # image_path must point at bytes the public route serves).
    image_dir = tmp_path_factory.mktemp("bor_kb_lazy_images")
    env["BOR_IMAGE_DIR"] = str(image_dir)
    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_kb() -> None:
    """The E2E isolation pattern (``test_kb_tree.py``): the suites share
    one Postgres, and this suite pins exact counts + totals, so the
    catalogue starts empty (a leftover source would move the top-level
    rows and the stat cards)."""
    with SessionLocal() as db:
        db.execute(
            text(
                "TRUNCATE chunks, documents, query_log, steering_notes, "
                "kb_overview, git_sources, folder_summaries"
            )
        )
        db.commit()  # without the commit the TRUNCATE rolls back


def _stamp(day: str) -> datetime:
    """The fixed UTC stamp for a seeded date (``indexed_at`` rides the
    same day, one hour later, so the totals' last_indexed is stable)."""
    moment = datetime.fromisoformat(day).replace(tzinfo=UTC)
    return moment


@pytest.fixture(scope="module")
def seeded_kb(
    app_server: str, tmp_path_factory: pytest.TempPathFactory
) -> Iterator[dict[str, Any]]:
    """The catalogue above, written straight through the ORM (the tree
    is database-only: no sync, no chunks-of-truth, no LLM). The image
    document's ``image_path`` points at a real PNG inside the app's own
    ``BOR_IMAGE_DIR``, and the fixture PROVES the public bytes route
    serves it before any render assertion runs. Yields the identity map
    (the document ids the assertions need), and truncates on teardown."""
    try:
        body = httpx.get(f"{APP_URL}/api/health", timeout=5).json()
    except Exception:  # noqa: BLE001 — unreachable is the skip case
        body = {"db": "down"}
    if body.get("db") != "up":
        pytest.skip("Postgres not reachable — run `podman compose up -d db` first")

    _truncate_kb()
    scratch = tmp_path_factory.mktemp("bor_kb_lazy_png")
    png = scratch / "pic.png"
    png.write_bytes(PNG_1X1)

    ids: dict[str, str] = {}
    last_indexed = ""
    with SessionLocal() as db:
        for source, path, title, day, chunks in SEED_DOCS:
            created = _stamp(day)
            indexed = created.replace(hour=min(23, created.hour + 1))
            is_image = path == ALPHA_PIC
            doc = Document(
                source=source,
                path=path,
                full_path=f"/fixture/{source}/{path}",
                title=title,
                content=PIC_DESCRIPTION if is_image else f"# {title}\n\nBody of {path}.\n",
                summary=PIC_DESCRIPTION if is_image else None,
                is_image=is_image,
                image_path=str(png) if is_image else None,
                content_hash=f"{abs(hash((source, path))):064x}"[:64],
                created_at=created,
                indexed_at=indexed,
            )
            db.add(doc)
            db.flush()
            ids[path] = str(doc.id)
            for position in range(chunks):
                db.add(
                    Chunk(
                        document_id=doc.id,
                        position=position,
                        content=f"{title} chunk {position}",
                    )
                )
            if indexed.isoformat() > last_indexed:
                last_indexed = indexed.isoformat()
        db.add_all(
            [
                FolderSummary(
                    source=ATLAS, folder_path="", summary=ATLAS_ROOT_SUM, manually_edited=False
                ),
                FolderSummary(
                    source=ATLAS,
                    folder_path="alpha/beta",
                    summary=BETA_SUM,
                    manually_edited=False,
                ),
                FolderSummary(
                    source=COBALT, folder_path="", summary=COBALT_SUM, manually_edited=False
                ),
            ]
        )
        db.commit()
    assert len(ids) == TOTAL_DOCS

    served = httpx.get(f"{APP_URL}/api/documents/{ids[ALPHA_PIC]}/image", timeout=10)
    assert served.status_code == 200, "the seeded image bytes must serve through the route"
    assert served.headers["content-type"] == "image/png"

    yield {"ids": ids, "last_indexed": last_indexed, "png": str(png)}

    with SessionLocal() as db:  # the shared Postgres leaves clean for the next suite
        db.execute(
            text(
                "TRUNCATE chunks, documents, query_log, steering_notes, "
                "kb_overview, git_sources, folder_summaries"
            )
        )
        db.commit()


# --------------------------------------------------------------------------
# Page + wire helpers
# --------------------------------------------------------------------------


def _folder_rows(page: Page) -> Locator:
    """The catalog's ONE folder table: the source rows at the top
    level, the subfolder rows at every deeper level. The in-flight
    loading/alert line shares that tbody and is NOT a catalog row, so
    every row-count claim excludes it."""
    return page.locator("#folders-tbody tr:not(.kb-level-loading)")


def _file_rows(page: Page) -> Locator:
    return page.locator("#docs-tbody tr")


def _folder_link(page: Page, name: str) -> str:
    return f'#folders-tbody a.folder-link:text-is("{name}")'


def _crumb_link(_page: Page, name: str) -> str:
    """The breadcrumb step back up to ``name`` ("Knowledge base" for the
    top level) — a crumb click is a client-side navigation."""
    return f'#kb-crumb a.kb-crumb-link:text-is("{name}")'


def _drill(page: Page, *names: str) -> None:
    """Walk one level per click (a source row, then the folder links).
    Since phase 167 an UNCACHED level click triggers exactly one node
    fetch, so the click is followed by the render it produced."""
    for name in names:
        page.click(_folder_link(page, name))
        expect(page.locator(".kb-level-loading")).to_have_count(0)


def _wait_top_level(page: Page) -> None:
    """The boot's first paint: every seeded source is a row (the ONE
    depth-1 fetch settled) and no breadcrumb shows."""
    expect(_folder_rows(page)).to_have_count(len(SOURCE_COUNTS), timeout=30_000)
    expect(page.locator("#kb-crumb")).to_be_hidden()


def _paths_of(requests: list[str], path: str) -> list[str]:
    """The logged request URLs whose PATH is exactly ``path`` — the
    exact-path match is the point: ``/api/docs/tree/node`` must never
    be counted as a tree request, and vice versa."""
    return [url for url in requests if urlparse(url).path == path]


class Wire:
    """The wire truth behind the catalogue fetches, straight off the CDP
    Network domain.

    WHY CDP: Chromium's HTTP cache adds ``If-None-Match`` ITSELF and, on
    a 304, hands ``fetch`` the cached 200 body — so Playwright's response
    event reports 200 and its request headers never carry the conditional
    header. The DevTools Network panel shows the 304; ``Network.enable``
    is that same panel, and it is the only honest way to pin "the second
    visit costs a 304 with no body" in a browser test.

    CDP names the URL on ONE event only (``requestWillBeSent``) and the
    cache layer's own events (the conditional headers, the wire response,
    the transfer size) can land BEFORE it — so every event that arrives
    without a URL stages itself by ``requestId`` and merges into the
    entry the moment its URL is known. Order never matters, and an
    unrelated request's stage is simply never adopted.
    """

    _BLANK: dict[str, Any] = {
        "request_id": "",
        "url": "",
        "path": "",
        "query": "",
        "if_none_match": None,
        "status": None,
        "etag": None,
        "content_encoding": None,
        "encoded_bytes": 0,
    }

    def __init__(self, page: Page) -> None:
        self.entries: dict[str, dict[str, Any]] = {}
        self.order: list[str] = []
        self.stash: dict[str, dict[str, Any]] = {}
        self.cdp = page.context.new_cdp_session(page)
        self.cdp.on("Network.requestWillBeSent", self._on_request)
        self.cdp.on("Network.requestWillBeSentExtraInfo", self._on_request_extra)
        self.cdp.on("Network.responseReceivedExtraInfo", self._on_response_extra)
        self.cdp.on("Network.loadingFinished", self._on_loading_finished)
        self.cdp.send("Network.enable")

    def _slot(self, request_id: str) -> dict[str, Any]:
        """The entry for a request, from the adopted table or the stage."""
        return self.entries.get(request_id) or self.stash.setdefault(
            request_id, dict(self._BLANK, request_id=request_id)
        )

    def _track(self, request_id: str, url: str) -> dict[str, Any]:
        entry = self.entries.get(request_id)
        if entry is None:
            entry = dict(self._slot(request_id), url=url, path=urlparse(url).path,
                         query=urlparse(url).query)
            self.stash.pop(request_id, None)
            self.entries[request_id] = entry
            self.order.append(request_id)
        return entry

    def _on_request(self, params: dict[str, Any]) -> None:
        url = params["request"]["url"]
        if urlparse(url).path in (TREE_PATH, NODE_PATH):
            self._track(params["requestId"], url)

    def _on_request_extra(self, params: dict[str, Any]) -> None:
        headers = {k.lower(): v for k, v in params.get("headers", {}).items()}
        if not headers.get("if-none-match"):
            return
        self._slot(params["requestId"])["if_none_match"] = headers["if-none-match"]

    def _on_response_extra(self, params: dict[str, Any]) -> None:
        headers = {k.lower(): v for k, v in params.get("headers", {}).items()}
        entry = self._slot(params["requestId"])
        entry["status"] = params["statusCode"]
        entry["etag"] = headers.get("etag")
        entry["content_encoding"] = headers.get("content-encoding")

    def _on_loading_finished(self, params: dict[str, Any]) -> None:
        """The bytes the network MOVED for a tracked request — a 304
        costs its headers only, the payload stays in the cache."""
        entry = self.entries.get(params["requestId"])
        if entry is not None:
            entry["encoded_bytes"] += int(params.get("encodedDataLength") or 0)

    def of(self, path: str) -> list[dict[str, Any]]:
        return [self.entries[rid] for rid in self.order if self.entries[rid]["path"] == path]

    def settled(self, path: str) -> list[dict[str, Any]]:
        """Only the entries whose wire response has ARRIVED — the
        assertion never races the network."""
        return [e for e in self.of(path) if e["status"] is not None]


def _admin_client() -> httpx.Client:
    """A cookie-authenticated admin client on the module app (the
    ``test_kb_tree.py`` pattern) — the API half of the never-stale
    proof (an edit that changes the tree's bytes)."""
    client = httpx.Client(base_url=APP_URL, timeout=30.0)
    r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
    assert r.status_code == 204, r.text
    return client


def _body_of(responses: list[Response], path: str) -> dict[str, Any]:
    """The parsed JSON body of the ONE response whose path is ``path`` —
    asserted on the PARSED body, never on bytes."""
    matches = [r for r in responses if urlparse(r.url).path == path]
    assert len(matches) == 1, [r.url for r in responses]
    return matches[0].json()  # type: ignore[Any]


def _walk(nodes: list[dict[str, Any]]) -> Iterator[dict[str, Any]]:
    for node in nodes:
        yield node
        yield from _walk(node.get("children") or [])


# --------------------------------------------------------------------------
# 1. The first load is small: one depth-1 request, totals, truncated names
# --------------------------------------------------------------------------


def test_first_boot_is_one_small_tree_request(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)

    responses: list[Response] = []
    requests: list[str] = []
    page.on("request", lambda r: requests.append(r.url))
    page.on("response", lambda r: responses.append(r))
    login(page, APP_URL)
    _wait_top_level(page)

    # THE TODO L5 REPRO, pinned: exactly ONE catalogue request, and it is
    # the DEPTH-LIMITED one. No per-level node fetch happens unprompted.
    tree_requests = _paths_of(requests, TREE_PATH)
    assert len(tree_requests) == 1, tree_requests
    assert _paths_of(requests, NODE_PATH) == []
    assert parse_qs(urlparse(tree_requests[0]).query) == {"depth": ["1"]}, tree_requests[0]

    body = _body_of(responses, TREE_PATH)

    # The server-computed totals: the stat cards' only source of truth,
    # because a depth-limited tree cannot be summed client-side.
    totals = body["totals"]
    assert totals["documents"] == TOTAL_DOCS, totals
    assert totals["chunks"] == TOTAL_CHUNKS, totals
    assert totals["last_indexed"], totals

    # The top level is the sources, with RECURSIVE counts — atlas counts
    # 5 documents while listing exactly two of its own children.
    sources = {node["name"]: node for node in body["sources"]}
    assert set(sources) == {ATLAS, BEACON, COBALT}
    for name, node in sources.items():
        assert node["documents"] == SOURCE_COUNTS[name][0], (name, node)
        assert node["updated_at"].startswith(SOURCE_COUNTS[name][1]), (name, node)

    # NOTHING deeper than level 1 is in the body: every node the payload
    # carries is a source, a level-1 folder, or a level-1 file — and
    # every level-1 folder says so instead of silently rendering short.
    levels = [node["path"].count("/") + 1 for node in _walk(sources[ATLAS].get("children", []))]
    assert levels, "atlas must list its level-1 children"
    assert max(levels) == 1, levels
    for node in _walk(list(sources.values())):
        if node.get("kind") == "folder":
            assert node.get("truncated") is True, node
            assert node.get("children") == [], node
            # D9 under truncation: the recursive max date survives the
            # depth budget — `alpha`'s newest file is TWO levels down.
            assert node["updated_at"], node
    alpha = next(c for c in sources[ATLAS]["children"] if c["path"] == "alpha")
    assert alpha["documents"] == 4, alpha
    assert alpha["updated_at"].startswith("2024-06-06"), alpha
    assert sources[ATLAS]["updated_at"].startswith("2024-07-01")

    # The stat cards ride `totals` (the pre-phase walk of the partial
    # tree would have read 2 documents for the whole catalogue).
    expect(page.locator("#stat-docs")).to_have_text(str(TOTAL_DOCS))
    expect(page.locator("#stat-chunks")).to_have_text(str(TOTAL_CHUNKS))
    expect(page.locator("#stat-last")).not_to_have_text("–")


# --------------------------------------------------------------------------
# 2. gzip on the wire (the browser's own negotiation)
# --------------------------------------------------------------------------


def test_tree_response_arrives_gzipped(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    """Chromium advertises ``Accept-Encoding: gzip, deflate, br`` on its
    own, so this is the real negotiation — no stub, no spoofed header.

    The threshold check is part of the assertion: the body must clear
    the app's ``BOR_COMPRESS_MIN_BYTES`` (the code default, pinned in
    the fixture env) so a payload that starved the compressor could
    never turn this test green.
    """
    page.set_default_timeout(30_000)

    responses: list[Response] = []
    page.on("response", lambda r: responses.append(r))
    login(page, APP_URL)
    _wait_top_level(page)

    body = _body_of(responses, TREE_PATH)
    threshold = _Settings.model_fields["compress_min_bytes"].default
    uncompressed = len(json.dumps(body))
    assert uncompressed > threshold, (uncompressed, threshold)

    headers = next(r for r in responses if urlparse(r.url).path == TREE_PATH).headers
    assert headers.get("content-encoding") == "gzip", headers
    assert "accept-encoding" in (headers.get("vary") or "").lower(), headers
    # The revalidation pair rides the same response (task 03), and the
    # A20 security headers survive outside the compressor (A20).
    assert (headers.get("etag") or "").startswith('W/"'), headers
    assert headers.get("cache-control") == "private, must-revalidate", headers
    assert headers.get("content-security-policy"), headers


# --------------------------------------------------------------------------
# 3. The drill: one node request for an uncached level, rendered in full
# --------------------------------------------------------------------------


def test_drill_into_a_folder_fetches_that_level_once(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    ids = seeded_kb["ids"]

    requests: list[str] = []
    page.on("request", lambda r: requests.append(r.url))
    login(page, APP_URL)
    _wait_top_level(page)

    # The top level renders the stored source-root descriptions and the
    # recursive counts the payload carries.
    rows = _folder_rows(page)
    expect(rows.nth(0).locator("a.folder-link")).to_have_text(ATLAS)
    expect(rows.nth(0).locator("td:nth-child(2)")).to_have_text("5")
    expect(rows.nth(0).locator("td:nth-child(4) span.kb-desc-text")).to_have_text(ATLAS_ROOT_SUM)

    # A SOURCE row costs nothing: the depth-1 payload already holds that
    # level (the phase-97 zero-request drill, kept where it is free).
    page.click(_folder_link(page, ATLAS))
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text(ATLAS)
    assert _paths_of(requests, NODE_PATH) == []
    expect(_folder_rows(page)).to_have_count(1)  # alpha only
    expect(_folder_rows(page).nth(0).locator("td:nth-child(2)")).to_have_text("4")
    # The D9 recursive max on a node the payload truncated: the newest
    # file inside alpha lives two levels below it.
    updated = _folder_rows(page).nth(0).locator("td:nth-child(3)")
    assert (updated.get_attribute("title") or "").startswith("2024-06-06")

    # THE DRILL: opening a folder the payload only NAMES costs EXACTLY
    # one node request for that level.
    page.click(_folder_link(page, "alpha"))
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("alpha")
    node_requests = _paths_of(requests, NODE_PATH)
    assert len(node_requests) == 1, node_requests
    query = parse_qs(urlparse(node_requests[0]).query)
    assert query.get("source") == [ATLAS], node_requests[0]
    assert query.get("path") == ["alpha"], node_requests[0]

    # What the level renders is what the pre-phase full tree rendered.
    # Folders first (beta, still truncated), then the level's own files
    # in catalogue order.
    folders = _folder_rows(page)
    expect(folders).to_have_count(1)
    expect(folders.nth(0).locator("a.folder-link")).to_have_text("beta")
    expect(folders.nth(0).locator("td:nth-child(2)")).to_have_text("2")  # recursive
    expect(folders.nth(0).locator("td:nth-child(4) span.kb-desc-text")).to_have_text(BETA_SUM)

    files = _file_rows(page)
    expect(files).to_have_count(2)
    expect(files.nth(0).locator("td:nth-child(1)")).to_have_text(ATLAS)
    expect(files.nth(0).locator("a.doc-link")).to_have_text(ALPHA_ONE)
    expect(files.nth(0).locator("td:nth-child(4)")).to_have_text("1")
    assert (files.nth(0).locator("td:nth-child(5)").get_attribute("title") or "").startswith(
        "2024-02-11"
    )
    expect(files.nth(1).locator("a.doc-link")).to_have_text(ALPHA_PIC)
    assert (files.nth(1).locator("td:nth-child(5)").get_attribute("title") or "").startswith(
        "2024-02-20"
    )

    # The image affordance survives the node route: the bounded image
    # map rides the drill, so the thumbnail is the SAME route the full
    # tree pointed at, and it carries the vision description as alt.
    thumb = files.nth(1).locator("img.kb-doc-thumb-img")
    expect(thumb).to_have_count(1)
    assert thumb.get_attribute("src") == f"/api/documents/{ids[ALPHA_PIC]}/image"
    assert thumb.get_attribute("alt") == PIC_DESCRIPTION

    # Phase 156 (UI Structure Check): the bounded scroll panel and its
    # sticky header survive a lazy render — the drill paints into the
    # same card, it does not rebuild the page.
    panel = page.locator("#folders-wrap")
    wrap_style = panel.evaluate(
        "el => { const s = getComputedStyle(el);"
        "return { maxHeight: s.maxHeight, overflowY: s.overflowY, "
        "borderRadius: s.borderTopLeftRadius, boxShadow: s.boxShadow }; }"
    )
    assert wrap_style["maxHeight"] not in ("none", ""), wrap_style
    assert wrap_style["overflowY"] in ("auto", "scroll"), wrap_style
    assert wrap_style["boxShadow"] != "none", wrap_style
    head = page.locator("#folders-table thead th").first
    head_style = head.evaluate(
        "el => { const s = getComputedStyle(el); return { p: s.position, t: s.top }; }"
    )
    assert head_style["p"] == "sticky", head_style
    assert head_style["t"] == "0px", head_style

    # …and the lazily rendered level still rides the ONE 72rem centered
    # reading column (AGENTS.md rule 5): a node-fetched level never
    # grows the page or escapes the container.
    geom = page.evaluate(
        """() => {
            const shell = document.querySelector(".container.sources-shell");
            const panel = document.querySelector("#folders-wrap");
            const s = shell.getBoundingClientRect();
            const p = panel.getBoundingClientRect();
            return { shell: s.width, left: p.left - s.left, right: s.right - p.right };
        }"""
    )
    assert geom["shell"] <= 1152 + 4, geom  # 72rem at this 1280px viewport
    assert geom["left"] >= -1 and geom["right"] >= -1, geom


# --------------------------------------------------------------------------
# 4. Two levels deep, then back: the cache answers, the server does not
# --------------------------------------------------------------------------


def test_two_levels_deep_then_back_uses_the_cache(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)

    requests: list[str] = []
    page.on("request", lambda r: requests.append(r.url))
    login(page, APP_URL)
    _wait_top_level(page)
    _drill(page, ATLAS)

    # Level 2 (`alpha`): fetch 1.
    page.click(_folder_link(page, "alpha"))
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("alpha")
    node_requests = _paths_of(requests, NODE_PATH)
    assert len(node_requests) == 1, node_requests
    assert parse_qs(urlparse(node_requests[0]).query).get("path") == ["alpha"], node_requests[0]

    # Level 3 (`alpha/beta`): fetch 2 — and with it the description that
    # NO depth-1 payload can carry (the level block is the unclamped
    # surface, and this row lives below the budget).
    page.click(_folder_link(page, "beta"))
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("beta")
    assert len(_paths_of(requests, NODE_PATH)) == 2, _paths_of(requests, NODE_PATH)
    assert parse_qs(urlparse(_paths_of(requests, NODE_PATH)[1]).query).get("path") == [
        "alpha/beta"
    ], _paths_of(requests, NODE_PATH)
    expect(_folder_rows(page)).to_have_count(1)  # gamma
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text("gamma")
    expect(page.locator("#kb-level")).to_be_visible()
    expect(page.locator("#kb-level-title")).to_have_text(f"{ATLAS}/alpha/beta")
    expect(page.locator("#kb-level-summary")).to_have_text(BETA_SUM)
    expect(_file_rows(page).nth(0).locator("a.doc-link")).to_have_text(BETA_TWO)

    # The stat cards never move: they are catalogue-wide `totals`, not a
    # walk of the level on screen.
    expect(page.locator("#stat-docs")).to_have_text(str(TOTAL_DOCS))
    expect(page.locator("#stat-chunks")).to_have_text(str(TOTAL_CHUNKS))

    # BACK to `alpha` (a crumb): zero new requests, the rows come from
    # what the client holds.
    page.click(_crumb_link(page, "alpha"))
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("alpha")
    assert len(_paths_of(requests, NODE_PATH)) == 2, _paths_of(requests, NODE_PATH)
    expect(_folder_rows(page)).to_have_count(1)
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text("beta")
    expect(_file_rows(page)).to_have_count(2)

    # BACK to the top (a crumb): still nothing on the wire.
    page.click(_crumb_link(page, "Knowledge base"))
    expect(page.locator("#kb-crumb")).to_be_hidden()
    assert len(_paths_of(requests, NODE_PATH)) == 2, _paths_of(requests, NODE_PATH)
    expect(_folder_rows(page)).to_have_count(len(SOURCE_COUNTS))

    # RE-ENTERING levels the client already holds is free: the two
    # fetches above are the whole session's tally, level by level.
    _drill(page, ATLAS, "alpha", "beta")
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("beta")
    assert len(_paths_of(requests, NODE_PATH)) == 2, _paths_of(requests, NODE_PATH)
    # …while the level the cache has NEVER seen costs exactly its own
    # one request (the deepest folder of the deepest chain).
    page.click(_folder_link(page, "gamma"))
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("gamma")
    assert len(_paths_of(requests, NODE_PATH)) == 3, _paths_of(requests, NODE_PATH)
    assert parse_qs(urlparse(_paths_of(requests, NODE_PATH)[2]).query).get("path") == [
        "alpha/beta/gamma"
    ], _paths_of(requests, NODE_PATH)
    expect(_file_rows(page).nth(0).locator("a.doc-link")).to_have_text(GAMMA_THREE)


# --------------------------------------------------------------------------
# 5. The pending marker renders in a lazily loaded level
# --------------------------------------------------------------------------


def test_pending_folder_marker_survives_the_lazy_level(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)

    requests: list[str] = []
    page.on("request", lambda r: requests.append(r.url))
    login(page, APP_URL)
    _wait_top_level(page)

    # `beacon` has no stored description anywhere → the source row and
    # its folder row both carry the phase-98 D4 marker (text + muted
    # color + tooltip, never color alone).
    beacon_row = _folder_rows(page).filter(has_text=BEACON)
    marker = beacon_row.locator("span.kb-desc-text")
    expect(marker).to_have_text("Summary pending")
    assert "kb-summary-pending" in (marker.get_attribute("class") or "")
    assert marker.get_attribute("title") == PENDING_TITLE

    page.click(_folder_link(page, BEACON))
    expect(_folder_rows(page)).to_have_count(1)  # notes — no fetch (level held)
    assert _paths_of(requests, NODE_PATH) == []

    # The lazy level carries it too: the node route answers
    # `summary_pending` exactly like the full tree did.
    notes_marker = _folder_rows(page).nth(0).locator("span.kb-desc-text")
    expect(notes_marker).to_have_text("Summary pending")
    assert "kb-summary-pending" in (notes_marker.get_attribute("class") or "")
    assert notes_marker.get_attribute("title") == PENDING_TITLE
    page.click(_folder_link(page, "notes"))
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("notes")
    assert len(_paths_of(requests, NODE_PATH)) == 1, _paths_of(requests, NODE_PATH)
    # A PENDING level's block shows the D4 note instead of an empty box.
    expect(page.locator("#kb-level")).to_be_visible()
    expect(page.locator("#kb-level-summary")).to_have_text(
        "No description stored yet — the next sync will generate one. "
        "(You can write one yourself.)"
    )
    assert "kb-summary-pending" in (page.locator("#kb-level-summary").get_attribute("class") or "")


# --------------------------------------------------------------------------
# 6. The second visit: If-None-Match on the way out, 304 on the wire
# --------------------------------------------------------------------------


def _conditional_tree_load(page: Page, wire: Wire, act: Any) -> dict[str, Any]:
    """Run ``act`` (a reload, a view switch), then return the tree
    entry the wire settled, asserting the two halves of the claim:
    the browser SENT the validator, the server answered 304."""
    act()
    _wait_top_level(page)
    deadline = time.monotonic() + 10.0
    settled = wire.settled(TREE_PATH)
    while len(settled) < 2 and time.monotonic() < deadline:
        page.wait_for_timeout(100)
        settled = wire.settled(TREE_PATH)
    assert len(settled) >= 2, settled
    return settled[-1]


def test_reload_revalidates_to_304(page: Page, seeded_kb: dict[str, Any], db_ready: None) -> None:
    page.set_default_timeout(30_000)
    wire = Wire(page)

    login(page, APP_URL)
    _wait_top_level(page)

    first = wire.settled(TREE_PATH)
    assert len(first) == 1, first
    assert first[0]["status"] == 200, first[0]
    assert first[0]["if_none_match"] is None, first[0]
    etag = first[0]["etag"]
    assert etag and etag.startswith('W/"'), first[0]
    assert first[0]["content_encoding"] == "gzip", first[0]
    full_transfer = first[0]["encoded_bytes"]
    assert full_transfer > 500, first[0]

    second = _conditional_tree_load(page, wire, lambda: page.reload())

    # The browser re-sent the validator, and the server answered 304 —
    # the second visit costs headers, not the payload.
    assert second["if_none_match"] == etag, second
    assert second["status"] == 304, second
    assert second["encoded_bytes"] < full_transfer / 2, (second, full_transfer)

    # And the table still renders: the cached body is what the owner
    # sees, row for row.
    expect(_folder_rows(page)).to_have_count(len(SOURCE_COUNTS))
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text(ATLAS)
    expect(page.locator("#stat-docs")).to_have_text(str(TOTAL_DOCS))


# --------------------------------------------------------------------------
# 7. The same cache hit through the SPA (History → Sources)
# --------------------------------------------------------------------------


def test_view_switch_revalidates_too(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    wire = Wire(page)

    login(page, APP_URL)
    _wait_top_level(page)
    assert len(wire.settled(TREE_PATH)) == 1

    # The router's mount-once view re-fires its load on a re-show (phase
    # 77) — phase 167 makes the SECOND of those cheap: same URL, same
    # validator, a 304 back.
    def switch() -> None:
        page.click("#nav-history")
        expect(page).to_have_url(APP_URL + "/history.html")
        page.click("#nav-sources")
        expect(page).to_have_url(APP_URL + "/sources.html")

    second = _conditional_tree_load(page, wire, switch)
    assert second["if_none_match"], second
    assert second["status"] == 304, second
    expect(_folder_rows(page)).to_have_count(len(SOURCE_COUNTS))


# --------------------------------------------------------------------------
# 8. Fresh content still lands (the never-stale half of the contract)
# --------------------------------------------------------------------------


def test_an_edit_busts_the_tag_so_fresh_content_lands(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    wire = Wire(page)

    login(page, APP_URL)
    _wait_top_level(page)
    first = wire.settled(TREE_PATH)[0]
    assert first["status"] == 200, first

    # An admin saves a description on a level the payload truncated. The
    # tag is a hash of the BODY (LOCKED A4), so this — not a generation
    # bump — is what invalidates the client's copy.
    client = _admin_client()
    try:
        r = client.patch(
            "/api/folders/summary",
            json={"source": ATLAS, "folder_path": "alpha", "summary": ALPHA_SUMMARY_EDIT},
        )
        assert r.status_code == 200, r.text
    finally:
        client.close()

    def reload() -> None:
        page.reload()

    second = _conditional_tree_load(page, wire, reload)
    assert second["if_none_match"], second
    assert second["status"] == 200, second
    assert second["etag"] != first["etag"], (first, second)

    # The new text renders where the owner looks: on the row, at the
    # level the description belongs to.
    _drill(page, ATLAS)
    row = _folder_rows(page).filter(has_text="alpha")
    expect(row.locator("span.kb-desc-text")).to_have_text(ALPHA_SUMMARY_EDIT)


# --------------------------------------------------------------------------
# 9. A folder that vanished under the open view (§7.4)
# --------------------------------------------------------------------------


def test_vanished_folder_resets_to_the_top_level(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)

    login(page, APP_URL)
    _wait_top_level(page)
    _drill(page, ATLAS)

    # The node route's 404 IS the vanished-folder signal (task 03/04):
    # the folder was pruned between the listing and the click.
    page.route(
        "**/api/docs/tree/node*",
        lambda route: route.fulfill(
            status=404, content_type="application/json", body=json.dumps({"detail": "unknown node"})
        ),
    )
    page.click(_folder_link(page, "alpha"))

    # The honest reset: no stale level, no stale breadcrumb, no row
    # left spinning — the top level renders, because that is all the
    # client can prove.
    expect(page.locator("#kb-crumb")).to_be_hidden()
    expect(_folder_rows(page)).to_have_count(len(SOURCE_COUNTS))
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text(ATLAS)
    expect(page.locator(".kb-level-loading")).to_have_count(0)
    expect(page.locator("#stat-docs")).to_have_text(str(TOTAL_DOCS))
    page.unroute("**/api/docs/tree/node*")


# --------------------------------------------------------------------------
# 10. A failed drill is honest, keeps the level, and recovers
# --------------------------------------------------------------------------


def test_failed_drill_keeps_the_level_and_retry_recovers(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)

    login(page, APP_URL)
    _wait_top_level(page)
    _drill(page, ATLAS)

    # The level the user is ON before the click — it must survive.
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text("alpha")
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text(ATLAS)
    rows_before = _folder_rows(page).count()

    page.route(
        "**/api/docs/tree/node*",
        lambda route: route.fulfill(
            status=500, content_type="text/plain", body="embedding model is down"
        ),
    )
    page.click(_folder_link(page, "alpha"))

    # The alert names the folder and the Retry is a REAL button with an
    # accessible name (phase 49's .retry-btn pill).
    alert = page.locator(".kb-level-loading [role='alert']")
    expect(alert).to_have_count(1)
    expect(alert).to_have_text("Couldn't load alpha — ")
    retry = page.locator(".kb-level-loading button.retry-btn")
    expect(retry).to_have_count(1)
    expect(retry).to_have_text("Retry")
    expect(retry).to_be_visible()
    # The UI Structure Check (AGENTS.md rule 5): the recovery control is
    # a >=44px target and it is keyboard-reachable — tabbing forward
    # from the row's own Edit button lands ON the Retry, with the global
    # :focus-visible ring applied (the same rule every control uses).
    assert retry.evaluate("el => el.getBoundingClientRect().height") >= 44
    page.locator("#folders-tbody .kb-summary-edit").first.evaluate("el => el.focus()")
    page.keyboard.press("Tab")
    focused = page.evaluate(
        "() => { const el = document.activeElement;"
        "return { cls: el.className, text: el.textContent, "
        "visible: el.matches(':focus-visible'), outline: getComputedStyle(el).outlineWidth }; }"
    )
    assert "retry-btn" in focused["cls"], focused
    assert focused["text"] == "Retry", focused
    assert focused["visible"] is True, focused
    assert focused["outline"] not in ("", "0px"), focused

    # Nothing else moved: the level, the breadcrumb and the stat cards
    # are exactly what they were (§7.4 — a failure never paints a hole).
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text(ATLAS)
    expect(_folder_rows(page)).to_have_count(rows_before)
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text("alpha")

    # Retry, with the fault lifted, finishes the drill the click started.
    page.unroute("**/api/docs/tree/node*")
    retry.click()
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("alpha")
    expect(page.locator(".kb-level-loading")).to_have_count(0)
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text("beta")
    expect(_file_rows(page)).to_have_count(2)


# --------------------------------------------------------------------------
# 11. The in-flight drill state, and the second click that costs nothing
# --------------------------------------------------------------------------


def test_in_flight_drill_marks_the_link_and_ignores_a_second_click(
    page: Page, seeded_kb: dict[str, Any], db_ready: None
) -> None:
    page.set_default_timeout(30_000)

    requests: list[str] = []
    page.on("request", lambda r: requests.append(r.url))
    login(page, APP_URL)
    _wait_top_level(page)
    _drill(page, ATLAS)

    # Hold the node request in the harness (the house hold-the-request
    # pattern) so the in-flight state is observed, never raced.
    held: list[Any] = []

    def hold(route: Any) -> None:
        held.append(route)

    page.route("**/api/docs/tree/node*", hold)
    link = page.locator(_folder_link(page, "alpha"))
    link.click()

    # The clicked control says "busy" and refuses a parallel fetch…
    expect(link).to_have_attribute("aria-busy", "true")
    expect(link).to_have_attribute("aria-disabled", "true")
    # …the ONE live line says what is loading…
    status_line = page.locator(".kb-level-loading [role='status']")
    expect(status_line).to_have_count(1)
    expect(status_line).to_have_text("Loading alpha…")
    # …and the level the user is LEAVING stays on screen (never stale).
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text(ATLAS)
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text("alpha")

    # A second click cannot start a second fetch.
    page.click(_folder_link(page, "alpha"), force=True)
    page.click(_folder_link(page, "alpha"), force=True)
    page.wait_for_timeout(300)
    assert len(_paths_of(requests, NODE_PATH)) == 1, _paths_of(requests, NODE_PATH)
    assert len(held) == 1, held

    # Release: the held request answers, the flags clear, the level
    # paints. Every settle path owns its own cleanup (§7.4).
    held[0].fallback()
    expect(page.locator("#kb-crumb .kb-crumb-current")).to_have_text("alpha")
    # The level re-rendered, so the clicked link is gone with it — the
    # settle's cleanup is pinned as "no control anywhere still claims to
    # be busy" (a flag surviving the paint would strand the view).
    expect(page.locator("[aria-busy='true']")).to_have_count(0)
    expect(page.locator("#folders-tbody [aria-disabled='true']")).to_have_count(0)
    expect(page.locator(".kb-level-loading")).to_have_count(0)
    expect(_folder_rows(page).nth(0).locator("a.folder-link")).to_have_text("beta")
    page.unroute("**/api/docs/tree/node*")