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

"""Phase 15 E2E (Playwright): tune how Brain answers (steering notes).

Phase 16 adaptation: tuning is admin-only — every test performs the real
form login (``e2e.auth_helpers.login``) before touching the tuning UI.

Story: ``.agents/user_stories/steering-notes.md``
Run in isolation (DB must be up: ``podman compose up -d db``):

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

The steering loop: "Tune" under a completed answer → short instruction →
stored in Postgres (``steering_notes``) → injected into the system prompt
of every subsequent turn as the ``<tuning>`` section. The mock LLM
echoes the first tuning note into its answer
(`` (tuning: <first note line>)``), so prompt injection is observable in
the UI deterministically. Notes are listed newest-first on the Tuning
page (``/tuning.html``), where each can be deleted — the header "Tuning"
toggle was removed from the navbar at owner request (2026-08-28).

Test → story mapping (Playwright Mapping Rule):
1. ``test_tune_under_answer_persists_and_steers``
2. ``test_delete_note_stops_steering``
3. ``test_note_rendered_as_text_xss_safe``
4. ``test_tuning_panel_a11y``
"""
from __future__ import annotations

import asyncio
from pathlib import Path
from threading import Thread
from typing import Any

from playwright.sync_api import Page, expect
from sqlalchemy import select, text

from app.config import Settings
from app.db import SessionLocal
from app.models import SteeringNote
from app.rag.importer import ImportSummary, import_sources
from app.rag.llm import LLMClient
from e2e.auth_helpers import login

