"""Phase 73 E2E (Playwright): a hidden tab never stops a generating answer.

Source: ``TODO.md`` L3 — "Clicking on another tab while an answer is
generating stops that answer from being generated. Reponses should
continue to generate unless you outright close the tab."
Run in isolation (DB must be up: ``podman compose up -d db``):

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

Real tab-switching is browser-environment-specific, so the suite
dispatches the synthetic hidden-``visibilityState`` + ``pagehide`` /
``pageshow`` sequence from the task file — the exact events the
browsers that fire ``pagehide`` on a merely-hidden tab deliver (e.g.
Safari; task 01: Chromium 151 real mode fires only
``visibilitychange``, so the C1 branch is latent there). Against the
deterministic mock LLM (the ~8 s long stream gives a guaranteed
mid-stream window) the pins are:

1. ``test_hidden_tab_does_not_stop_the_answer`` — a synthetic
   ``pagehide`` mid-turn never stops the answer: the bubble completes
   with the FULL mock answer (``LONG-ANSWER-END`` included), no error
   banner, the ``bor.chat.v1`` record holds EXACTLY ONE brain turn —
   the pagehide partial is REPLACED in place by the ``done`` settle
   (task 01's C1 double-record corruption stays pinned dead), the
   auto-saved ``saved_chats`` row (admin context, the
   ``persistConversation`` path) carries the same single brain turn,
   and the server settled the turn (one ``query_log`` row — no
   phase-48 ``turn cancelled`` teardown for a merely-hidden tab).
2. ``test_reload_after_hidden_tab_restores_one_bubble`` — the same
   mid-stream pagehide, then a reload: the restore renders exactly
   one brain bubble for the question (a C1 double-record would render
   two).
3. ``test_baseline_no_pagehide_still_completes`` — the same long
   question with NO dispatched events completes identically (guards
   against an over-eager fix changing the normal path).
4. ``test_pre_token_guard_survives_hidden_window`` — task 01's C2
   (confirmed): hidden time must not count toward the 120 s pre-token
   guard. Playwright's fake clock makes the guard's deadline
   deterministic: the clock is fast-forwarded past the guard's
   ORIGINAL 120 s deadline while the tab is (synthetically) hidden,
   with a return to visible in between — the phase-73 re-arm gives the
   still-armed guard a FRESH ``TURN_TIMEOUT_MS``, so the turn still
   settles with the answer. Without the re-arm the fast-forwarded
   original timer fires with the "stuck" error — the test
   discriminates. The task's suggested trigger
   (``think out loud then hesitate``) cannot exercise the re-arm: it
   streams its scratchpad FIRST, and the first thinking frame clears
   the guard — its 4 s pause sits in a post-clear window.
   ``pretend to think slowly`` (the phase-06 phrasing,
   ``test_stop_generation.py``'s SLOW_QUESTION) is the mock's only
   PURE pre-token silence (3 s, zero frames) — the guard stays armed
   the whole window.

The phase-48 teardown contract is covered by the regression runs
(``test_stop_generation.py``, ``test_chat_persistence.py``) — real
Stop / close / navigation still cancel the fetch.
"""
from __future__ import annotations

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

import httpx
from playwright.sync_api import Locator, 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
from e2e.mock_llm import LONG_ANSWER_LINES, long_answer

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

#: The phase-11 on-topic long-answer phrasing (house pattern,
#: ``test_stop_generation.py``): the honesty gate is HIGH and the
#: ~900-word answer streams for ~8 s (12 chars / 0.02 s) — the
#: guaranteed mid-stream window.
LONG_QUESTION = "How is my Kubernetes cluster set up? write a long answer"
#: The mock's byte-stable long answer — the EXACT string the stream
#: delivers, so "the full answer" is an exact comparison, not a
#: contains check.
LONG_ANSWER = long_answer()
LONG_ANSWER_END = "LONG-ANSWER-END"

#: The mock's only pure pre-token silence (phase-06 phrasing): a 3 s
#: delay before the FIRST frame, so the 120 s pre-token guard stays
#: armed the whole window (no thinking/delta/retry frame to clear it).
SLOW_QUESTION = "pretend to think slowly then tell me about kubernetes"
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"

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

