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

"""Phase 188 (task 05) E2E (Playwright, mock-only): the ``chunks.embedding``
HNSW index reaches the owner's chat, changes nothing the owner sees, and the
one retrieval knob the phase adds is honoured on a real turn.

Source: owner incident 2026-10-15. ``pg_indexes`` for ``chunks`` listed the
PK, ``ix_chunks_document_id`` and ``ix_chunks_tsv`` and NOTHING on
``embedding``, so every turn's A7 cosine tier (``ORDER BY embedding <=> :vec
LIMIT 100``) full-scanned a ~3 GB, 414,220-row relation on the interactive
path. Migration ``0029`` adds ``ix_chunks_embedding_hnsw`` (HNSW,
``vector_cosine_ops``, ``m = 16``, ``ef_construction = 64``) and task 02 adds
``BOR_RETRIEVAL_EF_SEARCH`` — shipped at **400** on task 04's measured
decision — applied as ``SET LOCAL hnsw.ef_search`` inside the turn's own
transaction.

**What this module owns, and what it does not.** The plan-level proof (the
tier's captured statement really names the index, and a turn's ``SET LOCAL``
reaches the connection) lives in
``tests/integration/test_chunks_hnsw_recall.py`` (tasks 03/04) and the DDL
round trip in ``tests/integration/test_chunks_hnsw_index.py`` (task 01).
THIS file is the last mile: one question, in a real browser, over a real
import, with the index DROPPED and RESTORED around it — an A/B the owner
could run on their own box. The corpus is 11 small documents, so Postgres'
cost model may well serve BOTH arms from a sequential scan; that is stated,
not hidden. An index at this size is not a speed claim, it is a
**non-regression** claim, and the only non-regression that matters is the
answer. So what is compared between the arms is what reached the model: the
durable retrieval record (``query_log.sources`` / ``top_score`` /
``chunk_hits`` / ``fts_hits`` — LOCKED A3's witness), the ``done`` frame's
tiers, and the rendered "Nearby docs, in case" row. The DDL state is read
back from the catalog INSIDE each arm, so "OFF" can never quietly mean "ON".

Both sentinels are calibrated against the deterministic mock embedder (the
``tests/e2e/test_retrieval_quality.py`` convention, not invented) and the
margins are quoted in the units the honesty gate reads:

* ANSWERED — ``"How do I raise the tilecache eviction watermark so the hot
  zoom levels stay pinned?"`` → best cosine **0.6187**, lexical (FTS) hits
  **10**, leading document ``hnswkb/infra/tilecache.md``. ``0.6187 >= 0.30``
  is A8's HIGH quadrant by a margin of 0.32, so neither arm can flip on a
  rounding difference.
* DEFLECTED — ``"Which bow should I draw for a baroque viola sonata?"`` (no
  note in this KB can answer it) → best cosine **0.1706**, lexical hits **0**.
  A8 deflects only on weak cosine AND zero lexical hits, and both halves hold
  with room: 0.13 BELOW the 0.30 threshold, and a lexical side that is empty
  rather than merely low.

Run in isolation (DB up: ``podman compose up -d db``, schema at head so
``0029`` has run — the module fails loudly, never vacuously, if it has not):

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

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the cosine and lexical
numbers above are properties of the deterministic mock, and the parity pins
compare them exactly between the two arms.

Isolation (the house shape): ONE module app on its own port (the conftest
session app owns 8123 in a combined run), the mock LLM on the conftest port,
a dedicated ``BOR_SOURCES_DIR`` / ``BOR_PDF_PAGE_DIR`` so nothing writes into
the operator's, the shared E2E Postgres truncated BEFORE **and** AFTER every
test, and the KB re-imported through the REAL importer for every test — so
the HNSW insert path runs in every arm, including the ones that then drop the
index. The admin enters through the real form login.

Test → observable mapping (Playwright Mapping Rule — the rendered DOM is the
observable, the durable row is the witness, and the two are read as one
comparison):

1. ``test_answered_question_cites_the_same_document_indexed_or_not`` — the answered
   sentinel, twice: with ``ix_chunks_embedding_hnsw`` present, then after the
   real ``0029`` downgrade drops it (restored in ``finally``, and again by the
   module guard). Both arms ground (no ``.is-deflected``, the mock marker
   renders), both cite the SAME document — ``hnswkb/infra/tilecache.md`` first
   in ``query_log.sources`` and first in the rendered row — and agree on
   ``top_score`` to within ``1e-9``, on ``chunk_hits`` and ``fts_hits``, and
   on the byte-identical answer. The DDL really moved; the answer did not.
2. ``test_index_exists_and_is_valid_after_import`` — the import that just ran
   left the index present, VALID in ``pg_index``, HNSW over
   ``vector_cosine_ops`` per ``pg_get_indexdef``, and carrying pages. An
   INVALID index is silent (the failure ``0029`` exists to catch), so the
   post-import state is pinned instead of assumed.
3. ``test_near_miss_question_still_deflects`` — the out-of-KB sentinel in
   BOTH DDL arms: the amber deflection, the "Maybe try" chips, the ``done``
   frame citing nothing, and the row's two gate inputs read in the units A8
   defines. An index must never manufacture a high ``top_score``.
4. ``test_ef_search_setting_is_honoured`` — two extra apps on their own ports
   with stdout captured (the surface the operator greps): ``BOR_RETRIEVAL_EF_SEARCH=150``
   answers the sentinel normally and logs NO clamp line; ``=50`` answers it
   just as well — same leading doc, same ``top_score`` — and logs the
   operator-visible ``retrieval: ef_search=50 raised to 100 (A7 candidate
   count)`` WARNING. The clamp keeps its promise in the browser, not only in
   a unit test.
"""
from __future__ import annotations

