Experimental/brain-of-reese
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(chat): stop autoscrolling while a reply streams (owner direction)

Browse Source 
TODO.md L5: "Get rid of the chat reply autoscroll, it's breaking things
like making it impossible for the user to scroll while a reply
generates." Owner direction 2026-08-27 (roadmap A1) revises the
phase-18 follow-the-bottom choice: the page NEVER auto-scrolls while a
turn streams. Kept (owner decision): the submit reveal (the user's own
message) and the one-shot phase-14 restore landing.

- frontend/assets/app.js: delete NEAR_BOTTOM_PX + isNearBottom;
  scrollReveal becomes the one unconditional scrollIntoView (still
  smooth, still "auto" under prefers-reduced-motion via SCROLL);
  addMessage(who, html, scroll = false) carries an explicit scroll
  intent — only the submit (", true") and the two restore landings
  scroll. The thinking/tool/delta handlers and the typing indicator
  drop their page-scroll calls; the thinking block's INTERNAL
  bottom-pin (textEl.scrollTop, phase 17 — reworked separately in
  phase 43) and the turn-end focus({ preventScroll: true }) survive.
- tests/unit/test_frontend_scroll.py: rewritten pin for the new
  contract — phase-18 gate absent, helper unconditional, explicit
  intent at submit/restore, no page-scroll call in the streaming
  handlers, typing bubble scroll-free, SCROLL reduced-motion intact.
- tests/unit/test_chat_persistence.py: restore-landing pin updated to
  the new signature (the old forced "auto" is gone; the landing
  rides the default SCROLL — noted at the call site).
- tests/e2e/test_no_reply_autoscroll.py (new, replaces the deleted
  test_follow_bottom_scroll.py): no autoscroll across >=10 samples
  (1px tolerance) during a long answer and during the thinking stream;
  submit-from-the-top still reveals the user message; the restore
  landing lands one-shot on the latest message and stays; long answer
  + sources and the collapsed thinking block persist and restore.

