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

"""Phase 52 E2E (Playwright): the composer never runs away from the user.

Source: ``TODO.md`` L3 — "The message input text box needs to be pinned to
the bottom of the screen so it doesn't \"run away\" from the user as they
try to click \"stop\"" (no user story file — TODO-derived phase).
Run in isolation (DB must be up: ``podman compose up -d db``):

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

Mock-only, no admin login (chat is public). The owner-locked contract
(2026-08-30) under test — the CSS pin (``position: sticky;
bottom: env(safe-area-inset-bottom)`` on ``.composer``, phase 52 task 01)
observed in a real browser:

* with an over-viewport conversation the composer box is fully inside the
  viewport, flush with its bottom edge, at EVERY scroll position — the
  reader never has to scroll to find the input;
* the original bug scenario: mid-turn, scrolled up reading earlier
  content, the Stop control (``#send-btn.is-stop``, phase 48) is visible
  and clickable WITHOUT scrolling, the click settles the turn (partial
  kept + ``.stopped-note``, no error banner), it moves the viewport not
  at all (the phase-42 never-auto-scroll contract — the pin adds no scroll
  call site), and a reload restores the ``stopped`` record through the
  normal ``bor.chat.v1`` path;
* on an empty/short chat the composer renders in its normal flow position
  — the pin does not float it over the footer (``.app-footer`` stays in
  flow below it), and at the document bottom the box has settled back into
  flow above the footer instead of hovering on the viewport edge;
* ≤640px the pin holds too (the composer is reachable, ≥44px Send target)
  and stays UNDER the sticky header — no z-index was added, so the pinned
  box can never cover the header or the hamburger dropdown (phase 46).

Determinism: the mock's grounded answers quote the question and end in the
``Deterministic mock answer for E2E`` marker; the long answer (``write a
long answer`` trigger, phase 11) streams ~900 words over ~8 s at 12 chars /
0.02 s — the in-flight window the Stop click rides — with a unique final
line (``LONG-ANSWER-END``) absent from any partial. Overflow is built from
real UI turns (no fixture, no DOM injection): 6 short turns put ~1.5k px
of conversation into the 800px viewport.

Test → story mapping (Playwright Mapping Rule):
1. ``test_composer_pinned_at_every_scroll_position``
2. ``test_stop_is_reachable_from_scrolled_up`` — the TODO.md L3 scenario
3. ``test_empty_chat_composer_sits_in_normal_flow``
4. ``test_pin_holds_on_mobile_under_the_header`` — ≤640px stacking
"""
from __future__ import annotations

import asyncio
import json
import re
import time
from collections.abc import Iterator
from pathlib import Path
from threading import Thread
from typing import Any

import pytest
from playwright.sync_api import Browser, Page, ViewportSize, expect
from sqlalchemy import text

from app.config import Settings
from app.db import SessionLocal
from app.rag.importer import ImportSummary, import_sources
from app.rag.llm import LLMClient

REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "docs"

#: Six short grounded questions (the house phrasing + on-topic variants —
#: the honesty gate is HIGH for all of them, so every turn renders an
#: answer bubble + source chips). Six turns overflow the 800px viewport by
#: a wide margin; the tests assert the overflow, so the count can never
#: silently stop being enough.
SHORT_QUESTIONS = (
    "How is my Kubernetes cluster set up?",
    "How do my backups work?",
    "How did I install gitlab?",
    "How does my homelab networking work?",
    "What is in the new-service deployment?",
    "What scripts do I have in the homelab?",
)
#: The phase-11 long-answer trigger: ~900 words ≈ 8 s of streaming — the
#: in-flight window the scrolled-up Stop click rides.
LONG_QUESTION = "How is my Kubernetes cluster set up? write a long answer"
#: The long answer's unique final line — absent from any partial.
LONG_ANSWER_END = "LONG-ANSWER-END"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
STORAGE_KEY = "bor.chat.v1"