import asyncio
import contextlib
import json
import os
import re
import subprocess
import sys
import time
from collections.abc import Iterator
from pathlib import Path
from threading import Thread
from typing import Any

import pytest
from alembic.config import Config as AlembicConfig
from playwright.sync_api import Page, expect
from sqlalchemy import func, select, text

from alembic import command
from app.config import Settings as _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 e2e.conftest import (
    ADMIN_PASSWORD,
    MOCK_PORT,
    SESSION_SECRET,
    USE_REAL_LLM,
    _wait_http,
)

REPO = Path(__file__).resolve().parents[2]

#: The tracked fixture KB — 11 markdown notes over 6 folders. Tracked rather
#: than generated: the cosine/lexical calibration in the module docstring
#: belongs to THESE bytes, so a re-run ranks the same way on any day.
FIXTURES = REPO / "tests" / "fixtures" / "hnswkb"

#: The ``source`` the importer records for a walked root (``Path(path).name``).
SOURCE_NAME = "hnswkb"
N_DOCS = 11

#: Dedicated module ports (the house env-overridable pattern): the conftest
#: session app owns 8123, phase 187's modules own 8203/8204.
APP_PORT = int(os.environ.get("E2E_APP_PORT_VECIDX", "8205"))
APP_URL = f"http://127.0.0.1:{APP_PORT}"
EF_OK_PORT = int(os.environ.get("E2E_APP_PORT_VECIDX_EF150", "8206"))
EF_OK_URL = f"http://127.0.0.1:{EF_OK_PORT}"
EF_CLAMP_PORT = int(os.environ.get("E2E_APP_PORT_VECIDX_EF50", "8207"))
EF_CLAMP_URL = f"http://127.0.0.1:{EF_CLAMP_PORT}"

#: Migration 0029 and the revision it sits on — the pair the drop/restore
#: walks, moved in the migration's OWN directions. This file contains no
#: inline ``CREATE INDEX``/``DROP INDEX``: 0029 is the single definition of
#: the index these pins name.
REV_HNSW = "0029"
REV_BEFORE = "0028"

#: The index 0029 builds.
INDEX_NAME = "ix_chunks_embedding_hnsw"

#: The answered sentinel (measured, module docstring): cosine 0.6187, 10
#: lexical hits, ``infra/tilecache.md`` first in the fused record.
ANSWER_QUESTION = (
    "How do I raise the tilecache eviction watermark so the hot zoom levels "
    "stay pinned?"
)
#: The document the owner asked about — the one both arms must cite.
CITED_DOC = f"{SOURCE_NAME}/infra/tilecache.md"

#: The near-miss sentinel (measured): cosine 0.1706, 0 lexical hits.
DEFLECT_QUESTION = "Which bow should I draw for a baroque viola sonata?"

#: The mock's grounded answer marker (``tests/e2e/mock_llm.py``).
MOCK_ANSWER_MARKER = "Deterministic mock answer for E2E"

#: The honest-deflection phrase the mock answers with (the house constant —
#: ``test_honest_deflection.py`` et al., phase 52's copy as the deterministic
#: mock speaks it).
DEFLECT_PHRASE = r"haven't done anything like that"

#: The app's honesty gate — the values the module app boots with, named once
#: here so the pins quote the gate they are measured against.
GATE_THRESHOLD = 0.30
GATE_LEXICAL_FLOOR = 0.15