REPO = Path(__file__).resolve().parents[2]
FIXTURES = REPO / "tests" / "fixtures" / "docs"
QUESTION = "How is my Kubernetes cluster set up?"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"
NOTE = "STEEER-MARKER be concise"
XSS_NOTE = "<script>window.__xss = true; alert('xss')</script>"
#: The shell (index.html — served for BOTH / and /tuning.html since
#: phase 76 task 01, when the Tuning view folded into it) ships exactly
#: FIVE classic/module script tags: the phase-39 brand.js classic layer
#: + markdown.js + the chat module (app.js) + the shell router
#: (router.js, which lazy-imports the tuning.js view module) + the
#: phase-79 token gate (token-gate.js, phase 79 task 05).
BASE_SCRIPT_COUNT = 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 + steering notes), re-import fixtures."""
    with SessionLocal() as db:
        db.execute(text("TRUNCATE chunks, documents, query_log, steering_notes"))
        db.commit()
    if not seed:
        return None
    return _run_in_thread(_import_fixtures(mock_port))


def _ask(page: Page, question: str) -> None:
    """Send one turn and wait until the grounded answer has fully landed."""
    page.fill("#message-input", question)
    page.click("#send-btn")
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)
    expect(page.locator(".msg.brain .bubble").last).to_contain_text(
        MOCK_ANSWER_MARKER, timeout=30_000
    )
    expect(page.locator("#send-btn")).to_be_enabled()
    expect(page.locator("#send-label")).to_have_text("Send")


def _tune_and_save(page: Page, note: str) -> None:
    """Tune the last completed brain bubble and save *note*."""
    tune = page.locator(".msg.brain .tune-btn").last
    expect(tune).to_be_visible()
    tune.click()
    form = page.locator(".msg.brain .tune-form").last
    expect(form).to_be_visible()
    form.locator("textarea").fill(note)
    form.locator(".tune-save").click()
    saved = page.locator(".msg.brain .tune-saved").last
    expect(saved).to_contain_text("Saved — future answers will follow this.", timeout=15_000)


# ---------------------------------------------------------------------------
# 1. Tune under an answer → persisted → next answer carries the note
# ---------------------------------------------------------------------------


def test_tune_under_answer_persists_and_steers(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    summary = _reset_db(mock_llm, seed=True)
    assert summary is not None and summary.added == 13  # A9 formats (phase 47 added quadlet+j2)
    page.set_default_timeout(30_000)
    page.goto(app_url)
    login(page, app_url, next="/")  # phase 16: tuning is admin-only
    _ask(page, QUESTION)

    # The Tune control: ghost button in the answer's meta row, ≥44px.
    tune = page.locator(".msg.brain .tune-btn").last
    expect(tune).to_have_count(1)
    expect(tune).to_have_attribute("type", "button")
    box = tune.bounding_box()
    assert box is not None and box["height"] >= 44

    _tune_and_save(page, NOTE)

    # Persisted in Postgres.
    with SessionLocal() as db:
        rows = db.scalars(select(SteeringNote)).all()
        assert [r.note for r in rows] == [NOTE]

    # The Tuning page (the steering-notes manager — the navbar toggle
    # was removed at owner request, 2026-08-28) lists the note.
    page.goto(app_url + "/tuning.html")
    expect(page.locator("#tune-list .tuning-note")).to_have_count(1)
    expect(page.locator("#tune-list .tuning-note-text")).to_have_text(NOTE)

    # The NEXT answer carries the note — it reached the system prompt.
    page.goto(app_url + "/")
    expect(page.locator("#send-btn")).to_be_enabled()
    _ask(page, QUESTION)
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(f"(tuning: {NOTE})")


# ---------------------------------------------------------------------------
# 2. Delete from the panel → count 0 → steering stops
# ---------------------------------------------------------------------------


def test_delete_note_stops_steering(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _reset_db(mock_llm, seed=True)
    page.set_default_timeout(30_000)
    page.goto(app_url)
    login(page, app_url, next="/")  # phase 16: tuning is admin-only
    _ask(page, QUESTION)
    _tune_and_save(page, NOTE)

    # Steering is live: one more answer carries the marker.
    _ask(page, QUESTION)
    expect(page.locator(".msg.brain .bubble").last).to_contain_text(f"(tuning: {NOTE})")

    # Delete the note from the Tuning page (the header panel is no
    # longer reachable — the navbar toggle is gone).
    page.goto(app_url + "/tuning.html")
    expect(page.locator("#tune-list .tuning-note")).to_have_count(1)
    page.locator(".tuning-delete").click()
    expect(page.locator("#tune-list .tuning-note")).to_have_count(0, timeout=15_000)
    expect(page.locator("#tune-empty")).to_be_visible()
    expect(page.locator("#tune-announcer")).to_contain_text("deleted")

    with SessionLocal() as db:
        assert db.scalars(select(SteeringNote)).all() == []

    # The next answer no longer carries the marker.
    page.goto(app_url + "/")
    expect(page.locator("#send-btn")).to_be_enabled()
    _ask(page, QUESTION)
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(MOCK_ANSWER_MARKER)
    expect(bubble).not_to_contain_text("STEEER-MARKER")


# ---------------------------------------------------------------------------
# 3. Notes render as text (XSS-safe)
# ---------------------------------------------------------------------------


def test_note_rendered_as_text_xss_safe(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _reset_db(mock_llm, seed=True)
    page.set_default_timeout(30_000)
    page.goto(app_url)
    login(page, app_url, next="/")  # phase 16: tuning is admin-only

    dialogs: list[str] = []

    def _handle_dialog(d) -> None:
        dialogs.append(d.message)
        d.dismiss()

    page.on("dialog", _handle_dialog)

    _ask(page, QUESTION)
    _tune_and_save(page, XSS_NOTE)

    # Tuning page: the payload is visible as LITERAL text…
    page.goto(app_url + "/tuning.html")
    expect(page.locator("#tune-list .tuning-note-text")).to_have_text(XSS_NOTE)

    # …never as an executed element: no script tag in the list, the page
    # still carries exactly its own three scripts, no dialog.
    assert page.locator("#tune-list script").count() == 0
    expect(page.locator("script")).to_have_count(BASE_SCRIPT_COUNT)
    assert dialogs == [], f"the note must never execute as script: {dialogs}"
    assert page.evaluate("() => window.__xss === undefined") is True


# ---------------------------------------------------------------------------
# 4. Tuning panel accessibility
# ---------------------------------------------------------------------------


def test_tuning_panel_a11y(page: Page, app_url: str, db_ready: None) -> None:
    """The steering surface's a11y now lives on the Tuning page (the
    header toggle was removed from the navbar at owner request,
    2026-08-28): the note list, the polite live region, and the labeled
    per-note delete (≥44px)."""
    _reset_db(mock_port=0, seed=False)  # no KB seeding needed for the page a11y
    page.set_default_timeout(30_000)
    page.goto(app_url)
    login(page, app_url, next="/tuning.html")  # phase 16: the notes are admin-only

    # The navbar no longer carries a steering toggle — absent for the
    # admin too — and the header panel section still ships hidden.
    assert page.locator("#steering-toggle").count() == 0
    expect(page.locator("#steering-panel")).to_be_hidden()

    # The page's own polite live region.
    announcer = page.locator("#tune-announcer")
    assert announcer.get_attribute("role") == "status"
    assert announcer.get_attribute("aria-live") == "polite"

    # Add a note (API), reload to refresh the list (tuning.js fetches
    # on boot).
    page.evaluate(
        """async () => {
            const r = await fetch('/api/steering', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({note: 'a11y note one'}),
            });
            if (!r.ok) throw new Error('steering POST failed: ' + r.status);
        }"""
    )
    page.reload()
    expect(page.locator("#sign-out-btn")).to_be_visible(timeout=15_000)
    note_item = page.locator("#tune-list .tuning-note")
    expect(note_item).to_have_count(1)
    expect(note_item.locator(".tuning-note-text")).to_have_text("a11y note one")

    # The per-note delete is a real, labeled button (≥44px target).
    delete = page.locator(".tuning-delete")
    expect(delete).to_have_attribute("type", "button")
    assert (delete.get_attribute("aria-label") or "").startswith("Delete tuning note:")
    box = delete.bounding_box()
    assert box is not None and box["height"] >= 44

    # Delete: list empties, the empty state shows, the live region
    # announces it.
    delete.click()
    expect(page.locator("#tune-list .tuning-note")).to_have_count(0, timeout=15_000)
    expect(page.locator("#tune-empty")).to_be_visible()
    expect(announcer).to_contain_text("deleted")