#: The conftest ``page`` viewport is 1280×800; the story's phone viewport
#: is the phase-46 one.
MOBILE: ViewportSize = {"width": 375, "height": 812}

#: Tolerance for "flush with the viewport bottom": `env(safe-area-inset-
#: bottom)` resolves to 0 on a desktop/notched-free viewport, so the only
#: slack is sub-pixel rounding. A notched device would shift the box up by
#: the inset — the same few px of headroom the assertion allows.
FLUSH_PX = 4
#: The sticky header's band (phase 12 pins it at 64px) — the pinned
#: composer must never enter it.
HEADER_PX = 64

# The typing indicator is itself a .msg.brain — exclude its bubble.
ANSWER = ".msg.brain .bubble:not(.typing)"


# ---------------------------------------------------------------------------
# KB seeding (same pattern as the phase 42/48/49 story suites)
# ---------------------------------------------------------------------------


async def _import_fixtures(mock_port: int) -> ImportSummary:
    kwargs: dict[str, Any] = {"_env_file": None, "llm_base_url": f"http://127.0.0.1:{mock_port}/v1"}
    settings = Settings(**kwargs)  # pyright: ignore[reportCallIssue]
    return await import_sources([FIXTURES], LLMClient(settings))


def _run_in_thread(coro: Any) -> Any:
    """Run a coroutine on a worker thread.

    Playwright's sync API keeps an asyncio loop running on the test
    thread, so ``asyncio.run`` cannot be called directly from a test body.
    """
    box: dict[str, Any] = {}

    def runner() -> None:
        try:
            box["value"] = asyncio.run(coro)
        except BaseException as e:  # noqa: BLE001 — re-raised on the test thread
            box["error"] = e

    t = Thread(target=runner)
    t.start()
    t.join()
    if "error" in box:
        raise box["error"]
    return box["value"]


def _reset_db(mock_port: int, seed: bool) -> ImportSummary | None:
    """Truncate the KB (and query log), then optionally re-import fixtures."""
    with SessionLocal() as db:
        db.execute(text("TRUNCATE chunks, documents, query_log"))
        db.commit()
    if not seed:
        return None
    return _run_in_thread(_import_fixtures(mock_port))


@pytest.fixture()
def seeded_kb(mock_llm: int, db_ready: None) -> Iterator[None]:
    """A fresh KB seeded from ``tests/fixtures/docs`` (13 docs, A9 formats),
    truncated again on teardown. ``db_ready`` (conftest) skips with clear
    instructions when Postgres is down."""
    summary = _reset_db(mock_llm, seed=True)
    assert summary is not None and summary.added == 13
    yield
    _reset_db(mock_llm, seed=False)


# ---------------------------------------------------------------------------
# Measurement + flow helpers
# ---------------------------------------------------------------------------


def scroll_state(page: Page) -> dict[str, float]:
    """The document scroller's state (there is no inner scroll container)."""
    return page.evaluate(
        "() => ({ y: window.scrollY, "
        "sh: document.documentElement.scrollHeight, "
        "ch: window.innerHeight })"
    )


def wait_settled(page: Page, timeout: int = 30_000) -> None:
    """The turn is over: the label is back to "Send" (phase 48 — the
    in-flight state is the enabled Stop control, so the label carries the
    wait; the explicit timeout covers the mock's ~8 s long turn)."""
    expect(page.locator("#send-btn")).to_be_enabled(timeout=timeout)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=timeout)


def submit(page: Page, question: str) -> None:
    """Submit through the composer (the real-user flow)."""
    page.fill("#message-input", question)
    page.click("#send-btn")
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)


def build_conversation(page: Page, n: int = 6) -> None:
    """Settle ``n`` short grounded turns through the UI (the overflow is
    real conversation content, not injected DOM)."""
    for question in SHORT_QUESTIONS[:n]:
        submit(page, question)
        wait_settled(page)