#: The A7 cosine tier's candidate count (``hybrid_vector_candidates``) — the
#: floor the ``ef_search`` clamp raises to. Read from the code default instead
#: of restated, so a change to the knob moves the pin with it.
A7_CANDIDATES = int(_Settings.model_fields["hybrid_vector_candidates"].default)

#: The suggested tier's size (``suggested_docs``) — the split between what the
#: prompt is seeded with and what renders as "Nearby docs, in case".
SUGGESTED_DOCS = int(_Settings.model_fields["suggested_docs"].default)

#: The clamp line ``app/rag/retriever.py`` logs (AGENTS rule 10: a knob that
#: would degrade recall is never silent), matched on the app's captured
#: stdout: ``… WARNING app.rag.retriever :: retrieval: ef_search=50 raised
#: to 100 (A7 candidate count)``.
CLAMP_LINE = re.compile(
    r"WARNING\s+app\.rag\.retriever\s+::\s+retrieval: ef_search=(\d+) "
    r"raised to (\d+) \(A7 candidate count\)"
)

#: Numeric slack between the arms' ``top_score``. Both arms run the same exact
#: cosine tier over this corpus, so the true slack is zero; the epsilon only
#: absorbs the float's travel through the row.
SCORE_SLACK = 1e-9

#: Turn budget: a mock turn lands in well under a second; 30 s is the house
#: head-room with no client-side hard timeout.
SETTLE_TIMEOUT_MS = 30_000


# ---------------------------------------------------------------------------
# The module app + the two ef_search apps
# ---------------------------------------------------------------------------


def _app_env(scratch: Path, ef_search: str | None = None) -> dict[str, str]:
    """App env: the mock LLM, the mock-calibrated honesty gate (the conftest
    convention — both floors stay ``<=`` the threshold or the app refuses to
    boot), no git-source fallback, dedicated directories so nothing writes
    into the operator's.

    ``ef_search=None`` pins the knob to the CODE default (the phase 188 leak
    guard: an operator's ``.env`` value never reaches a test); any other value
    is what the app boots with.
    """
    env = dict(os.environ)
    env.pop("DEBUGPY", None)
    env["BOR_ENVIRONMENT"] = "e2e"
    env["BOR_STATIC_DIR"] = str(REPO / "frontend")
    env["BOR_LLM_BASE_URL"] = f"http://127.0.0.1:{MOCK_PORT}/v1"
    env["BOR_RELEVANCE_THRESHOLD"] = str(GATE_THRESHOLD)
    env["BOR_LEXICAL_SUPPORT_FLOOR"] = str(GATE_LEXICAL_FLOOR)
    env["BOR_SOURCE_USEFULNESS_FLOOR"] = str(GATE_LEXICAL_FLOOR)
    env.setdefault(
        "BOR_DATABASE_URL",
        "postgresql+psycopg://reese:reese@localhost:5432/brain_of_reese",
    )
    env["BOR_ADMIN_PASSWORD"] = ADMIN_PASSWORD
    env["BOR_SESSION_SECRET"] = SESSION_SECRET
    env["BOR_GIT_SOURCES"] = ""
    env["BOR_SOURCES_DIR"] = str(scratch / "checkouts")
    env["BOR_PDF_PAGE_DIR"] = str(scratch / "pdf-pages")
    env["BOR_RETRIEVAL_EF_SEARCH"] = (
        ef_search
        if ef_search is not None
        else str(_Settings.model_fields["retrieval_ef_search"].default)
    )
    return env


@contextlib.contextmanager
def _running_app(
    url: str, port: int, env: dict[str, str], log_path: Path | None = None
) -> Iterator[tuple[str, Path | None]]:
    """Boot the app on *port*, wait for ``/api/health``, tear it down.

    With *log_path* the app's stdout+stderr go to that file — the same surface
    the operator greps (``grep ' :: ' server.log``), so the clamp pin reads
    what a human would read rather than a test-private logger.
    """
    args = [
        sys.executable,
        "-m",
        "uvicorn",
        "app.main:app",
        "--host",
        "127.0.0.1",
        "--port",
        str(port),
        "--log-level",
        "warning",
    ]
    if log_path is None:
        proc = subprocess.Popen(args, cwd=REPO, env=env)  # noqa: S603
        try:
            _wait_http(f"{url}/api/health")
            yield url, None
        finally:
            proc.terminate()
            with contextlib.suppress(subprocess.TimeoutExpired):
                proc.wait(timeout=10)
        return
    with open(log_path, "wb") as log:
        proc = subprocess.Popen(  # noqa: S603
            args, cwd=REPO, env=env, stdout=log, stderr=subprocess.STDOUT
        )
        try:
            _wait_http(f"{url}/api/health")
            yield url, log_path
        finally:
            proc.terminate()
            with contextlib.suppress(subprocess.TimeoutExpired):
                proc.wait(timeout=10)


