"""Phase 68 E2E (Playwright, mock-only): the ``grep`` tool (the
phase-68 search tool, renamed to the harness-aligned ``grep`` in
phase 70; the A5 match/output contract is unchanged).

Story: n/a (TODO-derived — the owner roadmap confirmation 2026-09-01,
TODO.md L4: "Add a search tool that allows the LLM to grep through the
uploaded documents for a given string").
Run in isolation (DB must be up: ``podman compose up -d db``):

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

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the gate is the
deterministic SEARCH marker flow in ``tests/e2e/mock_llm.py`` (user
message contains ``search your documents`` (``SEARCH_TRIGGER``)
**and** the system prompt carries the ``<tools>`` section of the HIGH
prompt):

1. request 1 (``tools`` offered, no search result yet) → streams ONLY
   ``tool_calls`` deltas calling ``grep`` with ``{"pattern":
   SEARCH_PATTERN}`` (id ``call_0``);
2. request 2 (a ``tool``-role search result — the
   ``source/path:line: text`` match line) → the content answer
   ``Found <first matched line's content up to 80 chars>`` — so this
   suite can assert the search result reached the model and landed in
   the answer.

KB fixture — ONE document, imported through the real importer (the same
admin-import pipeline the admin Sources page drives) against
``tests/fixtures/search_docs/`` with the deterministic mock embeddings:

* ``search_docs/reese-notes.md`` — homelab kubernetes backup notes; the
  marker question cosines ≈0.51 against it (mock bag-of-words, well
  past the E2E 0.30 threshold, and it FTS-matches too) → grounded, so
  the HIGH prompt carries the ``<tools>`` section. Its line 6 carries
  the sentinel ``reese-sentinel-42`` exactly once — the deterministic
  match of the mock's grep.

Regression-safe marker: ``search your documents`` appears in NO other
suite's question or fixture text (verified 2026-09-01 by repo grep;
``tests/unit/test_mock_tool_flow.py`` pins that the trigger does not
shadow the phase-37/45 flows and vice versa).

Test → phase mapping:
1. ``test_search_flow_searches_and_answers_from_match`` — the live
   search flow: the SSE carries the ``tool`` frame (``grep`` with
   ``argument = <sentinel>``, ahead of any delta), #send-status
   recorded the transient "… is searching for <sentinel>" state, the
   bubble shows ONE ``🔎 Searching for`` tool line with the sentinel in
   a ``<code>`` element, the answer quotes the matched line
   (``Found …`` — the match reached the model), and the turn settles to
   idle with no error banner.
2. ``test_search_adds_no_source_by_itself`` — context accounting
   (locked A5, phase-119 A1 shape): the search-only flow (no read)
   leaves ``done.sources`` / the source chips EMPTY (chips cite read
   docs only — the search adds no source by itself, and the retrieval
   doc was never read) while ``query_log.sources`` keeps the retrieval
   baseline (LOCKED A3, untouched).
3. ``test_search_tool_line_re_renders_after_reload`` — the persisted
   record (phase 14 convention: the generic ``{name, argument}``
   toolAcc) re-renders the search line through the same helper.
"""
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 select, text

from app.config import Settings
from app.db import SessionLocal
from app.models import QueryLog
from app.rag.importer import ImportSummary, import_sources
from app.rag.llm import LLMClient
from e2e.auth_helpers import login
from tests.e2e.mock_llm import SEARCH_PATTERN, SEARCH_TRIGGER

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

#: The one imported document (the importer derives ``source`` from the
#: fixture root's name and ``path`` from the file's relative position).
SEED_SOURCE = "search_docs"
SEED_PATH = "reese-notes.md"
SEED_SP = f"{SEED_SOURCE}/{SEED_PATH}"

