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

"""Phase 130 task 04 E2E (Playwright, mock-only): the chat answer renders
as organized clusters — chips row, images row, bottom-right actions row.

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

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

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the story's gate
is the deterministic cluster DOM of the rendered answer, which needs the
``done`` frame to cite ONE text doc AND ONE image doc in the same
answer. That mixed citation is driven by the house scripted turn in
``tests/e2e/mock_llm.py`` (``DRILL_TRIGGER``: the user message contains
``drill down the tree`` — the question carries its own tool call after
the colon, ``drill down the tree: read e2e-clusters/note.md`` — **and**
the system prompt carries the ``<tools>`` section of the HIGH prompt).
The scripted ``read`` cites the text doc (the phase-119 read-docs-only
rule) while the image doc joins the citation tier auto-cited (its
suggested-tier seed block IS its entire content — phase 124): the
``done`` frame's ``sources`` carries exactly the two refs — the text
ref WITHOUT ``image_url`` (the phase-122 omission rule) and the image
ref WITH it — the shape phase 130 renders as two clusters (chips row
text-only, images row figures-only, NO chip for the image doc) + the
actions row last.

KB fixture — ONE local source (a ``tmp_path_factory`` dir, registered
through the authenticated API + the real in-process ``POST /api/sync``
— the ``test_local_directory_sources.py`` / ``test_ls_tree_drilldown.
py`` pattern):

* ``e2e-clusters/note.md`` — a markdown doc whose body carries the
  trigger phrase's words + the diagram vocabulary: the mock's
  token-overlap cosine against the scripted question clears the
  mock-calibrated 0.30 gate (the turn grounds, the ``<tools>`` section
  is present — the flow's precondition);
* ``e2e-clusters/pic.png`` — a standalone 1×1 PNG: the module app runs
  ``BOR_IMAGES=true`` (the phase-122 pattern — the mock's
  ``IMAGE_DESCRIPTION_MODE`` writes the byte-stable
  ``IMAGE_DESCRIPTION_ANSWER`` as the doc's content + summary).

The module app ALSO pins ``BOR_DOCS_REPO`` to a non-empty scratch path
(the ``docs_repo_configured`` gate is ``bool(docs_repo.strip())`` — a
configured repo reveals the Save-as-doc button for the admin; the push
endpoint is never exercised — the button is never clicked) and
``BOR_ANSWER_IMAGES=false`` explicitly (the leak-guard convention: the
owner's local gitignored ``.env`` cannot leak the answer-image toggle
into the premise — the scripted turn makes NO ``show_image`` call, so
the bubble carries no ``.answer-image``).

Test → observable mapping (Playwright Mapping Rule):
1. ``test_answer_clusters_render_in_order`` — the LIVE turn: the wire
   carries the scripted ``read`` once, and the ``done`` frame cites
   BOTH docs — the read text doc first (read order), then the
   auto-cited image ref (``image_url`` present; the text ref's key is
   ABSENT); the last brain bubble's ``.msg-body`` direct children are
   exactly ``[reasoning, bubble, msg-meta, msg-meta msg-images,
   msg-meta msg-actions]`` — the ONE phase-161 reasoning container
   ABOVE the answer (this turn's tool record lives inside its
   ``.reasoning-body`` as a ``.tool-calls-disclosure`` block, the
   phase-130 timeline re-targeted by phase 161 D1) then the chips row (EXACTLY the
   text chip — NO chip for the image doc, phase 130 task 01) before
   the images row (the figure, ``<img>`` loaded from the bytes route,
   alt + caption settling to the canned description), no related row
   (the 2-doc KB has no related tier), and the actions row LAST
   (the "bottom" of the TODO's "bottom right") holding ALL THREE gated
   buttons (admin + configured docs repo — the phase-16/59 gates).
   Geometry (bounding boxes, not pixels): the actions row's box is
   BELOW the images row's, and the rightmost button's right edge sits
   at the row's right edge minus the row's 0.25rem padding (the
   per-button auto margins are gone — the row's
   ``justify-content: flex-end`` does the right-align, phase 130
   task 03).
2. ``test_image_without_chip_keeps_its_citation_via_figure`` — the
   same bubble: the ONLY ``.source-chip`` is the text doc's (the
   owner's exact ask — no chip whose label is the image's
   ``source/path``), while the image doc is fully cited: the figure's
   ``href`` is its ``documentUrl`` and a left-click opens the same-
   page modal (the phase-26 affordance — the modal shows the image +
   the description). The text chip keeps its own phase-26 affordance
   (left-click → the same modal) — the no-behavior-change half.
3. ``test_restore_renders_the_same_clusters`` — the turn auto-saves
   (phase 66); a page reload restores it through the phase-14 restore
   path (the same appenders): the restored last brain bubble carries
   the IDENTICAL ``.msg-body`` child sequence (chips row text-only,
   images row, actions row LAST with the three gated buttons) and the
   same geometry — the DOM order assertion pinned for BOTH the live
   and the restored bubble.
4. ``test_non_admin_sees_retry_only_actions`` — a FRESH non-admin
   browser context (no admin cookie — the real in-app token gate,
   ``login_with_token``, the phase-79 pattern) on a FRESH chat: the
   same scripted turn, and the ``.msg-actions`` row holds ONLY the
   ``.retry-btn`` (Tune admin-gated phase 16, Save-as-doc admin-
   gated phase 59 — the gates unchanged) while the cluster rows
   render identically (the render is not identity-gated).

DB isolation: the suites share one Postgres. ``saved_chats`` survives
the truncation (the house pattern) — every question text here is
UNIQUE (its auto-title therefore is) and the module boundary cleans
this suite's rows by title. The non-admin turn's auto-save 403s
(``/api/chats`` is admin-only — the quiet A2 status note, no row) and
its token is revoked at the end of the test.
"""
from __future__ import annotations

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

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