@pytest.fixture(scope="module")
def app_server(
    mock_llm: int, tmp_path_factory: pytest.TempPathFactory
) -> Iterator[str]:
    """The app under tests 1-3: the SHIPPED default ``retrieval_ef_search``
    (400 — task 04's measured decision) over the fixture KB."""
    if USE_REAL_LLM:
        pytest.skip(
            "mock-only suite — the cosine/lexical calibration and the two-arm "
            "parity pins belong to the deterministic mock "
            "(E2E_REAL_LLM unsupported)"
        )
    assert mock_llm
    scratch = tmp_path_factory.mktemp("bor_vector_index_app")
    with _running_app(APP_URL, APP_PORT, _app_env(scratch)) as (url, _log):
        yield url


@pytest.fixture(scope="module")
def app_url(app_server: str) -> str:
    return app_server


# ---------------------------------------------------------------------------
# The index state, read from the catalog (never inferred)
# ---------------------------------------------------------------------------


def _index_state(db: Any) -> dict[str, Any] | None:
    """The ``pg_index`` row for :data:`INDEX_NAME`, or ``None`` when absent.

    ``None`` is the OFF arm. ``valid=False`` is a state nothing here may treat
    as ON: an INVALID index exists, is maintained on every insert, and serves
    no queries — silently, which is the failure ``0029`` was written to catch.
    """
    row = db.execute(
        text(
            "SELECT i.indisvalid, pg_table_size(c.oid) AS size_bytes, "
            "pg_get_indexdef(c.oid) AS indexdef "
            "FROM pg_index i "
            "JOIN pg_class c ON c.oid = i.indexrelid "
            "JOIN pg_namespace n ON n.oid = c.relnamespace "
            "WHERE c.relname = :name AND n.nspname = 'public'"
        ),
        {"name": INDEX_NAME},
    ).first()
    if row is None:
        return None
    return {
        "valid": bool(row.indisvalid),
        "size_bytes": int(row.size_bytes),
        "indexdef": str(row.indexdef),
    }


def _alembic_config() -> AlembicConfig:
    cfg = AlembicConfig(str(REPO / "alembic.ini"))
    cfg.set_main_option("script_location", str(REPO / "alembic"))
    cfg.attributes["configure_logger"] = False
    return cfg


def _alembic_version() -> str:
    with SessionLocal() as db:
        version = db.execute(text("SELECT version_num FROM alembic_version")).scalar()
        db.rollback()
    return str(version)


def _set_index_present(present: bool) -> None:
    """Move the schema to the index OFF / ON state through 0029's own
    ``downgrade`` / ``upgrade``, and PROVE the move landed.

    Both directions run in the migration's ``autocommit_block`` (a
    ``CONCURRENTLY`` DDL cannot run inside a transaction). This process's own
    session is rolled back FIRST — an open snapshot is what parks a
    ``CONCURRENTLY`` build behind the test session forever, the failure task 01
    documented.
    """
    with SessionLocal() as probe:
        probe.rollback()
    cfg = _alembic_config()
    if present:
        command.upgrade(cfg, REV_HNSW)
    else:
        command.downgrade(cfg, REV_BEFORE)
    with SessionLocal() as db:
        state = _index_state(db)
        version = db.execute(text("SELECT version_num FROM alembic_version")).scalar()
        db.rollback()
    if present:
        assert state is not None and state["valid"], (
            f"restoring {REV_HNSW} left no VALID {INDEX_NAME}: {state}"
        )
        assert str(version) == REV_HNSW, f"alembic_version did not return: {version}"
    else:
        assert state is None, f"{INDEX_NAME} survived the {REV_BEFORE} downgrade: {state}"
        assert str(version) == REV_BEFORE, f"alembic_version did not move: {version}"


@pytest.fixture(scope="module", autouse=True)
def _index_at_head() -> Iterator[None]:
    """Module guard: the schema STARTS and ENDS at 0029 with a valid index.

    A suite that silently ran with the index missing would pass tests 1 and 3
    for the wrong reason, so entering is an assertion rather than a hope and
    the teardown repairs whatever a failed test left behind.
    """
    version = _alembic_version()
    if version != REV_HNSW:
        pytest.fail(
            f"phase 188's task-05 suite needs the schema at {REV_HNSW} "
            f"(found {version!r}) — run `uv run alembic upgrade head`"
        )
    with SessionLocal() as db:
        state = _index_state(db)
        db.rollback()
    if state is None or not state["valid"]:
        _set_index_present(True)
    yield
    _set_index_present(True)