#: The synthetic tab-switch sequence (task file): the exact events a
#: browser that fires ``pagehide`` on a merely-hidden tab delivers —
#: the visibility transition (the app's only visibility listener is
#: the phase-73 guard re-arm) + the PageTransitionEvents, with
#: ``persisted: false`` (a tab switch is not a bfcache navigation).
HIDE_JS = """
() => {
  Object.defineProperty(document, "visibilityState", { configurable: true, get: () => "hidden" });
  document.dispatchEvent(new Event("visibilitychange"));
  window.dispatchEvent(new PageTransitionEvent("pagehide", { persisted: false }));
}
"""
SHOW_JS = """
() => {
  Object.defineProperty(document, "visibilityState", { configurable: true, get: () => "visible" });
  document.dispatchEvent(new Event("visibilitychange"));
  window.dispatchEvent(new PageTransitionEvent("pageshow", { persisted: false }));
}
"""

# ---------------------------------------------------------------------------
# KB seeding (house pattern: TRUNCATE-then-import)
# ---------------------------------------------------------------------------


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:
    """House reset + the prompt-shaping tables: steering notes and the
    KB overview would otherwise append deterministic suffixes to every
    mock answer and break the exact-text assertions. ``saved_chats``
    is NOT touched (rows persist across suites; test 1 cleans up its
    own row)."""
    with SessionLocal() as db:
        db.execute(
            text("TRUNCATE chunks, documents, query_log, steering_notes, kb_overview")
        )
        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()


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


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:
    """Neither error copy may be up: the "stuck" guard copy (C2) or the
    stream-drop copy — a merely-hidden tab settles to idle like a
    normal turn."""
    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 _ask_long(page: Page) -> None:
    page.fill("#message-input", LONG_QUESTION)
    page.click("#send-btn")
    expect(page.locator(".msg.user .bubble").last).to_contain_text(LONG_QUESTION)


def _wait_streaming(page: Page) -> Locator:
    """Wait until answer text is visibly streaming (a few delta frames
    rendered — the task's mid-stream moment, well inside the ~8 s
    stream)."""
    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 tab switch"
    # In flight at the switch: the button IS the enabled Stop control.
    expect(page.locator("#send-label")).to_have_text("Stop")
    expect(page.locator("#send-btn")).to_have_class(re.compile(r"is-stop"))
    return answer


def _wait_done(page: Page, answer: Locator) -> str:
    """Wait for the ``done`` settle: the Send button is back and the
    bubble carries the unique final line — no error banner on the way."""
    expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)
    expect(page.locator("#send-btn")).not_to_have_class(re.compile(r"is-stop"))
    expect(answer).to_contain_text(LONG_ANSWER_END, timeout=30_000)
    _assert_no_error_banner(page)
    return answer.inner_text()


def _assert_full_answer(text: str) -> None:
    """The bubble carries the FULL mock answer — every one of the 40
    numbered steps plus the unique final line (a truncated stream
    would be missing its tail)."""
    for i in range(1, LONG_ANSWER_LINES + 1):
        assert f"Step {i}: configure node-{i}" in text, f"step {i} missing from the answer"
    assert LONG_ANSWER_END in text


def _assert_one_brain_turn(page: Page, full: bool) -> dict[str, Any]:
    """The C1 corruption pin: EXACTLY ONE brain turn for the question.

    With ``full`` the record's text must be the complete mock answer
    byte-for-byte (the ``done`` settle's record); otherwise it is the
    pagehide partial (a true prefix of the full answer — its final
    line never arrived).
    """
    stored = _stored_parsed(page)
    assert stored["v"] == 1
    whos = [m["who"] for m in stored["messages"]]
    assert whos == ["user", "brain"], (
        "exactly ONE brain turn for the question "
        f"(the pagehide partial must be replaced, never duplicated): {whos}"
    )
    assert stored["messages"][0]["text"] == LONG_QUESTION
    brain = stored["messages"][1]
    if full:
        assert brain["text"] == LONG_ANSWER, "the record's text is the FULL answer"
        assert brain.get("deflected") is False, "the done metadata rides the record"
    else:
        # The pagehide partial carries no done metadata (the turn never
        # settled when it was written) — only who + text.
        assert "deflected" not in brain
    return brain


def _admin_cookies(page: Page) -> dict[str, str]:
    """The signed session cookies the browser holds after a form login —
    used to call the admin API with plain httpx (the test's API side
    sees exactly what the signed-in browser sees)."""
    return {c["name"]: c["value"] for c in page.context.cookies() if "name" in c and "value" in c}


def _delete_rows_by_title(app_url: str, cookies: dict[str, str], title: str) -> None:
    """Best-effort cleanup of the auto-saved row (a 404 is fine)."""
    r = httpx.get(f"{app_url}/api/chats", timeout=10, cookies=cookies)
    if r.status_code != 200:
        return
    for c in r.json()["chats"]:
        if c["title"] == title:
            httpx.delete(f"{app_url}/api/chats/{c['id']}", timeout=10, cookies=cookies)