from app.config import Settings
from app.db import SessionLocal
from app.models import Document
from e2e.auth_helpers import login, login_with_token
from e2e.conftest import ADMIN_PASSWORD, SESSION_SECRET, USE_REAL_LLM, _wait_http
from e2e.mock_llm import IMAGE_DESCRIPTION_ANSWER

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_CLUSTERS", "8170"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"

SETTLE_TIMEOUT_MS = 30_000
FIGURE_TIMEOUT_MS = 15_000

# --------------------------------------------------------------------------
# The fixture (one local source: ONE text doc + ONE image doc)
# --------------------------------------------------------------------------

SOURCE_NAME = "e2e-clusters"  # the directory basename = the source name
TEXT_PATH = "note.md"
IMG_PATH = "pic.png"
TEXT_SP = f"{SOURCE_NAME}/{TEXT_PATH}"  # the chip's label (source/path)
IMG_SP = f"{SOURCE_NAME}/{IMG_PATH}"
TEXT_TITLE = "Homelab Network Note"  # the first H1 (extract_title)
IMG_TITLE = "pic"  # the image doc's title = the file stem (phase 122)
#: The chip/figure ``href`` (the phase-26 same-page-modal affordance —
#: ``documentUrl(source, path, "/")`` — the /document.html escape hatch).
TEXT_DOC_URL = f"/document.html?source={SOURCE_NAME}&path={TEXT_PATH}&back=%2F"
IMG_DOC_URL = f"/document.html?source={SOURCE_NAME}&path={IMG_PATH}&back=%2F"

#: The text doc's body carries the trigger phrase's words + the diagram
#: vocabulary, so the scripted questions (which repeat them) clear the
#: mock's token-overlap cosine gate deterministically (~0.5 vs the
#: mock-calibrated 0.30).
TEXT_CONTENT = (
    "# Homelab Network Note\n"
    "\n"
    "The drill down the tree fixture note for the e2e clusters suite: this\n"
    "homelab note describes the network diagram — the core router, the core\n"
    "switch, and the three VLAN subnets the homelab network diagram shows.\n"
)

# The scripted turn's answer (the phase-37 single-read shape): the mock
# quotes the read document's FIRST 80 chars — the H1 line (renders as a
# heading) + 56 body chars (the assertable prefix below, one text node).
READ_ANSWER_PREFIX = f"Read {TEXT_SP}."
READ_BODY_QUOTE = "The drill down the tree fixture note for the e2e cluster"

# A real 1×1 transparent PNG (the unit/integration suites' fixture —
# the pipeline is content-agnostic, the well-formed bytes keep the
# upload + serve + render path honest).
PNG_1X1 = base64.b64decode(
    "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJ"
    "AAAAC0lEQVR4nGP4DwQACfsD/fteaysAAAAASUVORK5CYII="
)

# --------------------------------------------------------------------------
# The scripted questions (the mock's ``DRILL_TRIGGER`` — the regex
# ``drill down the tree:\s*read\s+[a-z0-9_./-]+`` parses the read
# target from the RAW message, so the trigger line stays lowercase and
# the path stays in its character set). Each tail is unique — the
# auto-titles are unique on the shared e2e Postgres (the row-lookup
# convention of test_answer_images).
# --------------------------------------------------------------------------

