"""Phase 48 E2E (Playwright): stop / cancel an in-flight answer.

Source: ``TODO.md`` L3 — "Need a way to stop or cancel generation of text
in the chat" (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_stop_generation.py -v --no-cov

Mock-only, no admin login (chat is public). The owner-locked contract
(2026-08-29) under test:

* while a turn is in flight the Send button is the enabled **Stop**
  control ("Stop" label, ``.is-stop`` rose treatment) — a click *or* an
  Enter in the focused input stops the turn;
* a mid-stream stop keeps the partial answer on screen, shows the
  ``.stopped-note`` "Stopped" marker, no error banner, and persists the
  partial with the optional ``stopped: true`` marker in ``bor.chat.v1``
  (a reload restores it, conversation order intact);
* a pre-token stop leaves only the question in the conversation (no
  brain bubble, no brain-side record — phase-20 convention);
* the server tears the model stream down on the client disconnect and
  writes **no** ``query_log`` row for a cancelled turn (unit-proven in
  ``tests/unit/test_chat_cancel.py``; asserted here against the live
  database after the browser-side stop settles).

Determinism: the mock streams 12 chars / 0.02 s, so the ~900-word long
answer ("write a long answer" trigger, the phase-11 on-topic phrasing)
takes ~8 s — a comfortable stop window; the pre-token window is the
mock's 3 s "pretend to think slowly" warm-up (phase-06 phrasing). The
tests wait for observable states only (the one sanctioned 1.5 s
re-read is the partial-stability check from the task).
"""
from __future__ import annotations

import asyncio
import json
import re
import time
from pathlib import Path
from threading import Thread
from typing import Any

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
from e2e.auth_helpers import login

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

#: On-topic question + the mock's long-answer trigger — the phase-11
#: phrasing, so the honesty gate is HIGH and the ~900-word answer
#: streams for ~8 s (12 chars / 0.02 s): the stop window.
LONG_QUESTION = "How is my Kubernetes cluster set up? write a long answer"
#: Phase-06 phrasing: the mock's 3 s warm-up before the first token —
#: the observable pre-token window.
SLOW_QUESTION = "pretend to think slowly then tell me about kubernetes"
SLOW_QUESTION_2 = "pretend to think slowly then tell me about backups"
#: The long answer's unique final line — absent from any partial.
LONG_ANSWER_END = "LONG-ANSWER-END"

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


# ---------------------------------------------------------------------------
# KB seeding (same pattern as the phase 02/03/06/14 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) -> ImportSummary:
    with SessionLocal() as db:
        db.execute(text("TRUNCATE chunks, documents, query_log"))
        db.commit()
    return _run_in_thread(_import_fixtures(mock_port))