def box_of(page: Page, selector: str) -> dict[str, float]:
    """The selector's viewport-relative box as a plain dict.

    Measured WITHOUT scrolling (``bounding_box`` never scrolls the page), so
    the answer is "where the box sits where the user left it" — exactly the
    question this story asks. Playwright hands back a FloatRect; the helpers
    below speak ``dict[str, float]``.
    """
    rect = page.locator(selector).bounding_box()
    assert rect is not None, f"{selector} must be rendered (no bounding box)"
    return {
        "x": rect["x"],
        "y": rect["y"],
        "width": rect["width"],
        "height": rect["height"],
    }


def composer_box(page: Page) -> dict[str, float]:
    return box_of(page, "#composer")


def assert_flush_with_viewport_bottom(page: Page) -> dict[str, float]:
    """The composer box is FULLY inside the viewport with its bottom edge
    at the viewport bottom (± ``FLUSH_PX`` — the safe-area inset / sub-pixel
    rounding). This is the whole story: the input is where the user's
    pointer already is, at every scroll position."""
    box = composer_box(page)
    ch = scroll_state(page)["ch"]
    assert box["y"] >= -FLUSH_PX, (
        f"the pinned composer is clipped at the TOP of the viewport "
        f"(y={box['y']:.1f}, viewport={ch:.0f})"
    )
    assert box["y"] + box["height"] <= ch + FLUSH_PX, (
        f"the composer runs BELOW the viewport bottom — it ran away "
        f"(bottom={box['y'] + box['height']:.1f}, viewport={ch:.0f})"
    )
    assert box["y"] + box["height"] >= ch - FLUSH_PX, (
        f"the composer is not pinned to the bottom edge "
        f"(bottom={box['y'] + box['height']:.1f}, viewport={ch:.0f})"
    )
    return box


def assert_no_overlap(a: dict[str, float], b: dict[str, float]) -> None:
    """Two bounding boxes must not overlap vertically."""
    assert a["y"] + a["height"] <= b["y"] + FLUSH_PX, (
        f"boxes overlap: {a} vs {b}"
    )


def _assert_no_error_banner(page: Page) -> None:
    """A stop settles to idle through the same finally path — never the red
    role=alert error banner (phase 48 contract)."""
    banner = page.locator("#kb-banner")
    expect(banner).to_be_hidden()
    expect(banner).not_to_have_attribute("role", "alert")
    expect(banner).not_to_have_class(re.compile(r"is-error"))


def _stored_parsed(page: Page) -> dict[str, Any]:
    raw = page.evaluate(f"() => localStorage.getItem('{STORAGE_KEY}')")
    assert raw is not None, "the conversation key must exist in localStorage"
    return json.loads(raw)


# ---------------------------------------------------------------------------
# 1. Pinned while reading: over-viewport conversation, every scroll
#    position keeps the composer inside the viewport at the bottom edge,
#    and at the document bottom it settles back into flow above the footer
# ---------------------------------------------------------------------------