Q_ORDER = (
    "drill down the tree: read e2e-clusters/note.md — "
    "what does the homelab note say about the diagram?"
)
Q_FIGURE = (
    "drill down the tree: read e2e-clusters/note.md — "
    "which diagram does the note describe?"
)
Q_RESTORE = (
    "drill down the tree: read e2e-clusters/note.md — "
    "summarize the homelab network note (restore)"
)
Q_NONADMIN = (
    "drill down the tree: read e2e-clusters/note.md — "
    "what is the note's diagram? (non-admin)"
)
assert all(len(q) <= 120 for q in (Q_ORDER, Q_FIGURE, Q_RESTORE, Q_NONADMIN)), (
    "the auto-title cap is 120 chars — the questions must stay whole"
)

#: The .msg-body child sequence of a completed scripted-read bubble
#: (phase 130's fix, DOM-asserted; re-targeted by phase 161 D1): the
#: ONE reasoning container (the scripted read's tool record rides
#: INSIDE it — .reasoning .tool-calls-disclosure, asserted beside the
#: sequence), the answer bubble, the chips row (the bare .msg-meta),
#: the images row, the actions row LAST. The live and restore paths
#: share it.
EXPECTED_CHILD_CLASSES = [
    "reasoning",
    "bubble",
    "msg-meta",
    "msg-meta msg-images",
    "msg-meta msg-actions",
]


# ---------------------------------------------------------------------------
# Module app: images ON + a configured docs repo (the Save-as-doc gate)
# ---------------------------------------------------------------------------