def _wait_row_full(app_url: str, cookies: dict[str, str], title: str) -> dict[str, Any]:
    """Poll the auto-saved row until it carries the full answer as a
    single brain turn (the ``done`` settle's fire-and-forget
    ``persistConversation`` PUT is the last writer)."""
    deadline = time.monotonic() + 15
    last: list[dict[str, Any]] = []
    while time.monotonic() < deadline:
        r = httpx.get(f"{app_url}/api/chats", timeout=10, cookies=cookies)
        rows = (
            [c for c in r.json()["chats"] if c["title"] == title]
            if r.status_code == 200
            else []
        )
        for c in rows:
            row = httpx.get(f"{app_url}/api/chats/{c['id']}", timeout=10, cookies=cookies).json()
            brains = [m for m in row["messages"] if m["who"] == "brain"]
            if len(brains) == 1 and brains[0]["text"] == LONG_ANSWER:
                return row
            last = [row]
        time.sleep(0.2)
    raise AssertionError(
        "the auto-saved row never held the full answer as exactly one brain turn; last: "
        f"{last!r}"
    )


# ---------------------------------------------------------------------------
# 1. Hidden tab mid-stream: the answer completes exactly once, the
#    record holds one brain turn
# ---------------------------------------------------------------------------


def test_hidden_tab_does_not_stop_the_answer(
    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)
    # Admin login: the auto-save row (the persistConversation path) is
    # reachable, so the saved-chat side gets pinned too.
    login(page, app_url, next="/")
    cookies = _admin_cookies(page)
    _delete_rows_by_title(app_url, cookies, LONG_QUESTION)  # stale rows from crashed runs

    _ask_long(page)
    answer = _wait_streaming(page)

    # Mid-stream: the exact tab-switch event sequence (task file).
    page.evaluate(HIDE_JS)

    # The pagehide partial landed: EXACTLY ONE brain turn — the partial
    # (a true prefix of the full answer; its final line never arrived).
    # This is the record the C1 bug would have LEFT ALONGSIDE the full
    # answer (two brain turns for one question).
    partial = _assert_one_brain_turn(page, full=False)
    assert partial["text"], "the partial must carry the streamed text"
    assert partial["text"] in LONG_ANSWER, "the partial is a prefix of the full answer"
    assert LONG_ANSWER_END not in partial["text"], "the final line had not arrived yet"

    # Return to visible — the stream keeps filling the live bubble and
    # the turn settles.
    page.evaluate(SHOW_JS)
    done_text = _wait_done(page, answer)
    _assert_full_answer(done_text)

    # The ``done`` settle REPLACED the partial in place: one brain
    # turn, and its text is the full answer.
    _assert_one_brain_turn(page, full=True)

    # Server side: the turn SETTLED — one query_log row, so no
    # phase-48 "turn cancelled" teardown fired for a merely-hidden tab.
    assert _query_log_count() == 1

    # Auto-save (admin): the row carries the same single full brain
    # turn (the shared record shape).
    try:
        row = _wait_row_full(app_url, cookies, LONG_QUESTION)
        brains = [m for m in row["messages"] if m["who"] == "brain"]
        assert len(brains) == 1, "the saved row holds exactly one brain turn"
        assert brains[0]["text"] == LONG_ANSWER
    finally:
        _delete_rows_by_title(app_url, cookies, LONG_QUESTION)


# ---------------------------------------------------------------------------
# 2. Reload after the hidden tab: exactly one brain bubble restores
# ---------------------------------------------------------------------------


def test_reload_after_hidden_tab_restores_one_bubble(
    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="/")  # phase 79: chat is require_user-gated

    _ask_long(page)
    answer = _wait_streaming(page)
    page.evaluate(HIDE_JS)
    page.evaluate(SHOW_JS)
    done_text = _wait_done(page, answer)
    _assert_full_answer(done_text)
    _assert_one_brain_turn(page, full=True)

    # The restore path re-renders the bor.chat.v1 record: a C1
    # double-record (partial + full) would render TWO brain bubbles.
    page.reload()
    expect(page.locator("#empty-state")).to_be_hidden()
    expect(page.locator(".msg.user .bubble")).to_have_count(1)
    expect(page.locator(".msg.user .bubble").first).to_contain_text(LONG_QUESTION)
    expect(page.locator(ANSWER)).to_have_count(1)
    restored = page.locator(ANSWER).first
    _assert_full_answer(restored.inner_text())
    _assert_no_error_banner(page)
    expect(page.locator("#send-label")).to_have_text("Send")

    # The record the restore read: still exactly one brain turn, full
    # text (the restore is read-only).
    _assert_one_brain_turn(page, full=True)