#: The fixture's sentinel line (line 6) — the mock's grep matches it
#: exactly once, and its ``text`` part is what the "Found …" answer
#: quotes. Pinned against the fixture file itself below.
SENTINEL_LINE = f"The offsite vault passphrase marker is {SEARCH_PATTERN}."
FOUND_ANSWER = f"Found {SENTINEL_LINE[:80]}"

#: Carries ``SEARCH_TRIGGER`` and is on-topic (cosine ≈0.51 against the
#: fixture + FTS hits → HIGH gate, the ``<tools>`` section rides along).
SEARCH_QUESTION = (
    "Search your documents for the vault passphrase marker in my homelab "
    "kubernetes backup notes?"
)
assert SEARCH_TRIGGER in SEARCH_QUESTION.lower()
# The trigger must not collide with any other mock marker flow.
for other in (
    "use your tools",
    "read two documents",
    "write a long answer",
    "think in paragraphs",
    "think out loud",
    "show the end of your notes",
    "show me a table",
    "fail then answer",
    "always fail",
    "embed fail once",
    "pretend to think slowly",
):
    assert other not in SEARCH_QUESTION.lower(), other


def _pin_fixture() -> None:
    """The fixture carries the sentinel on line 6, exactly once."""
    content = (FIXTURES / SEED_PATH).read_text(encoding="utf-8")
    lines = content.split("\n")
    assert lines[5] == SENTINEL_LINE, lines[5]
    assert sum(SEARCH_PATTERN in line for line in lines) == 1