@pytest.fixture(scope="module")
def cluster_dir(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """The one-source fixture tree (see the module docstring): the app
    server runs on the same host, so the paths are visible to it. The
    directory NAME is the source name (``kind=local`` → the directory's
    basename, phase 38)."""
    scratch = tmp_path_factory.mktemp("bor_clusters")
    src = scratch / SOURCE_NAME
    src.mkdir()
    (src / TEXT_PATH).write_text(TEXT_CONTENT, encoding="utf-8")
    (src / IMG_PATH).write_bytes(PNG_1X1)
    assert (src / TEXT_PATH).is_file() and (src / IMG_PATH).is_file()
    return src


@pytest.fixture(scope="module")
def app_server(mock_llm: int, cluster_dir: Path) -> Iterator[str]:
    """The real app under test — per-module app (the conftest pattern):
    NO ``BOR_GIT_SOURCES`` (the env fallback is git-only — the source
    here is a DB-registered local directory), ``BOR_IMAGES=true`` (the
    PNG must become an image document), a configured ``BOR_DOCS_REPO``
    (the Save-as-doc reveal gate), and the leak-guarded code defaults.
    The session app is never started in this isolated run, so no port
    clash."""
    scratch = cluster_dir.parent  # the tmp dir holding the source tree
    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): the scripted questions
    # ground on these values (see TEXT_CONTENT).
    env["BOR_RELEVANCE_THRESHOLD"] = "0.30"
    env["BOR_LEXICAL_SUPPORT_FLOOR"] = "0.15"
    env["BOR_SOURCE_USEFULNESS_FLOOR"] = "0.15"
    # Phase 67: instant retry waits + the code-default budget.
    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 BOR_GIT_SOURCES (the app
    # reads it from cwd) — override it with an EMPTY value: the registry
    # must hold EXACTLY this suite's local directory.
    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"] = str(scratch / "docs")  # non-empty → configured
    env["BOR_DOCS_WORK_DIR"] = str(scratch / "docs-work")
    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
    # Phase 122: the PNG must become an image document; the image/upload
    # homes are in the suite's scratch (never the owner's real
    # ~/bor-sources). The answer-image toggle is pinned OFF explicitly —
    # the scripted turn makes no show_image call, so no .answer-image.
    env["BOR_IMAGES"] = "true"
    env["BOR_ANSWER_IMAGES"] = "false"
    env["BOR_UPLOAD_DIR"] = str(scratch / "uploads")
    env["BOR_SOURCES_DIR"] = str(scratch / "checkouts")
    env["BOR_IMAGE_DIR"] = str(scratch / "images")
    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


# ---------------------------------------------------------------------------
# DB + admin-HTTP helpers
# ---------------------------------------------------------------------------


def _truncate_all() -> None:
    """Fresh registry + KB (the E2E isolation pattern): the suites share
    one Postgres, and a leftover row would corrupt the 2-doc premise.
    ``saved_chats`` is never touched — the unique question texts keep
    the row lookups unambiguous, and :func:`_cleanup_saved_rows` owns
    this suite's rows (re-runnable on the shared DB)."""
    with SessionLocal() as db:
        db.execute(text(
            "TRUNCATE chunks, documents, query_log, steering_notes, "
            "kb_overview, git_sources, sources_meta, folder_summaries"
        ))
        db.commit()


def _health_db(url: str) -> bool:
    try:
        return httpx.get(f"{url}/api/health", timeout=5).json()["db"] == "up"
    except Exception:  # noqa: BLE001 — unreachable is the skip case
        return False


@contextmanager
def _admin_client(app_url: str) -> Iterator[httpx.Client]:
    """An httpx client holding a fresh admin session (the form login's
    API twin — the house httpx pattern of test_ls_tree_drilldown's
    ``synced_kb``), closed on exit."""
    client = httpx.Client(base_url=app_url, timeout=30.0)
    try:
        r = client.post("/api/login", json={"password": ADMIN_PASSWORD})
        assert r.status_code == 204, r.text
        yield client
    finally:
        client.close()


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_local_directory_sources pattern)."""
    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}")


def _auto_title(question: str) -> str:
    """The phase-50 auto-title convention: the first question,
    whitespace-collapsed, capped at 120 chars."""
    return " ".join(question.split())[:120]


#: This suite's auto-titles (the shared e2e Postgres keeps
#: ``saved_chats`` across runs — the test_answer_images cleanup
#: concern): a re-run must start clean.
_SUITE_TITLES = tuple(
    _auto_title(q) for q in (Q_ORDER, Q_FIGURE, Q_RESTORE, Q_NONADMIN)
)


def _cleanup_saved_rows() -> None:
    """Delete THIS suite's saved rows by title (best-effort: a failed
    delete (DB unreachable — the skip case) never masks the skip)."""
    try:
        with SessionLocal() as db:
            db.execute(
                text("DELETE FROM saved_chats WHERE title = ANY(:titles)"),
                {"titles": list(_SUITE_TITLES)},
            )
            db.commit()
    except Exception:  # noqa: BLE001 — DB unreachable: the skip handles it
        pass


def _wait_saved_row(
    client: httpx.Client, title: str, messages: int = 2
) -> dict[str, Any]:
    """Wait for the auto-saved row (phase 55/66: auto-saves are SILENT
    — so there is no status line to wait on)."""
    deadline = time.monotonic() + 15
    last: dict[str, Any] | None = None
    while time.monotonic() < deadline:
        r = client.get("/api/chats", timeout=10)
        assert r.status_code == 200
        rows = r.json()["chats"]
        last = next((c for c in rows if c["title"] == title), None)
        if last is not None and last["message_count"] >= messages:
            return last
        time.sleep(0.2)
    raise AssertionError(f"no auto-saved row for {title!r} (last: {last!r})")


def _create_token(client: httpx.Client, label: str) -> tuple[str, str]:
    """Admin-issued token (``POST /api/tokens`` — the house pattern of
    test_api_tokens): returns (row id, plaintext)."""
    r = client.post("/api/tokens", json={"label": label}, timeout=10)
    assert r.status_code == 201, r.text
    body = r.json()
    assert body["label"] == label
    return body["id"], body["token"]


def _revoke_token(client: httpx.Client, token_id: str) -> None:
    """Revoke the token (a revoked token is dead NOW — A10): the shared
    DB keeps no live credential behind the suite."""
    r = client.post(f"/api/tokens/{token_id}/revoke", timeout=10)
    assert r.status_code == 204, r.text


@pytest.fixture(scope="module")
def seeded(app_url: str, cluster_dir: Path) -> Iterator[dict[str, str]]:
    """The story's precondition: the one-source KB (ONE text doc + ONE
    image doc) synced under the deterministic mock — registered through
    the authenticated API, scanned by the REAL in-process
    ``POST /api/sync`` pipeline (walk → chunk → embed → the mock's
    vision description for the PNG → overview → version bump). Yields
    the seeded image doc's identity (the bytes route + the
    description) for the render assertions."""
    if not _health_db(app_url):
        pytest.skip("Postgres not reachable — run `podman compose up -d db` first")
    _cleanup_saved_rows()  # a previous run's rows would be stale (new doc id)
    _truncate_all()
    with _admin_client(app_url) as client:
        r = client.post(
            "/api/git-sources", json={"kind": "local", "path": str(cluster_dir)}
        )
        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"] == 2, detail  # the text doc + the image doc
    with SessionLocal() as db:
        text_doc = db.scalar(
            select(Document).where(
                Document.source == SOURCE_NAME, Document.path == TEXT_PATH
            )
        )
        img_doc = db.scalar(
            select(Document).where(
                Document.source == SOURCE_NAME, Document.path == IMG_PATH
            )
        )
    assert text_doc is not None, "the text doc must be indexed"
    assert img_doc is not None, "the PNG must become a document"
    assert img_doc.is_image is True and img_doc.image_path is not None
    # content == summary == the mock's description (the ONLY embedded
    # text of the image doc — the phase-122 pipeline contract).
    assert img_doc.content == IMAGE_DESCRIPTION_ANSWER
    assert img_doc.summary == IMAGE_DESCRIPTION_ANSWER
    assert text_doc.title == TEXT_TITLE  # the first H1 (extract_title)
    assert img_doc.title == IMG_TITLE  # the file stem (phase 122)
    yield {
        "image_url": f"/api/documents/{img_doc.id}/image",
        "description": IMAGE_DESCRIPTION_ANSWER,
    }
    _cleanup_saved_rows()
    _truncate_all()


@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 and the docs
    persist 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_ls_tree_drilldown house pattern)
# ---------------------------------------------------------------------------

#: Captures the raw SSE ``data:`` payloads of the /api/chat stream
#: (a response clone read in the background) — wire-level assertions
#: for the ``done`` frame's mixed citation, independent of the UI.
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) -> list[dict[str, Any]]:
    """The SSE frames captured since the last submit (``_submit``
    clears the buffer), once the hook's background read settles."""
    deadline = time.monotonic() + 30.0
    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[str, Any]]) -> list[dict[str, Any]]:
    return [f for f in frames if f.get("type") == "tool"]


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) -> None:
    """The turn is complete: answer text in the bubble, button recovered
    (the phase-48 settle wait, the test_agent_document_tools helper)."""
    expect(
        page.locator(".msg.brain .bubble").last
    ).not_to_have_text("", timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator("#send-btn")).to_be_enabled(timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=SETTLE_TIMEOUT_MS)