def test_composer_pinned_at_every_scroll_position(
    page: Page, app_url: str, seeded_kb: None
) -> None:
    page.set_default_timeout(30_000)
    page.goto(app_url)

    build_conversation(page, n=len(SHORT_QUESTIONS))

    # The story's precondition: a conversation longer than the viewport.
    state = scroll_state(page)
    assert state["sh"] > state["ch"] + 200, (
        f"the conversation must overflow the viewport (sh={state['sh']:.0f}, "
        f"ch={state['ch']:.0f}) — more turns are needed"
    )

    # Scrolled to the very top (the "reading the beginning" position): the
    # pinned box sits at the bottom edge of the viewport, fully visible.
    page.evaluate("() => window.scrollTo(0, 0)")
    assert scroll_state(page)["y"] <= 1
    assert_flush_with_viewport_bottom(page)

    # And at every stop through the sticky range — the range is bounded by
    # the composer's CONTAINING BLOCK (`.chat-shell`): the pin holds while
    # that column's bottom edge sits below the viewport bottom, which is
    # every reading position above the last screenful of conversation.
    limits = page.evaluate(
        "() => ({ shellBottom: document.querySelector('.chat-shell')"
        ".getBoundingClientRect().bottom + window.scrollY, "
        "sh: document.documentElement.scrollHeight, ch: window.innerHeight })"
    )
    max_scroll = limits["sh"] - limits["ch"]
    pin_limit = limits["shellBottom"] - limits["ch"]
    assert pin_limit > 200, (
        f"the conversation must keep the composer inside its sticky range "
        f"for at least a screen of scrolling (limit={pin_limit:.0f})"
    )
    assert max_scroll > pin_limit, (
        "the document bottom must leave the sticky range, so the settle "
        "back into flow is covered too"
    )
    for y in (0, pin_limit * 0.25, pin_limit * 0.5, pin_limit * 0.75, pin_limit - 1):
        page.evaluate("y => window.scrollTo(0, y)", y)
        assert abs(scroll_state(page)["y"] - y) <= 1, f"the test scroll to {y} must land"
        assert_flush_with_viewport_bottom(page)

    # Settled at the document bottom: the sticky box has returned to its
    # NORMAL FLOW position — above the footer, not hovering over it (the
    # pin must not float the composer over the footer).
    page.evaluate("() => window.scrollTo(0, document.documentElement.scrollHeight)")
    assert abs(scroll_state(page)["y"] - max_scroll) <= 1, "must land at the document bottom"
    footer_box = box_of(page, ".app-footer")  # the footer must be rendered
    composer = composer_box(page)
    assert_no_overlap(composer, footer_box)
    assert composer["y"] + composer["height"] < scroll_state(page)["ch"] - FLUSH_PX, (
        "at the document bottom the composer settles into flow — it is no "
        "longer glued to the viewport edge"
    )

    # The conversation itself is intact (6 turns, all answers landed) — the
    # pin changed the geometry, not the content.
    expect(page.locator(".msg.user .bubble")).to_have_count(len(SHORT_QUESTIONS))
    expect(page.locator(".msg.brain .bubble")).to_have_count(len(SHORT_QUESTIONS))
    expect(page.locator(".msg.brain .bubble").last).to_contain_text(MOCK_ANSWER_MARKER)


# ---------------------------------------------------------------------------
# 2. The TODO.md L3 scenario: mid-turn, scrolled up reading earlier
#    content, Stop is visible and clickable without scrolling — and the
#    click neither moves the viewport nor breaks the phase-48 contract
# ---------------------------------------------------------------------------