# --------------------------------------------------------------------------
# DB seeding (TRUNCATE-then-import via the real importer, cf.
# test_chat_rag.py — the same pipeline the admin Sources import drives)
# --------------------------------------------------------------------------


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 (the established house helper).
    """
    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:
    """Truncate the KB (plus the prompt-shaping tables), then re-import.

    ``steering_notes`` / ``kb_overview`` are truncated too, so the HIGH
    prompt is exactly ``<relevance>`` + ``<documents>`` + ``<tools>``
    regardless of leftovers from other suites — byte-stable prompts,
    byte-stable answers."""
    with SessionLocal() as db:
        db.execute(
            text("TRUNCATE chunks, documents, query_log, steering_notes, kb_overview")
        )
        db.commit()
    summary = _run_in_thread(_import_fixtures(mock_port))
    assert summary is not None and summary.added == 1, summary
    return summary


def _last_query_log() -> QueryLog:
    with SessionLocal() as db:
        rows = db.scalars(select(QueryLog)).all()
    assert len(rows) == 1, f"expected exactly one query_log row, got {len(rows)}"
    return rows[0]


# --------------------------------------------------------------------------
# Page helpers (the test_agent_document_tools.py pattern)
# --------------------------------------------------------------------------

#: Records every value #send-label takes during the turn (a
#: MutationObserver on the element), so the in-flight label state is
#: captured deterministically — no polling race. Phase 48: the label is
#: the Send↔Stop morph ("Stop" holds for the whole in-flight turn).
LABEL_RECORDER = """
() => {
  if (window.__labelsInstalled) return;
  window.__labelsInstalled = true;
  window.__labels = [];
  const el = document.querySelector('#send-label');
  if (!el) return;
  const rec = (v) => {
    const l = window.__labels;
    if (!l.length || l[l.length - 1] !== v) l.push(v);
  };
  rec(el.textContent);
  new MutationObserver(() => rec(el.textContent)).observe(el, {
    childList: true,
    subtree: true,
  });
}
"""

#: Records every value #send-status takes during the turn — the
#: transient "… is searching for <pattern>" calling-tool state is held
#: only from the first `tool` frame until the first answer delta, so the
#: pre-submit observer is the deterministic source of truth for it
#: (no polling race — the phase-44 task-03 flake fix).
STATUS_RECORDER = """
() => {
  if (window.__statusesInstalled) return;
  window.__statusesInstalled = true;
  window.__statuses = [];
  const el = document.querySelector('#send-status');
  if (!el) return;
  const rec = (v) => {
    const l = window.__statuses;
    if (!l.length || l[l.length - 1] !== v) l.push(v);
  };
  rec(el.textContent);
  new MutationObserver(() => rec(el.textContent)).observe(el, {
    childList: true,
    subtree: true,
  });
}
"""

#: Captures the raw SSE ``data:`` payloads of the /api/chat stream
#: (a response clone read in the background) — wire-level assertions
#: for the ``tool`` frames, independent of the UI rendering.
SSE_HOOK = """
() => {
  if (window.__sseInstalled) return;
  window.__sseInstalled = true;
  window.__sseFrames = [];
  const origFetch = window.fetch;
  window.fetch = async function (...args) {
    const res = await origFetch.apply(this, args);
    try {
      const url = typeof args[0] === 'string' ? args[0] : args[0].url;
      if (url.includes('/api/chat')) {
        res.clone().text().then((bodyText) => {
          for (const block of bodyText.split('\\n\\n')) {
            const line = block.trim();
            if (line.startsWith('data: ')) {
              window.__sseFrames.push(line.slice(6));
            }
          }
        });
      }
    } catch (e) { /* non-clonable responses: ignored */ }
    return res;
  };
}
"""


def _install_page_hooks(page: Page) -> None:
    """Install all hooks on the loaded page (post-goto, pre-submit)."""
    page.evaluate(SSE_HOOK)
    page.evaluate(LABEL_RECORDER)
    page.evaluate(STATUS_RECORDER)


def _frames(page: Page) -> list[dict]:
    """The captured SSE frames, once the hook's background read settles.

    The hook reads ``res.clone().text()`` in a background promise that
    resolves right after the stream closes — poll briefly until the
    final ``done`` frame lands (fail loud if the hook captured nothing).
    """
    deadline = time.monotonic() + 10.0
    while True:
        raw = page.evaluate("() => window.__sseFrames || []")
        parsed = [json.loads(line) for line in raw if line]
        if any(f.get("type") == "done" for f in parsed):
            return parsed
        if time.monotonic() > deadline:
            raise AssertionError(
                f"SSE hook captured no `done` frame (frames so far: "
                f"{len(parsed)}) — hook install failed?"
            )
        time.sleep(0.05)


def _tool_frames(frames: list[dict]) -> list[dict]:
    return [f for f in frames if f.get("type") == "tool"]


def _submit(page: Page, question: str) -> None:
    page.fill("#message-input", question)
    page.click("#send-btn")
    # The user bubble lands synchronously with the submit handler.
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)


def _wait_settled(page: Page) -> None:
    """The turn is complete: answer text in the bubble, button recovered.

    Phase 48: the label assertion carries the settle wait with an
    explicit timeout — the in-flight button is the enabled Stop control
    (never disabled), so ``to_be_enabled`` no longer blocks until the
    turn settles, and Playwright expect's default (5s) does not inherit
    the page default."""
    expect(page.locator(".msg.brain .bubble").last).not_to_have_text("", timeout=30_000)
    expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)


def _assert_no_error_banner(page: Page) -> None:
    """The turn must settle WITHOUT the terminal error banner (the red
    role=alert error banner — the KB-offline banner is a separate,
    non-error state)."""
    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"))


# --------------------------------------------------------------------------
# 1. The live search flow: search → matched line → "Found …" answer
# --------------------------------------------------------------------------


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

    _submit(page, SEARCH_QUESTION)
    # The "calling tool" STATUS window is transient: the first `tool`
    # frame sets #send-status and it holds until the FIRST answer delta
    # (the agent loop completes before the answer stream) — a polling
    # expect can stride straight over that window (the phase-44 flake),
    # so the pre-submit MutationObserver records below are the
    # deterministic source of truth for the status transitions.
    _wait_settled(page)

    # Phase 48 (owner-locked): the in-flight button is the Stop control
    # for the whole turn; #send-status walked "… is thinking" →
    # "… is searching for <sentinel>", in order.
    labels = page.evaluate("() => window.__labels")
    assert "Stop" in labels, labels
    statuses = page.evaluate("() => window.__statuses")
    i_search = next(
        (
            i
            for i, s in enumerate(statuses)
            if f"is searching for {SEARCH_PATTERN}" in s
        ),
        None,
    )
    assert i_search is not None, statuses
    i_think = next(
        (i for i, s in enumerate(statuses) if "is thinking" in s), None
    )
    assert i_think is not None and i_think < i_search, statuses

    # Wire level: exactly ONE `tool` frame — grep carrying the PATTERN
    # as its argument (phase 68 task 02; phase 70 renamed the tool) —
    # ahead of the first `delta` frame.
    frames = _frames(page)
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "grep", "argument": SEARCH_PATTERN}
    ]
    first_delta = next(i for i, f in enumerate(frames) if f.get("type") == "delta")
    assert all(
        i < first_delta for i, f in enumerate(frames) if f.get("type") == "tool"
    )
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is False

    # ONE tool line above the answer: "🔎 Searching for " + the sentinel
    # in a <code> element (the pattern is data, never markup).
    lines = page.locator(".msg.brain .tool-call")
    expect(lines).to_have_count(1)
    expect(lines.nth(0)).to_contain_text("Searching for")
    expect(lines.nth(0).locator("code")).to_have_text(SEARCH_PATTERN)

    # The answer quotes the MATCHED LINE — the search result reached the
    # model and landed in the answer (the mock's deterministic echo).
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(FOUND_ANSWER)

    # The turn settled to idle WITHOUT the error banner, and the durable
    # record is grounded with the retrieval doc only (see test 2 for the
    # sources contract).
    _assert_no_error_banner(page)
    row = _last_query_log()
    assert row.question == SEARCH_QUESTION
    assert row.deflected is False


# --------------------------------------------------------------------------
# 2. Context accounting (locked A5): a search adds no source by itself
# --------------------------------------------------------------------------


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

    _submit(page, SEARCH_QUESTION)
    _wait_settled(page)

    # The search really ran (its wire frame is present) — yet the
    # search-only flow (no read) adds NO cited source: phase 119,
    # LOCKED A1 — done.sources is the READ docs only, and nothing was
    # read (the retrieval doc was never read, so it never chips — the
    # retired phase-118 A4 suggested+read union is gone).
    frames = _frames(page)
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "grep", "argument": SEARCH_PATTERN}
    ]
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is False
    assert done["sources"] == [], done["sources"]

    # UI: ZERO source chips — the search renders no chip of its own, and
    # the never-read retrieval doc chips nothing (phase 119 A1).
    expect(page.locator(".msg.brain .source-chip")).to_have_count(0)

    # Durable record: the sources row keeps the retrieval baseline (the
    # search adds nothing to it either — LOCKED A3, untouched).
    row = _last_query_log()
    assert row.deflected is False
    assert row.sources == SEED_SP


# --------------------------------------------------------------------------
# 3. Persistence: the search tool line re-renders after a reload
# --------------------------------------------------------------------------


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

    _submit(page, SEARCH_QUESTION)
    _wait_settled(page)
    expect(page.locator(".msg.brain .tool-call")).to_have_count(1)

    page.reload()
    expect(page.locator("#empty-state")).to_be_hidden()

    # The persisted record (the generic {name, argument} toolAcc —
    # phase 14 convention) re-renders the search line through the same
    # append helper as the live frames: sentinel in a <code> element.
    restored = page.locator(".msg.brain .tool-call")
    expect(restored).to_have_count(1)
    expect(restored.nth(0)).to_contain_text("Searching for")
    expect(restored.nth(0).locator("code")).to_have_text(SEARCH_PATTERN)

    # The answer (the matched-line echo) is intact after the restore.
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(FOUND_ANSWER)