def _wait_answer(page: Page) -> None:
    """The scripted read's answer landed (the settle's answer-side half
    — used when the turn's wire is not asserted separately)."""
    expect(
        page.locator(".msg.brain .bubble").last
    ).to_contain_text(READ_ANSWER_PREFIX, timeout=SETTLE_TIMEOUT_MS)
    _wait_settled(page)


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


def _body_child_classes(page: Page) -> list[str]:
    """The LAST brain bubble's ``.msg-body`` direct children's class
    names, in DOM order — the cluster-order pin (the phase-130 fix,
    asserted exactly: pre-phase markup — buttons inside the chips row,
    figures interleaved — fails this sequence)."""
    return page.evaluate(
        """() => {
            const body = [...document.querySelectorAll(".msg.brain")].pop()
                .querySelector(".msg-body");
            return [...body.children].map((el) => el.className);
        }"""
    )


def _assert_cluster_rows(
    page: Page,
    seeded: dict[str, str],
    *,
    action_buttons: list[str],
) -> None:
    """The phase-130 cluster contract on the LAST brain bubble (the
    live and restore paths share it): the chips row holds EXACTLY the
    text doc's chip (NO chip for the image doc — the figure IS the
    citation, phase 130 task 01), the images row holds the one figure
    (the ``<img>`` loaded from the bytes route, alt + the visible
    caption settling to the document's summary), and the actions row
    — the bubble's LAST row — holds exactly the gated buttons
    (*action_buttons*; the row-level right-align is
    ``_assert_actions_geometry``'s job)."""
    bubble = _last_brain(page)
    # The three rows exist exactly once each (the bare .msg-meta is the
    # chips row — the images/actions/related rows carry their own
    # classes).
    chips_row = bubble.locator(
        ".msg-meta:not(.msg-images):not(.msg-actions):not(.related-docs)"
    )
    images_row = bubble.locator(".msg-meta.msg-images")
    actions_row = bubble.locator(".msg-meta.msg-actions")
    expect(chips_row).to_have_count(1)
    expect(images_row).to_have_count(1)
    expect(actions_row).to_have_count(1)
    # ARIA is kept (the house meta-row convention, unchanged).
    expect(chips_row).to_have_attribute("role", "list")
    expect(chips_row).to_have_attribute("aria-label", "Sources")
    expect(images_row).to_have_attribute("role", "list")
    expect(images_row).to_have_attribute("aria-label", "Source images")
    expect(actions_row).to_have_attribute("role", "list")
    expect(actions_row).to_have_attribute("aria-label", "Answer actions")

    # The chips row: EXACTLY the text doc's chip — and the bubble
    # carries NO other chip anywhere (the image row has none — the
    # owner's exact ask, TODO L6).
    chips = chips_row.locator(".source-chip")
    expect(chips).to_have_count(1)
    expect(chips.first).to_have_text(TEXT_SP)
    expect(chips.first).to_have_attribute("href", TEXT_DOC_URL)
    expect(bubble.locator(".source-chip")).to_have_count(1)

    # The images row: the one figure — the citation of the image doc.
    # Its href is the documentUrl (the phase-26 escape hatch), the
    # <img> loaded from the bytes route, alt + caption settling to the
    # document's summary (the fetchContentSummary pattern).
    fig = images_row.locator(".source-image")
    expect(fig).to_have_count(1)
    expect(fig).to_have_attribute("href", IMG_DOC_URL)
    img = fig.locator(".source-image-img")
    expect(img).to_be_visible(timeout=FIGURE_TIMEOUT_MS)
    expect(img).to_have_attribute("src", seeded["image_url"])
    expect(img).to_have_attribute("alt", seeded["description"], timeout=FIGURE_TIMEOUT_MS)
    expect(fig.locator(".source-image-caption")).to_have_text(
        seeded["description"], timeout=FIGURE_TIMEOUT_MS
    )

    # The actions row: exactly the gated buttons (the cluster — nothing
    # else joins it: the notes are absent on a completed turn), and
    # NOTHING after it — it is the bubble's LAST row (the "bottom" of
    # "bottom right").
    for cls in action_buttons:
        expect(actions_row.locator(f".{cls}")).to_have_count(1)
    expect(actions_row.locator("button")).to_have_count(len(action_buttons))
    expect(actions_row.locator("~ *")).to_have_count(0)