def test_stop_is_reachable_from_scrolled_up(
    page: Page, app_url: str, seeded_kb: None
) -> None:
    page.set_default_timeout(30_000)
    page.goto(app_url)

    # Turn 1 (settled long answer) overflows the viewport — the user now
    # has earlier content to read while turn 2 streams.
    submit(page, LONG_QUESTION)
    wait_settled(page)
    state = scroll_state(page)
    assert state["sh"] > state["ch"] + 200, "a long answer must make the document scrollable"

    # Turn 2 streams (~8 s). Wait until it is observably in flight — the
    # SECOND answer bubble exists and carries deltas (`.last` alone would
    # still resolve to turn 1's finished bubble: the typing indicator is
    # `.bubble.typing` and excluded from ANSWER).
    submit(page, LONG_QUESTION)
    page.wait_for_function(
        "() => { const els = document.querySelectorAll('.msg.brain .bubble:not(.typing)');"
        " return els.length >= 2 && els[els.length - 1].innerText.length > 80; }",
        timeout=30_000,
    )
    answer = page.locator(ANSWER).nth(1)
    expect(page.locator("#send-label")).to_have_text("Stop", timeout=10_000)

    # The user scrolls UP to read earlier content. Phase 42 leaves them
    # there — the app never follows the stream.
    page.evaluate("() => window.scrollTo(0, 0)")
    y0 = scroll_state(page)["y"]
    assert y0 <= 1, "the test scroll to the top must land"
    # The scrolled-up position is inside the composer's sticky range (the
    # long answer + its question still extend past this screenful) — the
    # premise of the owner's scenario.
    assert page.evaluate(
        "() => document.querySelector('.chat-shell').getBoundingClientRect().bottom"
    ) > scroll_state(page)["ch"]

    # ... and the Stop control is right there, WITHOUT scrolling: it is
    # the enabled `.is-stop` button inside the pinned composer, fully
    # inside the viewport. (Before phase 52 the composer sat below the
    # fold here — the "runs away as they try to click stop" bug.)
    btn = page.locator("#send-btn")
    expect(page.locator("#send-label")).to_have_text("Stop")
    expect(btn).to_be_enabled()
    expect(btn).to_have_class(re.compile(r"is-stop"))
    expect(btn).to_be_visible()
    btn_box = box_of(page, "#send-btn")
    ch = scroll_state(page)["ch"]
    assert btn_box["y"] >= 0 and btn_box["y"] + btn_box["height"] <= ch, (
        f"Stop is off-screen while the user reads up-page "
        f"(box={btn_box}, viewport={ch:.0f})"
    )
    # Clicking must not need a scroll: the click lands on Stop itself
    # (Playwright hits the element at its own position, no scrolling).
    assert_flush_with_viewport_bottom(page)

    # Stop it — from the scrolled-up position, no scrolling involved.
    btn.click()

    # The viewport did not move: the pin adds no scroll call site and the
    # stop path never scrolls (phase 42, sampled across the settle).
    samples = [scroll_state(page)["y"]]
    for _ in range(4):
        time.sleep(0.25)
        samples.append(scroll_state(page)["y"])
    assert max(samples) - min(samples) <= 1, (
        f"the Stop click moved the viewport ({samples}) — the reader is "
        "yanked away from the content they were reading"
    )

    # Settled to idle with the phase-48 contract intact.
    wait_settled(page, timeout=10_000)
    expect(btn).not_to_have_class(re.compile(r"is-stop"))
    _assert_no_error_banner(page)
    expect(page.locator("#send-status")).to_contain_text("Answer stopped.")

    # The partial is kept on screen with the "Stopped" marker; the full
    # long answer never arrived.
    stopped_text = answer.inner_text()
    assert stopped_text.strip()
    assert "Step 1:" in stopped_text
    assert LONG_ANSWER_END not in stopped_text
    expect(page.locator(".msg.brain .stopped-note")).to_have_count(1)
    expect(page.locator(".msg.brain .stopped-note").first).to_contain_text("Stopped")

    # ... and it persists through the normal bor.chat.v1 path: a fresh
    # load restores the stopped record (the reader sees it wherever they
    # were reading).
    stored = _stored_parsed(page)
    assert [m["who"] for m in stored["messages"]] == [
        "user",
        "brain",
        "user",
        "brain",
    ]
    last = stored["messages"][-1]
    assert last["stopped"] is True
    assert LONG_ANSWER_END not in last["text"]

    page.reload()
    expect(page.locator(".msg.user .bubble")).to_have_count(2)
    bubble = page.locator(ANSWER)
    expect(bubble).to_have_count(2)
    expect(bubble.nth(1)).to_contain_text("Step 1:")
    expect(page.locator(".msg.brain .stopped-note")).to_have_count(1)
    expect(page.locator(".msg.brain .stopped-note").first).to_contain_text("Stopped")
    expect(bubble.nth(1)).not_to_contain_text(LONG_ANSWER_END)
    wait_settled(page, timeout=10_000)
    # The restored page keeps the pin too (the composer is still the
    # last, sticky child of the column).
    assert composer_box(page)["height"] > 0
    assert page.evaluate(
        "() => getComputedStyle(document.querySelector('#composer')).position"
    ) == "sticky"