E2E (isolation): test_no_reply_autoscroll.py 5/5; regressions
test_chat_rag 3/3, test_thinking_display 5/5,
test_chat_persistence 4/4, test_long_answers 2/2, test_smoke 3/3;
unit+integration 723 passed, app/ coverage 99%; ruff + pyright clean.
This commit is contained in:
ducoterra
2026-08-28 00:29:25 -04:00
parent 6cf1df9bf2
commit 7c6763319b
5 changed files with 673 additions and 489 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/e2e/test_follow_bottom_scroll.py
-356
View File
@@ -1,356 +0,0 @@
"""Phase 18 E2E (Playwright, mock-only): the chat follows the bottom.
Story: ``.agent/user_stories/follow-bottom-scroll.md``
Run in isolation (DB must be up: ``podman compose up -d db``):
uv run pytest tests/e2e/test_follow_bottom_scroll.py -v --no-cov
The follow-the-bottom contract (owner choice 2026-08-23, option 1 — no
"↓ new content" pill): the page auto-scrolls *only while the user is
already pinned at the bottom*; submitting a question reveals the user's
own message; once the user scrolls up, nothing auto-scrolls for the rest
of the turn (thinking or answer); a restored conversation still lands on
the latest message.
MOCK-ONLY suite: the scenarios key off the deterministic mock's
``write a long answer`` trigger (~900 words ≈ 8s of streaming — a wide,
reliable window to scroll away in) and, for scenario 4, the phase-17
``think out loud`` trigger (both fire independently). ``E2E_REAL_LLM=1``
would make the scroll-away windows unpredictable, so it is not supported
here.
Measurement convention: the scroller is the DOCUMENT — there is no inner
scroll container (``body`` is ``min-height: 100dvh``; the page scrolls on
the window). Scroll position is read via ``page.evaluate`` as
``{ y: window.scrollY, sh: document.documentElement.scrollHeight,
ch: window.innerHeight }``; "near bottom" = ``sh - y - ch <= 200``
(mirrors the frontend's ``NEAR_BOTTOM_PX``); scrolling to the top is
``page.evaluate("() => window.scrollTo(0, 0)")``.
Real-user flow: the user submits from the composer — i.e. pinned at the
bottom (a normal ``fill`` + ``Enter``/click) — and only *after* the
stream starts do they scroll up to read earlier messages. The no-yank
scenarios follow exactly that sequence, so no off-screen input
manipulation is needed (and Playwright's own click/fill auto-scroll
never fires, because the composer is already in view).
Determinism note: the mock paces every SSE frame at 0.02s, so the long
answer streams for several seconds — "mid-stream" assertions land
comfortably inside the window on headless Chromium. Every "held still"
assertion compares against the exact ``scrollTo(0, 0)`` position
(tolerance 5px for rounding).
Test → story mapping (Playwright Mapping Rule):
1. ``test_submit_reveals_new_message``
2. ``test_stream_follows_while_pinned_at_bottom``
3. ``test_no_yank_while_scrolled_up_during_answer_stream``
4. ``test_no_yank_while_scrolled_up_during_thinking``
5. ``test_restore_lands_on_latest_message``
"""
from __future__ import annotations
import asyncio
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 Page, 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"
#: Mirror of app.js's exported ``NEAR_BOTTOM_PX`` (the 200px composer-zone
#: band that counts as "pinned to the bottom").
NEAR_BOTTOM_PX = 200
#: Mock long-answer trigger (~900 words ≈ 8s of streaming at the mock's
#: 0.02s/frame pace) — the wide, deterministic window to scroll away in.
LONG_QUESTION = "write a long answer about my kubernetes cluster"
#: Phase-17 thinking prefix + the long-answer trigger: both mock triggers
#: fire independently (a ~4.5s reasoning stream — lengthened in phase 21 —
#: then the long answer).
THINK_LONG_QUESTION = "think out loud — write a long answer about my kubernetes cluster"
#: The mock long answer's unique final line (mock_llm.LONG_ANSWER_END) —
#: proves the whole stream landed even while the viewport was at the top.
LONG_ANSWER_END = "LONG-ANSWER-END"
#: Line fragment the mock's deterministic scratchpad carries
#: (mock_llm.compose_thinking) — same key phase 17's suite uses.
THINKING_FRAGMENT = "Step 2: Check my notes"
#: Tolerance for "the viewport held still at the top" (rounding).
HOLD_TOLERANCE_PX = 5
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`` (8 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 == 8
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 near_bottom(state: dict[str, float]) -> bool:
"""Mirror of app.js's ``isNearBottom`` — the NEAR_BOTTOM_PX band."""
return state["sh"] - state["y"] - state["ch"] <= NEAR_BOTTOM_PX
def held_at_top(page: Page) -> bool:
"""The viewport has not moved from ``window.scrollTo(0, 0)`` (±5px)."""
return scroll_state(page)["y"] <= HOLD_TOLERANCE_PX
def wait_settled(page: Page) -> None:
"""The turn is over: the never-stale contract re-enabled the button."""
expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000)
expect(page.locator("#send-label")).to_have_text("Send")
def submit(page: Page, question: str) -> None:
"""Submit from the composer — the real-user flow (pinned at the
bottom, so Playwright's click/fill auto-scroll never kicks in)."""
page.fill("#message-input", question)
page.click("#send-btn")
expect(page.locator(".msg.user .bubble").last).to_contain_text(question)
def brain_bubble_longer_than(n: int) -> str:
"""JS predicate: the LAST brain bubble's rendered text is > n chars
(i.e. that far into the stream)."""
return (
"() => { const els = document.querySelectorAll('.msg.brain .bubble');"
f" const el = els[els.length - 1]; return !!el && el.innerText.length > {n}; }}"
)
def brain_message_in_view(page: Page) -> bool:
"""The last brain message intersects the viewport vertically. Partial
visibility counts: a long answer is taller than the window, and the
contract is that it is revealed (its lower edge in view), not that it
fits."""
box = page.locator(".msg.brain").last.bounding_box()
if box is None:
return False
ch = scroll_state(page)["ch"]
return box["y"] < ch and box["y"] + box["height"] > 0
# ---------------------------------------------------------------------------
# 1. Submit: the user's message and the answer reveal into view
# ---------------------------------------------------------------------------
def test_submit_reveals_new_message(page: Page, app_url: str, seeded_kb: None) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# A fresh chat page starts pinned at the bottom (short conversation —
# the composer, i.e. the user, sits in the band).
assert near_bottom(scroll_state(page))
submit(page, LONG_QUESTION)
wait_settled(page)
# The last brain message is inside the viewport ...
assert brain_message_in_view(page), (
"the answer must be revealed — the last brain message is not in view"
)
# ... and the page is still pinned at the bottom.
assert near_bottom(scroll_state(page))
# ---------------------------------------------------------------------------
# 2. Follow: while pinned, the page keeps up with the stream
# ---------------------------------------------------------------------------
def test_stream_follows_while_pinned_at_bottom(page: Page, app_url: str, seeded_kb: None) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
submit(page, LONG_QUESTION)
# ~2s into the stream: the answer bubble already carries >200 chars
# (the mock paces frames at 0.02s).
page.wait_for_function(brain_bubble_longer_than(200), timeout=30_000)
# Let the smooth follow scroll settle before measuring.
time.sleep(0.3)
# The follow behavior is alive — not accidentally removed.
assert near_bottom(scroll_state(page)), (
"the page must follow the stream while the user is pinned at the bottom"
)
wait_settled(page)
assert near_bottom(scroll_state(page))
# ---------------------------------------------------------------------------
# 3. No yank: scrolled up mid-ANSWER — the viewport holds for the rest
# of the turn (the answer finishes off-screen below, by design)
# ---------------------------------------------------------------------------
def test_no_yank_while_scrolled_up_during_answer_stream(
page: Page, app_url: str, seeded_kb: None
) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# Turn 1 (settled) makes the document overflow the 800px viewport.
submit(page, LONG_QUESTION)
wait_settled(page)
state = scroll_state(page)
assert state["sh"] > state["ch"], "a long answer must make the document scrollable"
assert near_bottom(state), "follow was active: the settled turn ends pinned"
# Turn 2: submit from the composer (pinned — normal flow), then let
# the new answer stream a bit.
submit(page, LONG_QUESTION)
page.wait_for_function(brain_bubble_longer_than(200), timeout=30_000)
# The user goes up to read while the stream is running.
page.evaluate("() => window.scrollTo(0, 0)")
# The stream kept running at the top ...
page.wait_for_function(brain_bubble_longer_than(600), timeout=30_000)
assert held_at_top(page), "the viewport must hold still while scrolled up"
# ... and nothing scrolls for the rest of the turn — the answer
# finishes off-screen below, by design.
wait_settled(page)
assert held_at_top(page)
# ---------------------------------------------------------------------------
# 4. No yank: scrolled up during THINKING — the whole reasoning stream
# plus the answer's start happen at the top
# ---------------------------------------------------------------------------
def test_no_yank_while_scrolled_up_during_thinking(
page: Page, app_url: str, seeded_kb: None
) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# One settled turn first, so the document overflows (scrollable).
submit(page, LONG_QUESTION)
wait_settled(page)
# The thinking turn: submit pinned (normal flow) ...
submit(page, THINK_LONG_QUESTION)
details = page.locator(".msg.brain").last.locator("details.thinking")
details.wait_for(state="attached", timeout=10_000)
# ... and, while the reasoning stream is still open (phase-17
# behavior: created open, ~1.3s before the first answer token) ...
expect(details).to_have_attribute("open", "")
# ... the user goes up to read.
page.evaluate("() => window.scrollTo(0, 0)")
# The whole thinking stream plus the answer's start happen at the top.
bubble = page.locator(".msg.brain").last.locator(".bubble")
expect(bubble).not_to_have_text("", timeout=30_000)
assert held_at_top(page), "the viewport must hold still during thinking"
# Settled: still at the top, and everything landed (off-screen,
# which is the point of the story).
wait_settled(page)
assert held_at_top(page)
expect(details.locator(".thinking-text")).to_contain_text(THINKING_FRAGMENT)
expect(bubble).to_contain_text(LONG_ANSWER_END)
# ---------------------------------------------------------------------------
# 5. Restore: the one-shot landing still puts the latest message in view
# (phase 14 behavior preserved — pinned so a future "remove all
# scrolling" change fails loudly instead of silently)
# ---------------------------------------------------------------------------
def test_restore_lands_on_latest_message(page: Page, app_url: str, seeded_kb: None) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# Two settled turns (user + brain × 2) — the document overflows.
submit(page, LONG_QUESTION)
wait_settled(page)
submit(page, LONG_QUESTION)
wait_settled(page)
page.reload()
# Restore re-renders from localStorage; wait until the last restored
# brain answer is fully back.
expect(page.locator(".msg.brain .bubble").last).to_contain_text(
LONG_ANSWER_END, timeout=30_000
)
wait_settled(page)
expect(page.locator(".msg.user .bubble")).to_have_count(2)
state = scroll_state(page)
assert state["sh"] > state["ch"]
# The forced one-shot landing (the only `force`d scrolls) puts the
# last brain message back in view ...
assert brain_message_in_view(page), (
"a restored conversation must land on its latest message"
)
# ... and the page sits at the bottom.
assert near_bottom(state)
tests/e2e/test_no_reply_autoscroll.py
+496
View File
@@ -0,0 +1,496 @@
"""Phase 42 E2E (Playwright, mock-only): the chat NEVER autoscrolls.
Story: ``.agent/user_stories/no-reply-autoscroll.md`` — owner direction
2026-08-27 (TODO.md L5) revising the phase-18 follow-the-bottom choice:
"Get rid of the chat reply autoscroll, it's breaking things like making
it impossible for the user to scroll while a reply generates."
Run in isolation (DB must be up: ``podman compose up -d db``):
uv run pytest tests/e2e/test_no_reply_autoscroll.py -v --no-cov
This is the INVERSE of the phase-18 contract: while a turn streams
(thinking, tool, or answer frames), nothing moves the viewport — a user
reading earlier content stays exactly where they put it for the rest of
the turn. The only scrolls left in the app are user intent: the submit
(the user's own message is revealed) and the phase-14 restore landing
(one-shot, load-time). The phase-18 suite
(``tests/e2e/test_follow_bottom_scroll.py``) is deleted with this one —
its behavior was intentionally removed.
MOCK-ONLY suite: the scenarios key off the deterministic mock's
``write a long answer`` trigger (~900 words ≈ 8–11s of streaming — a
wide, reliable window to scroll away in) and the phase-17
``think out loud`` trigger (~4.5s reasoning stream). ``E2E_REAL_LLM=1``
would make the scroll-away windows unpredictable, so it is not supported
here.
Measurement convention: the scroller is the DOCUMENT — there is no inner
scroll container. ``window.scrollY`` is read via ``page.evaluate``;
"stable" means every sample is within 1px of every other sample (the
story's tolerance). The stylesheet sets no ``scroll-behavior``, so
``window.scrollTo(0, y)`` is instant — the recorded position is the exact
position the stream must not move.
Test → story mapping (Playwright Mapping Rule):
1. ``test_no_autoscroll_during_long_answer``
2. ``test_no_autoscroll_during_thinking``
3. ``test_submit_reveals_user_message``
4. ``test_restore_landing_one_shot``
5. ``test_answer_content_intact``
"""
from __future__ import annotations
import asyncio
import json
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 Page, 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"
#: Mock long-answer trigger (~900 words ≈ 8–11s of streaming at the mock's
#: 0.02s/frame pace) — the wide, deterministic window to scroll away in.
LONG_QUESTION = "write a long answer about my kubernetes cluster"
#: Phase-17 thinking trigger: a ~4.5s reasoning stream, then a short
#: grounded mock answer (both fire independently of the long trigger).
THINKING_QUESTION = "think out loud about my kubernetes cluster"
#: Short grounded question (phase-14 marker answer).
SHORT_QUESTION = "How is my Kubernetes cluster set up?"
#: The mock long answer's unique final line (mock_llm.LONG_ANSWER_END) —
#: proves the whole stream landed even while the viewport was up.
LONG_ANSWER_END = "LONG-ANSWER-END"
#: Line fragment the mock's deterministic scratchpad carries
#: (mock_llm.compose_thinking) — same key phase 17's suite uses.
THINKING_FRAGMENT = "Step 2: Check my notes"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
STORAGE_KEY = "bor.chat.v1"
#: The story's stability tolerance: the viewport must not move more than
#: 1px while a turn streams with the user scrolled away.
STABLE_PX = 1
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`` (8 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 == 8
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 brain_bubble_longer_than(n: int, min_bubbles: int = 1) -> str:
"""JS predicate: there are at least ``min_bubbles`` brain bubbles and
the LAST one's rendered text is > n chars (i.e. that far into the
stream). ``min_bubbles=2`` guards a second turn: before its first
delta, ``.last`` would still be the PREVIOUS turn's bubble."""
return (
"() => { const els = document.querySelectorAll('.msg.brain .bubble');"
f" return els.length >= {min_bubbles} && els[els.length - 1].innerText.length > {n}; }}"
)
def wait_settled(page: Page) -> None:
"""The turn is over: the never-stale contract re-enabled the button."""
expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000)
expect(page.locator("#send-label")).to_have_text("Send")
def wait_scroll_still(page: Page, timeout: float = 10.0) -> float:
"""window.scrollY once the viewport has stopped moving (two consecutive
reads within STABLE_PX). The submit's smooth reveal and the restore
landing's smooth scroll are the only animations left — both settle
this way before any stream measurement begins."""
deadline = time.monotonic() + timeout
prev: float | None = None
while True:
y = scroll_state(page)["y"]
if prev is not None and abs(y - prev) <= STABLE_PX:
return y
prev = y
if time.monotonic() >= deadline:
raise AssertionError("the viewport did not settle within timeout")
time.sleep(0.25)
def submit(page: Page, question: str) -> None:
"""Submit through the composer (the real-user flow). Playwright's
fill/click scroll the composer into view first — a user-initiated
move, never an app scroll."""
page.fill("#message-input", question)
page.click("#send-btn")
expect(page.locator(".msg.user .bubble").last).to_contain_text(question)
def submit_from_top(page: Page, question: str) -> None:
"""Submit with the viewport where the user left it (the very top).
``page.fill``/``page.click`` would scroll the composer into view
first — which IS the viewport move under test — so the send goes
through the page's own DOM: set the value, fire ``input`` (autoGrow),
click the submit button. A JS click never scrolls the page."""
page.evaluate(
"""(q) => {
const input = document.querySelector('#message-input');
input.value = q;
input.dispatchEvent(new Event('input', { bubbles: true }));
document.querySelector('#send-btn').click();
}""",
question,
)
expect(page.locator(".msg.user .bubble").last).to_contain_text(question)
def user_message_in_view(page: Page) -> bool:
"""The LAST user message's box is fully inside the viewport
(the submit reveal aligns it to the bottom edge)."""
box = page.locator(".msg.user").last.bounding_box()
if box is None:
return False
ch = scroll_state(page)["ch"]
return box["y"] >= -1 and box["y"] + box["height"] <= ch + 1
# ---------------------------------------------------------------------------
# 1. No autoscroll: scrolled up mid-ANSWER — the viewport holds for the
# rest of the turn (the answer finishes off-screen below, by design)
# ---------------------------------------------------------------------------
def test_no_autoscroll_during_long_answer(
page: Page, app_url: str, seeded_kb: None
) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# Turn 1 (settled) makes the document overflow the 800px viewport.
submit(page, LONG_QUESTION)
wait_settled(page)
state = scroll_state(page)
assert state["sh"] > state["ch"], "a long answer must make the document scrollable"
# Turn 2: the same long answer. The submit reveals the user's message
# (the one kept app scroll) — let that smooth reveal settle first.
submit(page, LONG_QUESTION)
page.wait_for_function(brain_bubble_longer_than(200, min_bubbles=2), timeout=30_000)
y0 = wait_scroll_still(page)
# The user scrolls UP ~2× the answer's current height to read earlier
# context while the stream is still running.
box = page.locator(".msg.brain").last.bounding_box()
assert box is not None
target = max(0.0, y0 - 2 * box["height"])
assert target <= y0 - STABLE_PX, "the scroll-up must actually move the viewport"
page.evaluate("y => window.scrollTo(0, y)", target)
assert abs(scroll_state(page)["y"] - target) <= STABLE_PX
# Sample the viewport across the rest of the stream ...
samples: list[float] = []
mid_stream = 0
deadline = time.monotonic() + 40
while time.monotonic() < deadline:
time.sleep(0.25)
samples.append(scroll_state(page)["y"])
if not page.locator("#send-btn").is_enabled():
mid_stream += 1
if len(samples) >= 10 and page.locator("#send-btn").is_enabled():
# ... and a few more AFTER `done` (the turn is over; nothing
# queued behind the stream may move the page either).
for _ in range(3):
time.sleep(0.3)
samples.append(scroll_state(page)["y"])
break
else:
raise AssertionError("the long turn did not settle within 40s")
assert mid_stream >= 8, "the samples must land while the stream is running"
spread = max(samples) - min(samples)
assert spread <= STABLE_PX, (
f"the viewport moved {spread:.1f}px while the user was scrolled up "
"(no-reply-autoscroll contract)"
)
# The whole answer still landed (off-screen below — by design).
expect(page.locator(".msg.brain .bubble").last).to_contain_text(LONG_ANSWER_END)
# ---------------------------------------------------------------------------
# 2. No autoscroll: scrolled up during THINKING — the whole reasoning
# stream plus the answer's start happen with the viewport held
# ---------------------------------------------------------------------------
def test_no_autoscroll_during_thinking(
page: Page, app_url: str, seeded_kb: None
) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# One settled long turn so the document overflows (scrollable).
submit(page, LONG_QUESTION)
wait_settled(page)
# The thinking turn: the submit reveals the user's message (the kept
# app scroll) ...
submit(page, THINKING_QUESTION)
details = page.locator(".msg.brain").last.locator("details.thinking")
details.wait_for(state="attached", timeout=10_000)
expect(details).to_have_attribute("open", "") # created open (phase 17)
# ... and, once the reasoning stream is clearly running ...
page.wait_for_function(
"() => { const el = document.querySelector('details.thinking .thinking-text');"
" return !!el && el.innerText.length > 300; }",
timeout=30_000,
)
# ... let the submit's smooth reveal settle, then the user goes up.
wait_scroll_still(page)
page.evaluate("() => window.scrollTo(0, 0)")
# Sample across the remaining thinking stream: no per-chunk follow.
samples: list[float] = []
open_samples = 0
deadline = time.monotonic() + 25
while time.monotonic() < deadline:
time.sleep(0.3)
state = page.evaluate(
"""() => {
const block = document.querySelector('details.thinking');
const wrap = block ? block.closest('.msg.brain') : null;
const el = wrap ? wrap.querySelector('.bubble') : null;
return { y: window.scrollY,
open: !!(block && block.open),
bubble: el ? el.innerText.length : 0 };
}"""
)
samples.append(state["y"])
if state["open"]:
open_samples += 1
if state["bubble"] > 0 and len(samples) >= 8:
break
else:
raise AssertionError("the first answer token never arrived")
assert open_samples >= 6, "the samples must land while the thinking stream is open"
spread = max(samples) - min(samples)
assert spread <= STABLE_PX, (
f"the viewport moved {spread:.1f}px during the thinking stream "
"(no per-chunk page follow)"
)
# Settled: still at the top, everything landed (off-screen, which is
# the point of the story).
wait_settled(page)
assert abs(scroll_state(page)["y"]) <= STABLE_PX
expect(details.locator(".thinking-text")).to_contain_text(THINKING_FRAGMENT)
expect(page.locator(".msg.brain .bubble").last).to_contain_text(MOCK_ANSWER_MARKER)
# ---------------------------------------------------------------------------
# 3. Submit: scrolled to the very top, sending a question still reveals
# the user's own message (the kept, user-initiated scroll)
# ---------------------------------------------------------------------------
def test_submit_reveals_user_message(
page: Page, app_url: str, seeded_kb: None
) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# A populated conversation that overflows the viewport.
submit(page, LONG_QUESTION)
wait_settled(page)
state = scroll_state(page)
assert state["sh"] > state["ch"], "a long answer must make the document scrollable"
# The user is reading at the very top ...
page.evaluate("() => window.scrollTo(0, 0)")
assert scroll_state(page)["y"] <= STABLE_PX
# ... and sends a question without first scrolling down.
submit_from_top(page, SHORT_QUESTION)
# The submit's reveal is the kept app scroll: the user's own message
# ends up in view (its box fully inside the viewport).
deadline = time.monotonic() + 10
while time.monotonic() < deadline and not user_message_in_view(page):
time.sleep(0.2)
assert user_message_in_view(page), (
"the submit must reveal the user's message — its box is not in the viewport"
)
# The turn completes; the answer lands off-screen below, but the user
# message stays revealed (nothing re-positions it).
wait_settled(page)
assert user_message_in_view(page)
expect(page.locator(".msg.brain .bubble").last).to_contain_text(MOCK_ANSWER_MARKER)
# ---------------------------------------------------------------------------
# 4. Restore landing (phase 14, owner-kept): a reload lands one-shot on
# the latest message and stays there while idle
# ---------------------------------------------------------------------------
def test_restore_landing_one_shot(
page: Page, app_url: str, seeded_kb: None
) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# Settle a conversation (phase-14 persistence): long + short turns.
submit(page, LONG_QUESTION)
wait_settled(page)
submit(page, SHORT_QUESTION)
wait_settled(page)
time.sleep(0.5) # let the persistence writes land before the reload
page.reload()
# Restore re-renders from localStorage; wait until the last restored
# brain answer (the short one — the long answer carries no mock
# marker) is fully back.
expect(page.locator(".msg.brain .bubble").last).to_contain_text(
MOCK_ANSWER_MARKER, timeout=30_000
)
# The one-shot landing rides a smooth scroll — let it settle ...
wait_scroll_still(page)
state = scroll_state(page)
assert state["sh"] > state["ch"]
# ... and it lands on the latest message: its bubble is in view, in
# the lower half of the viewport (the chips + composer sit just
# below the fold — the landing predates them by design).
box = page.locator(".msg.brain").last.locator(".bubble").bounding_box()
assert box is not None
assert box["y"] < state["ch"] and box["y"] + box["height"] >= state["ch"] * 0.6, (
"the restore landing must put the latest message in view"
)
# ... and STAYS: idle samples (no stream active) never move.
samples = [state["y"]]
for _ in range(4):
time.sleep(0.4)
samples.append(scroll_state(page)["y"])
spread = max(samples) - min(samples)
assert spread <= STABLE_PX, (
f"the restored page moved {spread:.1f}px while idle"
)
# ---------------------------------------------------------------------------
# 5. Content intact (regression): the long answer completes with sources;
# a thinking turn persists + restores its collapsed block (phase 17)
# ---------------------------------------------------------------------------
def test_answer_content_intact(page: Page, app_url: str, seeded_kb: None) -> None:
page.set_default_timeout(30_000)
page.goto(app_url)
# The long answer streams to completion with its sources ...
submit(page, LONG_QUESTION)
wait_settled(page)
expect(page.locator(".msg.brain .bubble").last).to_contain_text(LONG_ANSWER_END)
expect(
page.locator(".msg.brain .source-chip", has_text="kubernetes.md")
).to_have_count(1)
# ... and a thinking turn completes with its block auto-collapsed
# (phase 17: open while streaming, closed from the first delta on).
submit(page, THINKING_QUESTION)
wait_settled(page)
details = page.locator(".msg.brain").last.locator("details.thinking")
expect(details).not_to_have_attribute("open")
expect(details.locator(".thinking-text")).to_contain_text(THINKING_FRAGMENT)
expect(page.locator(".msg.brain .bubble").last).to_contain_text(MOCK_ANSWER_MARKER)
# Persistence: four messages, the thinking text + sources stored raw.
raw = page.evaluate(f"() => localStorage.getItem('{STORAGE_KEY}')")
stored = json.loads(raw)
assert [m["who"] for m in stored["messages"]] == ["user", "brain", "user", "brain"]
assert LONG_ANSWER_END in stored["messages"][1]["text"]
assert THINKING_FRAGMENT in stored["messages"][3]["thinking"]
assert any(
s["path"] == "homelab/kubernetes.md" for s in stored["messages"][3]["sources"]
)
# Restore: the long answer (with its chip) and the COLLAPSED thinking
# block come back intact.
page.reload()
expect(page.locator(".msg.user .bubble")).to_have_count(2)
expect(page.locator(".msg.brain .bubble")).to_have_count(2)
expect(page.locator(".msg.brain .bubble").first).to_contain_text(LONG_ANSWER_END)
expect(
page.locator(".msg.brain .source-chip", has_text="kubernetes.md")
).to_have_count(2)
restored = page.locator(".msg.brain").last.locator("details.thinking")
expect(restored).not_to_have_attribute("open")
expect(restored.locator(".thinking-text")).to_contain_text(THINKING_FRAGMENT)
wait_settled(page)
tests/unit/test_chat_persistence.py
+7 -4
View File
@@ -85,10 +85,13 @@ def test_raw_text_only_stored_and_re_rendered_on_restore() -> None:
on restore) — no HTML is ever stored. Restore re-applies the full
brain-message chrome: is-deflected styling, maybe-try chips, sources."""
js = _js()
# Phase 18: restore landings are forced ("auto" + force) one-shot
# scrollReveal calls — the only forced scrolls in the app.
assert 'addMessage("user", renderMarkdown(m.text), "auto", true)' in js
assert 'addMessage("brain", renderMarkdown(m.text), "auto", true)' in js
# Phase 42 (owner direction 2026-08-27): the reply autoscroll is gone;
# restore landings keep their one-shot load-time scroll via the
# explicit intent (scroll=true) — the new addMessage signature has no
# per-call behavior override (default SCROLL instead of forced
# "auto" — documented at the call site).
assert 'addMessage("user", renderMarkdown(m.text), true)' in js
assert 'addMessage("brain", renderMarkdown(m.text), true)' in js
assert "wrap.classList.add(\"is-deflected\")" in js
assert "appendMaybeTry(wrap, m.suggestions)" in js
assert "appendSources(wrap, m.sources)" in js
tests/unit/test_frontend_scroll.py
+124 -83
View File
@@ -1,11 +1,21 @@
"""Unit: the follow-the-bottom scroll contract in the static frontend
(phase 18, owner choice 2026-08-23).
"""Unit: the no-reply-autoscroll contract in the static frontend
(phase 42, owner direction 2026-08-27, TODO.md L5).
The JS behavior itself is E2E-covered (tests/e2e/test_follow_bottom_scroll.py);
here we pin the exported band constant and the single-gate markers that the
story depends on — scrollIntoView appears exactly once in app.js, inside
scrollReveal — so a silent regression back to unconditional per-delta /
per-chunk scrolls is caught without a browser.
The owner removed the phase-18 follow-the-bottom auto-follow: the page
NEVER auto-scrolls while a turn streams (thinking / tool / delta frames
all leave the viewport alone), so a user reading earlier content is no
longer yanked down mid-answer. Scrolls happen only on explicit user
intent: the submit (the user's own message is revealed) and the phase-14
restore landing (one-shot, load-time).
The JS behavior itself is E2E-covered
(tests/e2e/test_no_reply_autoscroll.py); here we pin the source markers
of the new contract — the phase-18 gate is gone (no NEAR_BOTTOM_PX /
isNearBottom), scrollReveal scrolls unconditionally and is the single
scrollIntoView in app.js, addMessage takes an explicit `scroll` intent,
and the streaming handlers contain no page-scroll call at all — so a
silent regression back to per-frame autoscroll is caught without a
browser.
"""
from __future__ import annotations
@@ -28,109 +38,140 @@ def _fn_body(js: str, name: str) -> str:
return js[fn : js.find("\n}\n", fn)]
def test_near_bottom_constant_exported_at_200px() -> None:
"""The "pinned to the bottom" band (the composer zone) must be an
*exported* constant — unit-pinned, same pattern as TURN_TIMEOUT_MS."""
def test_phase18_gate_is_gone() -> None:
"""The follow-the-bottom machinery (phase 18) is removed by owner
direction 2026-08-27: no band constant, no gate function, and no
document-scroller measurement left anywhere in app.js."""
js = _js()
assert re.search(r"export\s+const\s+NEAR_BOTTOM_PX\s*=\s*200\s*;", js), (
"app.js must export `const NEAR_BOTTOM_PX = 200`"
assert "NEAR_BOTTOM_PX" not in js, "the 200px band constant must be gone"
assert "isNearBottom" not in js, "the pinned-to-bottom gate must be gone"
assert "window.scrollY" not in js, (
"nothing in app.js measures the page scroll offset anymore"
)
def test_is_near_bottom_uses_document_scroller() -> None:
"""isNearBottom measures the DOCUMENT scroller (there is no inner
scroll container — the page scrolls on the window): distance from the
bottom of the document <= NEAR_BOTTOM_PX."""
js = _js()
body = _fn_body(js, "isNearBottom")
for ref in (
"documentElement.scrollHeight",
"window.scrollY",
"window.innerHeight",
"NEAR_BOTTOM_PX",
):
assert ref in body, f"isNearBottom must reference {ref!r}"
assert "<=" in body, "the pinned band is an upper bound, not exact equality"
def test_single_scroll_gate() -> None:
"""scrollReveal is the ONE scroll call site in app.js: it fires only
when forced or when the user is pinned to the bottom, keeps
`block: "end"`, and both addMessage (behavior + force passthrough) and
addTyping (defaults) delegate to it."""
def test_scroll_helper_is_unconditional() -> None:
"""scrollReveal is still the ONE scrollIntoView in app.js, but it now
scrolls unconditionally — no force-or-near-bottom condition in its
body, and the phase-18 `force` parameter is gone. A page scroll can
only ever happen where scrollReveal is CALLED (submit + restore)."""
js = _js()
body = _fn_body(js, "scrollReveal")
assert "force || isNearBottom()" in body, "gate: force OR pinned to the bottom"
assert "scrollIntoView" in body
assert 'block: "end"' in body
# The regression pin: exactly one scrollIntoView in the whole file, and
# it lives inside scrollReveal.
assert js.count("scrollIntoView") == 1, (
assert "if (" not in body, "the helper must have no gate — it scrolls when called"
assert "force" not in body, "the phase-18 force parameter must be gone"
# Still smooth / reduced-motion-aware through the default behavior.
assert "behavior = SCROLL" in body
# The regression pin: exactly one actual scrollIntoView CALL in the
# whole file, and it lives inside scrollReveal (the word may appear
# in comments; the call must not).
assert js.count(".scrollIntoView(") == 1, (
"app.js must call scrollIntoView exactly once (inside scrollReveal)"
)
assert js.find("scrollIntoView") > js.find("function scrollReveal")
# addMessage passes its behavior/force through; addTyping uses defaults.
add_body = _fn_body(js, "addMessage")
assert "scrollReveal(wrap, scrollBehavior, force)" in add_body
assert "force = false" in add_body
typing_body = _fn_body(js, "addTyping")
assert "scrollReveal(wrap)" in typing_body
assert js.find(".scrollIntoView(") > js.find("function scrollReveal")
def test_submit_reveal_is_gated() -> None:
"""Submit keeps the plain default call — no force: the gate decides,
and it does in real use because submitting from the composer means the
user is pinned (inside the 200px band); a submit with the viewport away
from the bottom does not yank it."""
def test_add_message_takes_explicit_scroll_intent() -> None:
"""addMessage(who, html, scroll = false): the phase-18
scrollBehavior/force parameters are gone; the bubble scrolls only
when the caller explicitly asks (submit reveal, restore landing)."""
js = _js()
body = _fn_body(js, "addMessage")
assert "function addMessage(who, html, scroll = false)" in body
assert "if (scroll) scrollReveal(wrap)" in body
assert "force" not in body
assert "scrollBehavior" not in body
def test_submit_reveals_user_message() -> None:
"""User intent kept by the owner: submitting scrolls the viewport down
so the user's own message is visible — the submit addMessage passes
the scroll intent; the streaming brain-bubble creations in the same
function never do."""
js = _js()
send = js.find("async function handleSend")
assert send != -1, "handleSend must exist"
call = 'addMessage("user", renderMarkdown(text));'
idx = js.find(call, send)
assert idx != -1, "handleSend must reveal the user message via the plain default"
assert 'addMessage("user", renderMarkdown(text),' not in js, (
"the submit call must not pass a third/fourth argument (no force)"
body = js[send : js.find("\n}\n", send)]
assert 'addMessage("user", renderMarkdown(text), true)' in body, (
"the submit must reveal the user message (scroll intent true)"
)
for call in re.findall(r'addMessage\("brain"([^)]*)\)', body):
assert "true" not in call, (
f"streaming brain bubbles must not scroll the page: {call!r}"
)
def test_restore_force_landing() -> None:
"""Both restore call sites are the only `force`d scrolls: one-shot,
non-smooth ("auto") landing on the last restored message (phase-14
behavior preserved)."""
def test_streaming_handlers_never_scroll_the_page() -> None:
"""The heart of the phase-42 contract: the thinking / tool / delta
branches contain NO page-scroll call (no scrollReveal, no raw
scrollIntoView). The thinking branch keeps the block-INTERNAL pin
(textEl.scrollTop — phase 17, reworked in phase 43): that scrolls
the block's own clip, not the page."""
js = _js()
think = js.find('ev.type === "thinking"')
tool = js.find('ev.type === "tool"')
delta = js.find('ev.type === "delta"')
done = js.find('ev.type === "done"')
assert -1 < think < tool < delta < done, "the turn handler must branch in order"
for name, branch in (
("thinking", js[think:tool]),
("tool", js[tool:delta]),
("delta", js[delta:done]),
):
assert "scrollReveal" not in branch, f"the {name} branch must not scroll the page"
assert ".scrollIntoView(" not in branch, (
f"the {name} branch must not scroll the page"
)
# The thinking window pin survives (phase 17 — untouched by this phase).
thinking_branch = js[think:delta]
assert "textEl.scrollTop = textEl.scrollHeight" in thinking_branch
assert js.count("textEl.scrollTop = textEl.scrollHeight") == 1
def test_restore_landing_is_one_shot() -> None:
"""The phase-14 restore landing keeps its one-shot scroll
(owner-kept): both restore call sites pass the explicit scroll
intent and they are the only two restore scrolls; with the submit's
single reveal, exactly three `true` intents exist in the whole file.
The old forced "auto" landing is gone, and the one-shot, load-time
contract is documented at the call site."""
js = _js()
body = _fn_body(js, "renderStoredMessage")
assert 'addMessage("user", renderMarkdown(m.text), "auto", true)' in body
assert 'addMessage("brain", renderMarkdown(m.text), "auto", true)' in body
# Forced restores are restore-only: exactly two ("auto", true) sites.
assert js.count('"auto", true') == 2, "only the two restore calls may force"
assert 'addMessage("user", renderMarkdown(m.text), true)' in body
assert 'addMessage("brain", renderMarkdown(m.text), true)' in body
assert js.count('"auto", true') == 0, "the old forced 'auto' landing must be gone"
# Submit reveal + the two restore landings — nothing else scrolls.
assert js.count(", true)") == 3, "only submit + the two restore calls may scroll"
# The marker comment documents the one-shot, load-time contract.
assert "restore landing" in body
assert "one-shot" in body
def test_streaming_scrolls_only_through_gate() -> None:
"""The per-chunk scrolls that used to yank the viewport (the phase-17
thinking branch and the streaming delta branch) now go through
scrollReveal with no raw scrollIntoView at either call site; the
block's internal bottom-pinning (its own overflow, not the page) stays."""
def test_typing_bubble_does_not_scroll() -> None:
"""A typing indicator appearing must not yank the page — the phase-18
scrollReveal call in addTyping is removed with the gate."""
js = _js()
thinking_idx = js.find('ev.type === "thinking"')
delta_idx = js.find('ev.type === "delta"')
done_idx = js.find('ev.type === "done"')
assert -1 < thinking_idx < delta_idx < done_idx
thinking_branch = js[thinking_idx:delta_idx]
delta_branch = js[delta_idx:done_idx]
assert "scrollReveal(wrap)" in thinking_branch
assert "scrollReveal(wrap)" in delta_branch
assert "scrollIntoView" not in thinking_branch
assert "scrollIntoView" not in delta_branch
assert "textEl.scrollTop = textEl.scrollHeight" in thinking_branch
body = _fn_body(js, "addTyping")
assert "scrollReveal" not in body
assert ".scrollIntoView(" not in body
def test_scroll_constant_reduced_motion_intact() -> None:
"""The SCROLL constant is untouched (calm, don't remove): smooth by
default, "auto" under prefers-reduced-motion — the two kept scroll
call sites ride it as the default behavior."""
js = _js()
assert 'const SCROLL = reducedMotion ? "auto" : "smooth";' in js
assert 'matchMedia("(prefers-reduced-motion: reduce)")' in js
assert "Calm, don't remove" in js
def test_turn_end_focus_does_not_scroll() -> None:
"""The turn-end focus-back (phase 06's "always focus back") must not
move the viewport: focusing the composer while the user is scrolled up
would yank them to the bottom at the moment the turn ends — the exact
defect phase 18 removes. preventScroll keeps the keyboard flow.
startNewChat keeps plain focus (the list is cleared, nothing to yank
past)."""
move the viewport: focusing the composer while the user is scrolled
up would yank them to the bottom at the moment the turn ends.
preventScroll keeps the keyboard flow without the scroll."""
js = _js()
finally_idx = js.find("// done | error → idle: always settle, always focus back")
assert finally_idx != -1, "the turn's finally block must exist"