def _assert_actions_geometry(page: Page) -> None:
    """The TODO's "bottom right" geometry (bounding boxes, not pixels):
    the actions row is BELOW the images row, and the rightmost
    button's right edge sits at the row's right edge minus the row's
    0.25rem inline padding (≤ 5px — the per-button auto margins are
    gone: the row's ``justify-content: flex-end`` does the
    right-align, phase 130 task 03)."""
    bubble = _last_brain(page)
    a_box = bubble.locator(".msg-meta.msg-actions").bounding_box()
    i_box = bubble.locator(".msg-meta.msg-images").bounding_box()
    assert a_box and i_box, "the cluster rows must be laid out (visible)"
    assert a_box["y"] >= i_box["y"] + i_box["height"] - 1, (
        f"the actions row must be BELOW the images row: actions.y="
        f"{a_box['y']} vs images bottom {i_box['y'] + i_box['height']}"
    )
    btn_box = (
        bubble.locator(".msg-meta.msg-actions button").last.bounding_box()
    )
    assert btn_box, "the rightmost button must be laid out (visible)"
    gap = (a_box["x"] + a_box["width"]) - (btn_box["x"] + btn_box["width"])
    assert 0 <= gap <= 5, (
        f"the rightmost button must sit at the row's right edge (the "
        f"only slack is the row's 0.25rem padding): gap={gap}px"
    )


# ---------------------------------------------------------------------------
# 1. The live turn: chips row → images row → actions row LAST (the
#    owner-visible fix, DOM-asserted), the wire cites both docs
# ---------------------------------------------------------------------------