# ---------------------------------------------------------------------------
# The KB, imported through the REAL importer with the index in place
# ---------------------------------------------------------------------------


def _mock_llm_client() -> LLMClient:
    """An embedder aimed at the mock (no ``.env``: the test process must not
    inherit the operator's live endpoint)."""
    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 LLMClient(settings)


async def _import_kb() -> ImportSummary:
    return await import_sources([FIXTURES], _mock_llm_client())


def _run_in_thread(coro: Any) -> Any:
    """Run a coroutine on a worker thread (Playwright owns the test loop)."""
    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

    thread = Thread(target=runner)
    thread.start()
    thread.join()
    if "error" in box:
        raise box["error"]
    return box["value"]


def _truncate_all() -> None:
    """Fresh KB + registry + query log (the phase-184 table list)."""
    with SessionLocal() as db:
        db.execute(
            text(
                "TRUNCATE chunks, documents, query_log, kb_overview, "
                "git_sources, sources_meta, folder_summaries"
            )
        )
        db.commit()  # without the commit the TRUNCATE rolls back


@pytest.fixture(autouse=True)
def kb(mock_llm: int) -> Iterator[ImportSummary]:
    """A fresh KB of the tracked fixture tree, imported with the HNSW index
    PRESENT, so every arm — including the ones that then drop it — put its
    embeddings into the graph.

    One fixture owns the whole boundary so the order cannot drift: truncate →
    assert the index is live → import → test → truncate. The index assertion
    runs AFTER the truncate because a TRUNCATE of an HNSW-indexed table is
    exactly the insert path this suite is about.
    """
    _truncate_all()
    with SessionLocal() as db:
        state = _index_state(db)
        db.rollback()
    assert state is not None and state["valid"], (
        f"{INDEX_NAME} must be present and valid BEFORE the import — the "
        f"point of this suite is inserts into a live graph: {state}"
    )

    summary = _run_in_thread(_import_kb())
    assert summary.errors == 0, f"the fixture KB failed to import: {summary}"
    assert summary.added == N_DOCS, f"expected {N_DOCS} fixture notes: {summary}"
    assert summary.summaries == N_DOCS, f"every note needs a summary: {summary}"
    yield summary
    _truncate_all()


# ---------------------------------------------------------------------------
# Helpers: the wire, the DOM, the durable row
# ---------------------------------------------------------------------------

#: Captures the raw SSE ``data:`` payloads of the /api/chat stream (a response
#: clone read in the background) — the wire-level witness, independent of the
#: rendering (the ``test_agent_document_tools.py`` hook).
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 response: ignored */ }
    return res;
  };
}
"""


def _frames(page: Page, since: int) -> list[dict[str, Any]]:
    """This turn's SSE frames, once the hook's background read settles."""
    deadline = time.monotonic() + 10.0
    while True:
        raw = page.evaluate("() => window.__sseFrames || []")
        parsed = [json.loads(line) for line in raw[since:] 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 ({len(parsed)} frames) — "
                "was the hook installed before the turn?"
            )
        time.sleep(0.05)


def _done(frames: list[dict[str, Any]]) -> dict[str, Any]:
    """The turn's ONE ``done`` frame — the shape task 03 pins."""
    dones = [f for f in frames if f.get("type") == "done"]
    assert len(dones) == 1, f"expected exactly one done frame, got {dones}"
    return dones[0]


def _new_chat(page: Page) -> None:
    """Clear the conversation so the NEXT turn is a first turn (no history
    from the previous arm rides the prompt). The control the owner uses."""
    page.click("#new-chat-btn")
    expect(page.locator(".msg")).to_have_count(0)


def _submit(page: Page, question: str) -> int:
    """Send *question* as the turn's ONLY message (phase 162's rule: a trailing
    phrase changes what retrieval sees). Returns the frame count so far."""
    seen = int(page.evaluate("() => (window.__sseFrames || []).length"))
    page.fill("#message-input", question)
    page.click("#send-btn")
    expect(page.locator(".msg.user .bubble").last).to_contain_text(question)
    return seen


def _wait_settled(page: Page) -> None:
    """The settle sync: the client re-labels Send on the ``done`` frame, and
    the server writes the ``query_log`` row just before yielding it."""
    expect(page.locator("#send-label")).to_have_text("Send", timeout=SETTLE_TIMEOUT_MS)