def _query_log_count() -> int:
    with SessionLocal() as db:
        return db.execute(text("SELECT count(*) FROM query_log")).scalar_one()


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)


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 (the KB-offline banner is a separate,
    health-driven state the db_ready fixture keeps away)."""
    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"))


# ---------------------------------------------------------------------------
# Shared flows
# ---------------------------------------------------------------------------


def _stop_mid_stream(page: Page) -> str:
    """Ask the long on-topic question and Stop it once a few words of
    answer have streamed. Returns the rendered partial text (stable
    after the stop)."""
    page.fill("#message-input", LONG_QUESTION)
    page.click("#send-btn")

    # First deltas: the brain bubble exists and grows past a few words.
    answer = page.locator(ANSWER)
    answer.wait_for(state="visible", timeout=30_000)
    partial = ""
    deadline = time.monotonic() + 15
    while time.monotonic() < deadline:
        partial = answer.inner_text()
        if len(partial.split()) >= 8:
            break
        time.sleep(0.05)
    assert len(partial.split()) >= 8, "no answer deltas before the stop"

    # In flight: the button IS the enabled Stop control.
    btn = page.locator("#send-btn")
    expect(btn).to_be_enabled()
    expect(page.locator("#send-label")).to_have_text("Stop")
    expect(btn).to_have_class(re.compile(r"is-stop"))

    # Stop it.
    btn.click()

    # Settled to idle: Send again, stop treatment gone, no error banner.
    expect(page.locator("#send-label")).to_have_text("Send", timeout=5_000)
    expect(btn).to_be_enabled()
    expect(btn).not_to_have_class(re.compile(r"is-stop"))
    _assert_no_error_banner(page)

    # The partial is kept on screen — real answer text, and far shorter
    # than the mock's full long answer (its unique final line 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

    # Stable: no further growth ~1.5 s after the stop.
    page.wait_for_timeout(1_500)
    assert answer.inner_text() == stopped_text, (
        "the partial answer kept growing after the stop"
    )
    return stopped_text


def _stop_pre_token(page: Page, question: str, via_enter: bool) -> None:
    """Ask a slow question and stop it during the 3 s pre-token warm-up —
    by clicking the Stop control, or by pressing Enter in the focused
    input (owner-locked: click *or* Enter)."""
    page.fill("#message-input", question)
    if via_enter:
        page.keyboard.press("Enter")  # the fill focused the input; it submits
    else:
        page.click("#send-btn")

    # In the warm-up window the button reads Stop (thinking state).
    expect(page.locator("#send-label")).to_have_text("Stop", timeout=5_000)
    if via_enter:
        # The input kept focus through the submit — Enter stops the turn.
        expect(page.locator("#message-input")).to_be_focused()
        page.keyboard.press("Enter")
    else:
        page.locator("#send-btn").click()

    # Settled to idle: no error banner, no brain bubble at all (the typing
    # indicator is gone with it), the question survived, focus is back.
    expect(page.locator("#send-label")).to_have_text("Send", timeout=5_000)
    btn = page.locator("#send-btn")
    expect(btn).to_be_enabled()
    expect(btn).not_to_have_class(re.compile(r"is-stop"))
    _assert_no_error_banner(page)
    expect(page.locator(".msg.brain")).to_have_count(0)
    expect(page.locator(ANSWER)).to_have_count(0)
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)
    expect(page.locator("#message-input")).to_be_focused()


# ---------------------------------------------------------------------------
# 1. Mid-stream stop: partial kept + "Stopped" note + stopped persistence
# ---------------------------------------------------------------------------


def test_stop_mid_stream(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    summary = _reset_db(mock_llm)
    assert summary is not None and summary.added == 13  # A9 formats (phase 47 added quadlet+j2)
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")

    stopped_text = _stop_mid_stream(page)

    # The bubble keeps the partial and carries the "Stopped" note.
    expect(page.locator(ANSWER)).to_contain_text(stopped_text)
    note = page.locator(".msg.brain .stopped-note")
    expect(note).to_have_count(1)
    expect(note.first).to_contain_text("Stopped")
    # No sources on a stopped turn — it never settled.
    expect(page.locator(".msg.brain .source-chip")).to_have_count(0)
    # The live region confirms the stop (no error wording).
    expect(page.locator("#send-status")).to_contain_text("Answer stopped.")

    # Persistence: question, then the partial with the optional
    # ``stopped`` marker (raw text — no HTML, no LONG-ANSWER-END).
    stored = _stored_parsed(page)
    assert [m["who"] for m in stored["messages"]] == ["user", "brain"]
    assert stored["messages"][0]["text"] == LONG_QUESTION
    last = stored["messages"][-1]
    assert last["stopped"] is True
    assert last["text"].strip()
    assert "Step 1:" in last["text"]
    assert LONG_ANSWER_END not in last["text"]

    # Server side: the cancelled turn wrote no durable record (the
    # stability wait above already gave the teardown its margin).
    assert _query_log_count() == 0


# ---------------------------------------------------------------------------
# 2. Pre-token stop: question kept, nothing brain-side (click and Enter)
# ---------------------------------------------------------------------------


def test_stop_pre_token(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _reset_db(mock_llm)
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")

    _stop_pre_token(page, SLOW_QUESTION, via_enter=False)

    # The conversation holds exactly the question — no brain-side record.
    stored = _stored_parsed(page)
    assert [m["who"] for m in stored["messages"]] == ["user"]
    assert stored["messages"][0]["text"] == SLOW_QUESTION
    assert "stopped" not in stored["messages"][0]

    # The owner-locked "click *or* Enter": the same pre-token stop, this
    # time submitted and stopped through the focused input's Enter.
    _stop_pre_token(page, SLOW_QUESTION_2, via_enter=True)
    expect(page.locator(".msg.user .bubble")).to_have_count(2)
    expect(page.locator(".msg.brain")).to_have_count(0)
    stored = _stored_parsed(page)
    assert [m["who"] for m in stored["messages"]] == ["user", "user"]
    assert stored["messages"][-1]["text"] == SLOW_QUESTION_2

    # Server side: neither pre-token stop left a durable record. The UI
    # settle was observed above; the short margin covers the server's
    # disconnect teardown (mock cadence 0.02 s).
    page.wait_for_timeout(500)
    assert _query_log_count() == 0


# ---------------------------------------------------------------------------
# 3. A stopped turn survives a reload
# ---------------------------------------------------------------------------


def test_stopped_turn_survives_reload(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    _reset_db(mock_llm)
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")

    stopped_text = _stop_mid_stream(page)
    assert _query_log_count() == 0

    page.reload()

    # Restored: the user message first, then the stopped brain bubble —
    # the same partial text, the "Stopped" note, and the idle Send button.
    expect(page.locator("#empty-state")).to_be_hidden()
    msgs = page.locator("#messages > .msg")
    expect(msgs).to_have_count(2)
    expect(msgs.nth(0)).to_have_class(re.compile(r"msg user"))
    expect(msgs.nth(1)).to_have_class(re.compile(r"msg brain"))
    expect(page.locator(".msg.user .bubble")).to_contain_text(LONG_QUESTION)
    bubble = page.locator(ANSWER)
    expect(bubble).to_have_count(1)
    assert bubble.inner_text() == stopped_text, "the restored partial differs"
    note = page.locator(".msg.brain .stopped-note")
    expect(note).to_have_count(1)
    expect(note.first).to_contain_text("Stopped")

    expect(page.locator("#send-label")).to_have_text("Send")
    expect(page.locator("#send-btn")).to_be_enabled()
    expect(page.locator("#send-btn")).not_to_have_class(re.compile(r"is-stop"))

    # The marker still rides the stored record after the restore.
    stored = _stored_parsed(page)
    assert [m["who"] for m in stored["messages"]] == ["user", "brain"]
    assert stored["messages"][-1]["stopped"] is True