def test_answer_clusters_render_in_order(
    page: Page, app_url: str, seeded: dict[str, str], db_ready: None
) -> None:
    """The scripted mixed-citation turn (the mock's ``DRILL_TRIGGER``
    scripted read of the text doc; the image doc auto-cited — see the
    module docstring): the ``done`` frame cites BOTH docs, and the
    last brain bubble's ``.msg-body`` reads as organized clusters top
    to bottom — the answer bubble, the chips row (EXACTLY the text
    chip — NO chip for the image doc), the images row (the figure,
    loaded, caption settled to the canned description), and the
    actions row LAST with ALL THREE gated buttons (admin + configured
    docs repo) — right-aligned and below the images row (the "bottom
    right" geometry)."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)
    login(page, app_url, next="/")
    expect(page.locator("#kb-banner")).to_be_hidden()
    _install_page_hooks(page)

    _submit(page, Q_ORDER)
    _wait_settled(page)

    # The WIRE: exactly the scripted read (one tool frame), and the
    # done frame carries the mixed citation — the read text doc FIRST
    # (the phase-119 read order), then the auto-cited image ref:
    # the text ref's ``image_url`` key is ABSENT (the phase-122
    # omission rule), the image ref carries it. No related tier (the
    # 2-doc KB), no answer figure (no show_image call).
    frames = _frames(page)
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "read", "argument": TEXT_SP}
    ], _tool_frames(frames)
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is False, done
    assert done["sources"] == [
        {"source": SOURCE_NAME, "path": TEXT_PATH, "title": TEXT_TITLE},
        {
            "source": SOURCE_NAME,
            "path": IMG_PATH,
            "title": IMG_TITLE,
            "image_url": seeded["image_url"],
        },
    ], done["sources"]
    assert done["related"] == []
    assert done["suggestions"] == []
    assert "answer_image" not in done

    # The scripted read reached the model and landed in the answer
    # (the phase-37 citation shape: the prefix + the document's first
    # 80 chars — the H1 renders as a heading, the body quote stays in
    # its paragraph).
    bubble = _last_brain(page).locator(".bubble")
    expect(bubble).to_contain_text(READ_ANSWER_PREFIX)
    expect(bubble).to_contain_text(TEXT_TITLE)
    expect(bubble).to_contain_text(READ_BODY_QUOTE)

    # The DOM ORDER (the fix): reasoning container (tool record
    # inside), answer, chips row, images row, actions row LAST —
    # asserted exactly (this sequence fails against the pre-phase
    # soup div, where the buttons shared the chips row and the figure
    # interleaved with it).
    assert _body_child_classes(page) == EXPECTED_CHILD_CLASSES, (
        _body_child_classes(page)
    )
    # The tool record lives INSIDE the ONE reasoning container above
    # the answer (phase 161 D1/D5 re-target of the phase-130
    # timeline) — the read line carries the read document's path.
    tool_record = _last_brain(page).locator(".reasoning .tool-calls-disclosure")
    expect(tool_record).to_have_count(1)
    expect(tool_record.locator(".tool-call").last).to_contain_text(TEXT_SP)

    # The clusters + the actions row's contents (all three buttons —
    # admin, docs repo configured) + nothing after the actions row.
    _assert_cluster_rows(
        page, seeded,
        action_buttons=["tune-btn", "retry-btn", "save-as-doc-btn"],
    )
    _assert_actions_geometry(page)


# ---------------------------------------------------------------------------
# 2. No chip for the image doc — the figure is its full citation
#    (documentUrl href + the phase-26 same-page modal)
# ---------------------------------------------------------------------------


def test_image_without_chip_keeps_its_citation_via_figure(
    page: Page, app_url: str, seeded: dict[str, str], db_ready: None
) -> None:
    """The owner's exact ask in one bubble: NO ``.source-chip`` whose
    label is the image's ``source/path`` (the ONLY chip is the text
    doc's), while the image doc is fully cited — the figure's ``href``
    is its ``documentUrl`` and a LEFT-CLICK opens the same-page modal
    (the phase-26 affordance, the modal shows the image + the
    description). The text chip keeps its own phase-26 affordance
    (left-click → the same modal) — the no-behavior-change half."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)
    login(page, app_url, next="/")
    expect(page.locator("#kb-banner")).to_be_hidden()

    _submit(page, Q_FIGURE)
    _wait_answer(page)

    bubble = _last_brain(page)
    # The ONLY chip is the text doc's — there is NO chip for the image
    # doc (the figure is the citation — TODO L6).
    chips = bubble.locator(".source-chip")
    expect(chips).to_have_count(1)
    expect(chips.first).to_have_text(TEXT_SP)

    # The figure is the image doc's full citation: the documentUrl
    # href, and a left-click opens the same-page modal (phase 26).
    fig = bubble.locator(".msg-meta.msg-images .source-image")
    expect(fig).to_have_count(1)
    expect(fig).to_have_attribute("href", IMG_DOC_URL)
    fig.click()
    expect(page.locator("#doc-modal")).to_be_visible(timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator("#doc-modal-title")).to_have_text(IMG_TITLE)
    img = page.locator("#doc-modal .doc-image-img")
    expect(img).to_have_count(1)
    expect(img).to_have_attribute("src", seeded["image_url"])
    # Phase 165: the description renders in the wrapping .doc-prose
    # block (an image description is prose — no scrolling pre).
    expect(page.locator("#doc-modal .doc-prose")).to_have_text(
        seeded["description"]
    )
    page.keyboard.press("Escape")  # the phase-26 close affordance
    expect(page.locator("#doc-modal")).to_be_hidden()

    # The text chip keeps its own phase-26 affordance (the no behavior
    # change half of the phase): left-click → the same modal, the
    # text doc's title + content.
    chips.first.click()
    expect(page.locator("#doc-modal")).to_be_visible(timeout=SETTLE_TIMEOUT_MS)
    expect(page.locator("#doc-modal-title")).to_have_text(TEXT_TITLE)
    page.keyboard.press("Escape")
    expect(page.locator("#doc-modal")).to_be_hidden()


# ---------------------------------------------------------------------------
# 3. Reload: the phase-14 restore path renders the IDENTICAL clusters
# ---------------------------------------------------------------------------