def _row(question: str) -> dict[str, Any]:
    """The durable record for *question* as a plain snapshot — exactly one row
    (the caller clears the log between arms that ask the same question).

    Read as values, not as an ORM instance: the row outlives its session here
    (the arm compares it against another arm's), and a detached instance
    raises the moment an attribute is touched.
    """
    with SessionLocal() as db:
        rows = db.execute(
            select(
                QueryLog.top_score,
                QueryLog.fts_hits,
                QueryLog.chunk_hits,
                QueryLog.deflected,
                QueryLog.sources,
            ).where(QueryLog.question == question)
        ).all()
    assert len(rows) == 1, f"expected one query_log row for {question!r}, got {len(rows)}"
    row = rows[0]
    return {
        "top_score": float(row.top_score),
        "fts_hits": row.fts_hits,
        "chunk_hits": int(row.chunk_hits),
        "deflected": bool(row.deflected),
        "sources": str(row.sources),
    }


def _recorded(row: dict[str, Any]) -> list[str]:
    """``query_log.sources`` as the ordered ``source/path`` list the server
    wrote: the retrieval tier that reached the model (LOCKED A3)."""
    return [part.strip() for part in row["sources"].split(",") if part.strip()]


def _arm(page: Page, question: str, *, expect_index: bool) -> dict[str, Any]:
    """Ask *question* once and collect everything that must not move between
    the arms — the DOM, the wire, the durable row, and the DDL state the arm
    ACTUALLY ran under."""
    with SessionLocal() as db:
        state = _index_state(db)
        db.rollback()
    if expect_index:
        assert state is not None and state["valid"], (
            f"the ON arm ran WITHOUT a valid {INDEX_NAME}: {state}"
        )
    else:
        assert state is None, f"the OFF arm still had {INDEX_NAME}: {state}"

    since = _submit(page, question)
    _wait_settled(page)
    done = _done(_frames(page, since))

    brain = page.locator(".msg.brain").last
    row = _row(question)
    recorded = _recorded(row)
    return {
        "answer": brain.locator(".bubble").first.inner_text(),
        "nearby": brain.locator(".related-docs .related-doc").all_inner_texts(),
        "chips": brain.locator(".source-chip").count(),
        "deflected_bubbles": page.locator(".msg.brain.is-deflected").count(),
        "done": done,
        "index_state": state,
        "row_deflected": row["deflected"],
        "top_score": float(row["top_score"]),
        "fts_hits": row["fts_hits"],
        "chunk_hits": row["chunk_hits"],
        "recorded": recorded,
        "suggested": recorded[:SUGGESTED_DOCS],
        "nearby_recorded": recorded[SUGGESTED_DOCS:],
    }


# ---------------------------------------------------------------------------
# 1 — the answered sentinel: the SAME document, index ON and index OFF
# ---------------------------------------------------------------------------