# ---------------------------------------------------------------------------
# 3. Natural bottom: an empty/short chat keeps the composer in its flow
#    position — the pin does not float it over the footer or shift the page
# ---------------------------------------------------------------------------


def test_empty_chat_composer_sits_in_normal_flow(
    page: Page, app_url: str, seeded_kb: None
) -> None:
    page.set_default_timeout(30_000)
    page.goto(app_url)

    # A fresh visitor: the empty state, nothing to scroll.
    expect(page.locator("#empty-state")).to_be_visible()
    state = scroll_state(page)
    assert state["sh"] <= state["ch"] + 1, (
        "an empty chat must not be scrollable — there is nothing to pin against"
    )

    composer = composer_box(page)
    assert composer["y"] + composer["height"] <= state["ch"] + FLUSH_PX, (
        "the composer may never hang below the viewport bottom"
    )

    # The footer is present, in normal flow, BELOW the composer: the sticky
    # box settles in its own flow slot (last child of .chat-shell) instead
    # of floating over the footer on a short page.
    footer = page.locator(".app-footer")
    expect(footer).to_be_visible()
    assert_no_overlap(composer, box_of(page, ".app-footer"))

    # A single short turn: still in flow, still nothing to scroll past.
    submit(page, SHORT_QUESTIONS[0])
    wait_settled(page)
    state = scroll_state(page)
    composer = composer_box(page)
    assert composer["y"] + composer["height"] <= state["ch"] + FLUSH_PX
    assert_no_overlap(composer, box_of(page, ".app-footer"))
    expect(page.locator(ANSWER).last).to_contain_text(MOCK_ANSWER_MARKER)


# ---------------------------------------------------------------------------
# 4. ≤640px: the pin holds on a phone and stays UNDER the sticky header
#    (no z-index was added, so the pinned box cannot cover the bar or the
#    hamburger dropdown — phase 46 stacking)
# ---------------------------------------------------------------------------


def test_pin_holds_on_mobile_under_the_header(
    browser: Browser, app_url: str, seeded_kb: None
) -> None:
    page = browser.new_page(viewport=MOBILE)
    try:
        page.set_default_timeout(30_000)
        page.goto(app_url)
        build_conversation(page, n=4)

        state = scroll_state(page)
        assert state["sh"] > state["ch"] + 200, (
            f"four turns must overflow an 812px phone (sh={state['sh']:.0f}, "
            f"ch={state['ch']:.0f})"
        )

        page.evaluate("() => window.scrollTo(0, 0)")
        assert page.evaluate(
            "() => document.querySelector('.chat-shell').getBoundingClientRect().bottom"
        ) > scroll_state(page)["ch"], "the phone position must sit in the sticky range"
        composer = assert_flush_with_viewport_bottom(page)

        # The pinned box never enters the sticky header's band, and the
        # header stays on top: the composer carries NO z-index, so DOM
        # order + the header's z 20 keep the bar (and its dropdown) above.
        assert composer["y"] >= HEADER_PX, (
            f"the pinned composer overlaps the header band "
            f"(y={composer['y']:.1f}, header={HEADER_PX}px)"
        )
        assert page.evaluate(
            "() => getComputedStyle(document.querySelector('#composer')).zIndex"
        ) in {"auto", "0"}, "the composer must not be lifted above the header"
        header_z = page.evaluate(
            "() => getComputedStyle(document.querySelector('.app-header')).zIndex"
        )
        assert header_z == "20"

        # Reachable touch target: the Send button stays ≥44px on a phone.
        btn_box = box_of(page, "#send-btn")
        assert btn_box["height"] >= 44 and btn_box["width"] >= 44

        # The composer is usable there: typing + sending works from the pin.
        submit(page, SHORT_QUESTIONS[1])
        wait_settled(page)
        expect(page.locator(".msg.user .bubble")).to_have_count(5)
    finally:
        page.close()