def test_restore_renders_the_same_clusters(
    page: Page, app_url: str, seeded: dict[str, str], db_ready: None
) -> None:
    """The turn auto-saves (phase 66); a page RELOAD restores the
    conversation through the phase-14 restore path (the same
    appenders — ``renderStoredMessage`` + the after-the-loop
    ``markLastRetryable``): the restored last brain bubble carries the
    IDENTICAL ``.msg-body`` child sequence (chips row text-only,
    images row, ``.msg-actions`` LAST with the three gated buttons)
    and the same "bottom right" geometry — the DOM order assertion
    pinned for the restored bubble too."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)
    login(page, app_url, next="/")
    expect(page.locator("#kb-banner")).to_be_hidden()

    _submit(page, Q_RESTORE)
    _wait_answer(page)
    _assert_cluster_rows(
        page, seeded,
        action_buttons=["tune-btn", "retry-btn", "save-as-doc-btn"],
    )
    _assert_actions_geometry(page)
    live_children = _body_child_classes(page)

    # The auto-save settled (the row exists with both messages) before
    # the reload (phase 55/66 — the silent upsert).
    with _admin_client(app_url) as client:
        _wait_saved_row(client, _auto_title(Q_RESTORE))

    # RELOAD: the localStorage restore (the done handler's save point
    # ran synchronously, so the record is durable before we reload).
    page.reload()
    expect(page.locator("#kb-banner")).to_be_hidden()
    expect(page.locator(".msg.user .bubble")).to_have_count(1)
    expect(page.locator(".msg.brain .bubble")).to_have_count(1)
    restored = page.locator(".msg.brain .bubble").last
    expect(restored).to_contain_text(READ_ANSWER_PREFIX, timeout=SETTLE_TIMEOUT_MS)

    # The IDENTICAL structure (the restore renders through the same
    # appenders — the child sequence is the live one, exactly) — the
    # segments path re-renders the reasoning container with the tool
    # record inside (phase 161 D4).
    assert _body_child_classes(page) == EXPECTED_CHILD_CLASSES == live_children, (
        _body_child_classes(page)
    )
    tool_record = _last_brain(page).locator(".reasoning .tool-calls-disclosure")
    expect(tool_record).to_have_count(1)
    expect(tool_record.locator(".tool-call").last).to_contain_text(TEXT_SP)
    _assert_cluster_rows(
        page, seeded,
        action_buttons=["tune-btn", "retry-btn", "save-as-doc-btn"],
    )
    _assert_actions_geometry(page)


# ---------------------------------------------------------------------------
# 4. A non-admin (token user) sees ONLY the Retry in the actions row
# ---------------------------------------------------------------------------


def test_non_admin_sees_retry_only_actions(
    browser: Browser,
    app_url: str,
    seeded: dict[str, str],
    db_ready: None,
) -> None:
    """A FRESH non-admin context (a new browser context — no admin
    cookie, no cached token, no localStorage → a FRESH chat) signs in
    through the REAL in-app token gate (``login_with_token``, phase
    79) with an admin-issued token, and sends the same scripted turn:
    the cluster rows render identically (the render is not
    identity-gated — chips + images + the ARIA names), and the
    ``.msg-actions`` row holds ONLY the ``.retry-btn`` (Tune admin-
    gated phase 16, Save-as-doc admin-gated phase 59 — the gates
    unchanged). The token is revoked at the end."""
    with _admin_client(app_url) as client:
        token_id, token = _create_token(client, "e2e-clusters-guest")
        try:
            ctx = browser.new_context(viewport={"width": 1280, "height": 800})
            try:
                anon = ctx.new_page()
                anon.set_default_timeout(SETTLE_TIMEOUT_MS)
                login_with_token(anon, app_url, token, next="/")
                expect(anon.locator("#kb-banner")).to_be_hidden()

                _submit(anon, Q_NONADMIN)
                _wait_answer(anon)

                bubble = _last_brain(anon)
                # The cluster rows are identical for the token user
                # (the actions row carries the retry-only cluster).
                _assert_cluster_rows(anon, seeded, action_buttons=["retry-btn"])
                # The gates, unchanged: no Tune (phase 16), no
                # Save-as-doc (phase 59) — the ONE button in the row
                # is the Retry.
                actions = bubble.locator(".msg-meta.msg-actions")
                expect(actions.locator(".tune-btn")).to_have_count(0)
                expect(actions.locator(".save-as-doc-btn")).to_have_count(0)
                expect(actions.locator("button")).to_have_count(1)
                expect(actions.locator(".retry-btn")).to_have_count(1)
            finally:
                ctx.close()
        finally:
            _revoke_token(client, token_id)