def test_answered_question_cites_the_same_document_indexed_or_not(
    page: Page, app_url: str
) -> None:
    """The A7/A8 contract in the UI: an index is an ACCESS PATH, so the same
    question must reach the same document whether or not the graph exists."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)
    login(page, app_url, next="/")  # phase 79: the chat is login-gated
    page.evaluate(SSE_HOOK)

    # --- ON: the shipped state (0029 applied, index valid) ------------------
    on = _arm(page, ANSWER_QUESTION, expect_index=True)
    assert on["row_deflected"] is False
    assert on["deflected_bubbles"] == 0
    assert MOCK_ANSWER_MARKER in on["answer"]
    # The measured margin, not just the direction.
    assert on["top_score"] >= GATE_THRESHOLD
    assert on["recorded"][0] == CITED_DOC, on["recorded"]
    assert on["chunk_hits"] > 0
    # Phase 119 (LOCKED A1): a zero-read grounded turn chips nothing — the
    # index changes nothing about that either.
    assert on["chips"] == 0
    # The tier that reached the model, and the tier the page rendered.
    assert CITED_DOC in on["suggested"]
    assert on["nearby"], "the fixture KB must overflow into the rendered row"
    assert on["nearby"] == on["nearby_recorded"]
    assert on["done"]["sources"] == []
    assert [f"{s['source']}/{s['path']}" for s in on["done"]["related"]] == (
        on["nearby_recorded"]
    )

    # --- OFF: the real 0029 downgrade, restored in `finally` ----------------
    with SessionLocal() as db:
        db.execute(text("TRUNCATE query_log"))
        db.commit()
    _new_chat(page)
    _set_index_present(False)
    try:
        off = _arm(page, ANSWER_QUESTION, expect_index=False)
    finally:
        _set_index_present(True)

    # Same verdict, same document, same numbers: the DDL moved, the answer
    # did not.
    assert off["row_deflected"] is False
    assert off["deflected_bubbles"] == 0
    assert MOCK_ANSWER_MARKER in off["answer"]
    assert off["recorded"] == on["recorded"], (
        f"retrieval order moved with the index:\n ON={on['recorded']}\n"
        f" OFF={off['recorded']}"
    )
    assert off["recorded"][0] == CITED_DOC
    assert off["suggested"] == on["suggested"]
    assert off["nearby"] == on["nearby"]
    assert abs(off["top_score"] - on["top_score"]) <= SCORE_SLACK, (
        f"top_score moved with the index: {on['top_score']} vs "
        f"{off['top_score']}"
    )
    assert off["chunk_hits"] == on["chunk_hits"]
    assert off["fts_hits"] == on["fts_hits"]
    assert off["answer"] == on["answer"]
    for key in ("deflected", "sources", "related", "suggestions"):
        assert off["done"].get(key) == on["done"].get(key), (
            f"the done frame's {key} moved with the index: "
            f"{on['done'].get(key)!r} vs {off['done'].get(key)!r}"
        )


# ---------------------------------------------------------------------------
# 2 — the index exists and is VALID after a real import
# ---------------------------------------------------------------------------


def test_index_exists_and_is_valid_after_import(kb: ImportSummary) -> None:
    """The import that just ran left a USABLE graph, not a silent INVALID one.

    ``0029`` post-validates its own build because a ``CONCURRENTLY`` build that
    dies leaves the index INVALID — costing insert time on every sync and
    serving nothing, with no error anywhere. The same check belongs after a
    real import: chunks go into the graph on every sync, and an index that
    drifted INVALID is invisible to every other surface the app has.
    """
    with SessionLocal() as db:
        state = _index_state(db)
        counts = db.execute(
            text("SELECT count(*) AS n, count(embedding) AS with_vec FROM chunks")
        ).one()
        stored = db.scalar(select(func.count()).select_from(QueryLog))
        db.rollback()
    assert stored == 0, "this pin reads the catalog only — no turn has run"

    assert state is not None, f"{INDEX_NAME} is MISSING after the import"
    assert state["valid"] is True, (
        f"{INDEX_NAME} exists but is INVALID after the import — every sync now "
        f"pays for a graph that answers nothing: {state['indexdef']}"
    )
    definition = state["indexdef"]
    assert INDEX_NAME in definition, definition
    assert "hnsw" in definition, definition
    assert "vector_cosine_ops" in definition, definition
    assert "embedding" in definition, definition

    # Not an empty shell over an empty table: the import put real embeddings
    # in, and the graph carries pages for them.
    assert int(counts.n) > 0, "the fixture import produced no chunks"
    assert int(counts.with_vec) == int(counts.n), (
        f"chunks without an embedding would be invisible to the tier: {counts}"
    )
    # Content chunks + one summary chunk per document (phase 30/118) — the
    # importer's own two counts, so a row that never reached the graph shows
    # up here instead of as an invisible hole.
    assert int(counts.n) == kb.chunks + kb.summaries, (
        f"the graph holds {counts.n} rows, the import reported "
        f"{kb.chunks} content + {kb.summaries} summary chunks"
    )
    assert state["size_bytes"] > 0, f"{INDEX_NAME} carries no pages: {state}"


# ---------------------------------------------------------------------------
# 3 — the near-miss question still deflects (both arms)
# ---------------------------------------------------------------------------


def test_near_miss_question_still_deflects(page: Page, app_url: str) -> None:
    """A8 in the browser, index ON and OFF: weak cosine AND zero lexical hits
    still deflects, and an index never manufactures a high ``top_score``."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)
    login(page, app_url, next="/")
    page.evaluate(SSE_HOOK)

    on = _arm(page, DEFLECT_QUESTION, expect_index=True)
    assert on["row_deflected"] is True
    assert on["deflected_bubbles"] == 1
    # The gate's two inputs, in the units A8 defines them.
    assert 0.0 < on["top_score"] < GATE_THRESHOLD
    assert on["fts_hits"] == 0, "deflection is only reached with zero hits"
    # The owner-visible shape of the deflection (phase 52/118).
    expect(page.locator(".msg.brain.is-deflected .bubble")).to_contain_text(
        DEFLECT_PHRASE, timeout=SETTLE_TIMEOUT_MS
    )
    chips = page.locator(".msg.brain.is-deflected .maybe-try .suggestion-chip")
    expect(chips.first).to_be_visible(timeout=SETTLE_TIMEOUT_MS)
    assert chips.count() >= 2, "a deflection must offer 2-3 alternatives"
    assert all(text.strip() for text in chips.all_inner_texts())
    assert on["done"]["deflected"] is True
    assert on["done"]["sources"] == [], "a deflected answer cites nothing"

    # OFF arm: the same verdict, for the same measured reason.
    with SessionLocal() as db:
        db.execute(text("TRUNCATE query_log"))
        db.commit()
    _new_chat(page)
    _set_index_present(False)
    try:
        off = _arm(page, DEFLECT_QUESTION, expect_index=False)
    finally:
        _set_index_present(True)

    assert off["row_deflected"] is True
    assert off["deflected_bubbles"] == 1
    assert 0.0 < off["top_score"] < GATE_THRESHOLD
    assert off["fts_hits"] == 0
    assert abs(off["top_score"] - on["top_score"]) <= SCORE_SLACK
    assert off["recorded"] == on["recorded"]
    assert MOCK_ANSWER_MARKER not in off["answer"]
    assert off["done"]["deflected"] is True