# ---------------------------------------------------------------------------
# 3. Baseline: the same long question, NO dispatched events
# ---------------------------------------------------------------------------


def test_baseline_no_pagehide_still_completes(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    """The normal path, untouched by the fix: the long answer completes
    identically without any synthetic tab-switch events (guards against
    an over-eager fix changing the ordinary settle)."""
    _reset_db(mock_llm)
    page.set_default_timeout(30_000)
    login(page, app_url, next="/")  # phase 79: chat is require_user-gated

    _ask_long(page)
    answer = _wait_streaming(page)
    done_text = _wait_done(page, answer)
    _assert_full_answer(done_text)
    _assert_one_brain_turn(page, full=True)
    assert _query_log_count() == 1


# ---------------------------------------------------------------------------
# 4. C2 (task 01, confirmed): the pre-token guard survives a hidden
#    window — hidden time does not count toward TURN_TIMEOUT_MS
# ---------------------------------------------------------------------------


def test_pre_token_guard_survives_hidden_window(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    """The 120 s pre-token guard is a plain ``setTimeout`` — without the
    phase-73 visibility re-arm, hidden time counts toward it and a turn
    whose first frame lands past the 120 s mark errors with the "stuck"
    copy (task 01 scenario B, C2 confirmed). The fake clock makes the
    deadline deterministic (waiting 120 s of real time is not an
    option for a regression suite): the mock's 3 s real-time pre-token
    silence keeps the guard armed while the FAKE clock is fast-forwarded
    across the original deadline — with a return to visible in between,
    which re-arms the still-armed guard with a fresh TURN_TIMEOUT_MS.

    Timeline (fake clock, the guard armed at t=0, deadline t=120 s):

    * t=0    — the slow question is sent; the guard arms (t=120 s);
    * t~0    — the tab goes hidden (synthetic; pre-token, so the
               pagehide handler persists nothing — the question is
               already saved);
    * t=115  — fast-forward while hidden: the original timer (120 s)
               is not due yet;
    * t=115  — back to visible: the re-arm fires (the guard is still
               armed — no frame has arrived) → new deadline t=235 s;
    * t=229  — fast-forward again: PAST the original 120 s deadline —
               without the re-arm the guard fires HERE with the "stuck"
               error — but short of the re-armed 235 s deadline.

    Meanwhile (real time, ~3.5 s after the send) the mock's first frame
    lands, clears the guard for good, and the turn settles with the
    full answer.
    """
    _reset_db(mock_llm)
    page.set_default_timeout(30_000)

    # The fake clock BEFORE the first navigation: every app timer (the
    # 120 s guard among them) becomes test-controlled, while the mock's
    # real-time stream is unaffected (the network is not a timer).
    page.clock.install()
    login(page, app_url, next="/")  # phase 79: chat is require_user-gated

    page.fill("#message-input", SLOW_QUESTION)
    page.click("#send-btn")
    expect(page.locator("#send-label")).to_have_text("Stop", timeout=30_000)
    # The pre-token window: no answer frame yet (the mock's 3 s real
    # silence is running) — the armed guard is the only thing that can
    # now stop the turn.
    expect(page.locator(ANSWER)).to_have_count(0)

    page.evaluate(HIDE_JS)
    page.clock.fast_forward(115_000)
    page.evaluate(SHOW_JS)  # the re-arm: a fresh TURN_TIMEOUT_MS
    page.clock.fast_forward(114_000)  # past 120 s, short of the re-armed 235 s

    # The turn still settles with the full answer — no "taking a long
    # time" error, no aborted stream.
    answer = page.locator(ANSWER)
    expect(answer).to_contain_text(MOCK_ANSWER_MARKER, timeout=30_000)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)
    _assert_no_error_banner(page)

    # One brain turn for the question — the settled answer (pre-token
    # hidden time persisted nothing, as the phase-20 convention says).
    stored = _stored_parsed(page)
    assert [m["who"] for m in stored["messages"]] == ["user", "brain"]
    assert stored["messages"][0]["text"] == SLOW_QUESTION
    assert MOCK_ANSWER_MARKER in stored["messages"][1]["text"]