# ---------------------------------------------------------------------------
# 4 — the ef_search knob on a real turn, and the clamp it cannot bypass
# ---------------------------------------------------------------------------


def _log_lines(log_path: Path, pattern: re.Pattern[str]) -> list[str]:
    """Every clamp line the app logged — read off the operator's surface."""
    return [
        line
        for line in log_path.read_text(encoding="utf-8", errors="replace").splitlines()
        if pattern.search(line)
    ]


def test_ef_search_setting_is_honoured(
    page: Page, tmp_path_factory: pytest.TempPathFactory
) -> None:
    """``BOR_RETRIEVAL_EF_SEARCH`` reaches the turn, and a value BELOW the A7
    candidate count is neither silent nor harmful."""
    page.set_default_timeout(SETTLE_TIMEOUT_MS)

    # --- 150 (>= the 100-candidate tier): a normal turn, no clamp line -----
    ok_dir = tmp_path_factory.mktemp("bor_ef_ok")
    with _running_app(
        EF_OK_URL, EF_OK_PORT, _app_env(ok_dir, "150"), ok_dir / "ef150.log"
    ) as (ok_url, ok_log):
        assert ok_log is not None
        login(page, ok_url, next="/")
        page.evaluate(SSE_HOOK)
        on_ok = _arm(page, ANSWER_QUESTION, expect_index=True)
        assert on_ok["row_deflected"] is False
        assert MOCK_ANSWER_MARKER in on_ok["answer"]
        assert on_ok["recorded"][0] == CITED_DOC
        assert _log_lines(ok_log, CLAMP_LINE) == [], (
            "ef_search=150 sits ABOVE the A7 candidate count and must not "
            f"clamp: {_log_lines(ok_log, CLAMP_LINE)}"
        )

    # --- 50 (< the tier): the clamp fires and the turn is UNCHANGED --------
    clamp_dir = tmp_path_factory.mktemp("bor_ef_clamp")
    with _running_app(
        EF_CLAMP_URL, EF_CLAMP_PORT, _app_env(clamp_dir, "50"), clamp_dir / "ef50.log"
    ) as (clamp_url, clamp_log):
        assert clamp_log is not None
        with SessionLocal() as db:
            db.execute(text("TRUNCATE query_log"))
            db.commit()
        login(page, clamp_url, next="/")
        page.evaluate(SSE_HOOK)
        clamped = _arm(page, ANSWER_QUESTION, expect_index=True)
        assert clamped["row_deflected"] is False
        assert MOCK_ANSWER_MARKER in clamped["answer"]
        assert clamped["recorded"][0] == CITED_DOC
        assert clamped["recorded"] == on_ok["recorded"]
        assert abs(clamped["top_score"] - on_ok["top_score"]) <= SCORE_SLACK

        lines = _log_lines(clamp_log, CLAMP_LINE)
        assert lines, (
            "ef_search=50 is BELOW the A7 candidate count and MUST log the "
            "clamp (AGENTS rule 10 — a knob that would degrade recall is never "
            "silent). Tail of the captured log:\n"
            f"{clamp_log.read_text(encoding='utf-8', errors='replace')[-3000:]}"
        )
        for line in lines:
            match = CLAMP_LINE.search(line)
            assert match is not None, line
            assert int(match.group(1)) == 50, line
            assert int(match.group(2)) == A7_CANDIDATES, line