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

"""The 2026-09-05 incident E2E (Playwright, mock-only): the
grep-regex-teaching self-correction loop through the real UI.

Context: the sample question "What are the correct llama.cpp arguments
for Qwen 3.8?" failed repeatedly against the live KB. The harness
prior is that ``grep(pattern)`` takes a REGEX (pi.dev's grep, ripgrep,
grep itself); this app's grep is a case-insensitive fixed substring
(owner-locked A5 — the contract does not change). A regex-shaped
first grep (``qwen.*3\\.8``) can therefore NEVER match, and the bare
"no matches" result made the turbo model trust the miss and end the
turn with a wrong "I searched the entire knowledge base" refusal.
The fix under test (``app.rag.agent``): a no-match for a regex-shaped
pattern is the TEACHING line (``NO_MATCHES_REGEX`` /
``NO_MATCHES_REGEX_SCOPED``) — the plain-substring contract stated,
the ``plain_form`` retry hint handed over — so the model self-corrects
in one round.

Run in isolation (DB must be up: ``podman compose up -d db``):

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

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the gate is the
deterministic GREP-TEACH flow in ``tests/e2e/mock_llm.py``
(``GREP_TEACH_TRIGGER`` — "what are the correct llama.cpp arguments"
— + the HIGH prompt's ``<tools>`` section): the incident's regex-
shaped first grep (``grep`` with ``{"pattern": "qwen.*3\\.8"}``, id
``call_0``) → the agent's TEACHING no-match line (the mock's plain
step keys on the marker ``grep matches a plain substring``) → the
plain-form retry (``grep`` with ``{"pattern": "qwen3.8"}``, id
``call_1``) → the match → the ``read`` of the first match line's
document (id ``call_2``) → the deterministic echo answer.

KB fixture (TRUNCATE-then-seed, house pattern): ONE source with TWO
documents of known ``source``/``path``/``title`` (catalog order =
``(source, path)``, so the first match line is deterministic):

* ``Homelab/llama-server-args.md`` — the CATALOG-FIRST document,
  indexed WITHOUT chunks (catalog-only; never in the retrieval
  context, so the flow's ``read`` of it is NOT deduped as
  already-in-context). It carries the literal line the plain grep
  finds (``qwen3.8-27b`` — NOT the regex-shaped ``qwen.*3\\.8``,
  which can never match a fixed-substring grep) and the search-flow
  sentinel line (the no-regression turn). Its FIRST line is longer
  than 80 chars, so the mock's first-80-chars quote stays
  newline-free.
* ``Homelab/ai-stack-notes.md`` — the retrievable document: one chunk
  whose embedding is the mock's own bag-of-words vector (the trigger
  question cosines well past the E2E 0.30 threshold → grounded, the
  ``<tools>`` section rides along). It carries NO ``qwen3.8`` line
  (the plain grep's match is the catalog-first document alone) and
  its name carries no name-hit token (it stays the vector seed only).

Test → phase mapping (Playwright Mapping Rule):
1. ``test_regex_grep_self_corrects_to_plain_form`` — the grounded
   GREP-TEACH turn: the turn settles (composer re-enables, ``done``
   observed), the answer bubble carries the read document's echo
   (``Read Homelab/llama-server-args.md. <first 80 chars>``), the UI
   shows the three tool lines (two ``Searching for`` lines + the
   ``Reading`` line), and no error banner. Wire level: the ``tool``
   frames arrive in order — ``grep`` ``qwen.*3\\.8`` → ``grep``
   ``qwen3.8`` → ``read`` the combined identity — and there is NO
   fourth ``tool`` frame (the loop ended in one correction, not at
   the round cap).
2. ``test_plain_search_flow_not_swallowed_by_new_trigger`` — in the
   SAME session, the GREP-TEACH turn settles and a follow-up question
   carrying ``SEARCH_TRIGGER`` still settles with the search flow's
   ``Found …`` answer (the new flow did not swallow the existing
   trigger).
"""
from __future__ import annotations

import hashlib
import json
import re
import time
from datetime import UTC, datetime
from pathlib import Path

from playwright.sync_api import Page, expect
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.db import SessionLocal
from app.models import Chunk, Document
from e2e.auth_helpers import login
from tests.e2e.mock_llm import (
    GREP_TEACH_MARKER,
    GREP_TEACH_PATTERN,
    GREP_TEACH_PLAIN,
    GREP_TEACH_TRIGGER,
    SEARCH_PATTERN,
    SEARCH_TRIGGER,
    embed_text,
)

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

# --------------------------------------------------------------------------
# The one-source, two-document fixture (see the module docstring)
# --------------------------------------------------------------------------

SEED_SOURCE = "Homelab"
DOC1_PATH = "llama-server-args.md"
DOC1_TITLE = "Llama Server Args"
DOC1_SP = f"{SEED_SOURCE}/{DOC1_PATH}"

DOC2_PATH = "ai-stack-notes.md"
DOC2_TITLE = "AI Stack Notes"
DOC2_SP = f"{SEED_SOURCE}/{DOC2_PATH}"

#: The catalog-first document (catalog order = (source, path) — DOC1
#: sorts before DOC2): the plain grep's ONLY match, and the document
#: the flow reads (so it must NOT be the seed — a seed read dedupes to
#: "Already in your context.", which the mock flow does not model).
#: Indexed WITHOUT chunks: catalog-only, never in the retrieval
#: context. Line 1 (the read quote) is >80 chars and newline-free; it
#: carries the literal ``qwen3.8-27b`` text (NEVER the regex-shaped
#: ``qwen.*3\.8`` — that is the whole point of the incident).
DOC1_CONTENT = (
    "The llama.cpp server launch line for the qwen3.8-27b juggernaut "
    "deployment pins the sampling and speculative-decoding flags.\n"
    "Server command (verbatim): --port 8000 -ctk q8_0 -ctv q8_0 "
    "--kv-unified -fa on --n-gpu-layers all --jinja.\n"
    "Model file: /models/qwen3.8-27b/Qwen3.8-27B-UD-Q6_K.gguf with "
    "the mmproj-BF16.gguf projector and the custom jinja template.\n"
    f"Regression sentinel line: {SEARCH_PATTERN} must stay findable "
    "by the plain search flow.\n"
)
assert "\n" not in DOC1_CONTENT[:63]  # the read quote's content part stays one line
assert GREP_TEACH_PLAIN in DOC1_CONTENT  # the plain grep matches DOC1
assert GREP_TEACH_PATTERN not in DOC1_CONTENT  # the regex never matches
assert GREP_TEACH_MARKER not in DOC1_CONTENT  # the marker stays tool-side

#: The retrievable document (the grounded seed context): repeated lines
#: carry the trigger question's key tokens (llama, cpp, arguments,
#: qwen) — well past the E2E 0.30 cosine threshold — but NO literal
#: ``qwen3.8`` line (the plain grep's match stays DOC1 alone) and the
#: name carries no name-hit token (the seed stays the vector side
#: only). FIRST line >80 chars, newline-free (the seed context stays
#: one clean line).
DOC2_CONTENT = (
    "Notes on the self-hosted ai stack: the llama cpp server arguments "
    "for every qwen model are kept next to the quadlet files.\n"
    + (
        "The llama cpp server arguments — sampling, context, kv cache "
        "quantization — are documented per model in the quadlet notes.\n"
    )
    * 10
    + "\n## Server notes\n\n"
    "Every qwen deployment shares the same llama cpp sampling "
    "defaults; the per-model file overrides the speculative flags.\n"
)
assert "\n" not in DOC2_CONTENT[:80]
assert GREP_TEACH_PLAIN not in DOC2_CONTENT  # the match stays DOC1 alone

#: Carries ``GREP_TEACH_TRIGGER`` (the incident's sample question,
#: near-verbatim) and nothing else — no other mock marker.
GREP_TEACH_QUESTION = (
    "What are the correct llama.cpp arguments for Qwen 3.8? Show me "
    "the exact server launch line from my notes."
)
assert GREP_TEACH_TRIGGER in GREP_TEACH_QUESTION.lower()
for _other in (
    "use your tools",
    "read two documents",
    "search your documents",
    "list the files in this directory",
    "emit raw tool markup",
    "always emit raw tool markup",
    "show me a table",
    "think in paragraphs",
    "think out loud then hesitate",
    "think out loud",
    "show the end of your notes",
    "write a long answer",
    "fail then answer",
    "always fail",
    "embed fail once",
    "pretend to think slowly",
):
    assert _other not in GREP_TEACH_QUESTION.lower(), _other

#: Carries ``SEARCH_TRIGGER`` (the phase-68 search flow) and nothing
#: else — the no-regression follow-up question in the same session.
SEARCH_QUESTION = (
    "Search your documents for the reese-sentinel-42 marker and tell "
    "me the line that carries it."
)
assert SEARCH_TRIGGER in SEARCH_QUESTION.lower()
for _other in (
    GREP_TEACH_TRIGGER,
    "use your tools",
    "read two documents",
    "list the files in this directory",
    "emit raw tool markup",
    "always emit raw tool markup",
    "show me a table",
    "think in paragraphs",
    "think out loud then hesitate",
    "think out loud",
    "show the end of your notes",
    "write a long answer",
    "fail then answer",
    "always fail",
    "embed fail once",
    "pretend to think slowly",
):
    assert _other not in SEARCH_QUESTION.lower(), _other

#: The mock's deterministic read echo (the read document reached the
#: model and landed in the answer) — the flow reads DOC1 (the plain
#: grep's first — only — match line). Phase 106 (D5): the read result's
#: ``date:`` SECOND line rides into the first-80-chars quote — the
#: date line (the fixture's fixed ``created_at`` UTC date part, 17
#: chars; its trailing newline renders as a markdown soft break — no
#: text between the date and the content) + the first 63 content chars
#: (80 − 17).
READ_ANSWER_PREFIX = f"Read {DOC1_SP}."
READ_ANSWER_QUOTE = "date: 2024-06-15" + DOC1_CONTENT[:63]


def _seed_fixture(db: Session) -> None:
    """The one-source, two-document fixture (see the module docstring).

    DOC1 (catalog-first, the grep match, the read target) is indexed
    WITHOUT chunks; DOC2 carries the single chunk (the mock's own
    embedding → the trigger question cosines well past the E2E 0.30
    threshold → grounded, the ``<tools>`` section rides along).
    """
    # Phase 106 (D5): explicit dates — byte-stable prompts/quotes (the
    # mock's first-80-chars read quote carries the date line).
    db.add(
        Document(
            source=SEED_SOURCE,
            path=DOC1_PATH,
            full_path=f"/tmp/{DOC1_PATH}",
            title=DOC1_TITLE,
            content=DOC1_CONTENT,
            content_hash=hashlib.sha256(DOC1_CONTENT.encode()).hexdigest(),
            indexed_at=datetime.now(UTC),
            created_at=datetime(2024, 6, 15, tzinfo=UTC),
        )
    )
    doc2 = Document(
        source=SEED_SOURCE,
        path=DOC2_PATH,
        full_path=f"/tmp/{DOC2_PATH}",
        title=DOC2_TITLE,
        content=DOC2_CONTENT,
        content_hash=hashlib.sha256(DOC2_CONTENT.encode()).hexdigest(),
        indexed_at=datetime.now(UTC),
        created_at=datetime(2024, 6, 15, tzinfo=UTC),
    )
    db.add(doc2)
    db.flush()
    db.add(
        Chunk(
            document_id=doc2.id,
            position=0,
            content=DOC2_CONTENT,
            embedding=embed_text(DOC2_CONTENT),
        )
    )


def _reset_db_fixture() -> None:
    """Truncate the KB (plus the prompt-shaping tables), then seed the
    one-source, two-document fixture. ``steering_notes`` /
    ``kb_overview`` are truncated too, so the HIGH prompt is exactly
    ``<relevance>`` + ``<documents>`` + ``<tools>`` — byte-stable
    prompts, byte-stable answers."""
    with SessionLocal() as db:
        db.execute(
            text("TRUNCATE chunks, documents, query_log, steering_notes, kb_overview")
        )
        db.commit()
        _seed_fixture(db)
        db.commit()


# --------------------------------------------------------------------------
# Page helpers (the house pattern — cf. test_tool_path_teaching.py)
# --------------------------------------------------------------------------

#: 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_sse_hook(page: Page) -> None:
    page.evaluate(SSE_HOOK)


def _drain_frames(page: Page) -> list[dict]:
    """One turn's SSE frames: wait for that turn's ``done`` frame, then
    return EVERY frame captured since the last drain (the hook's
    background read appends the whole stream at once after it closes, so
    clearing-and-reading is race-free per turn)."""
    deadline = time.monotonic() + 10.0
    while True:
        raw = page.evaluate(
            "() => { const f = window.__sseFrames || []; "
            "window.__sseFrames = []; return f; }"
        )
        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")
    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."""
    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 settled through the normal done 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"))


# --------------------------------------------------------------------------
# 1. The grounded GREP-TEACH turn: the incident's regex-shaped first
#    grep → the teaching no-match line → the plain-form retry → the
#    match → the read → the echo answer — the loop settles in ONE
#    correction (three tool rounds), pinned on the SSE wire
# --------------------------------------------------------------------------


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

    _submit(page, GREP_TEACH_QUESTION)
    _wait_settled(page)

    # Self-correction: the answer quotes the READ document — the plain
    # grep's match was located, the document reached the model, and the
    # echo landed in the answer (the wrong-deflection end state — no
    # tool success, an "I searched everything" refusal — is impossible
    # on this wire: the done frame below is not deflected).
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(READ_ANSWER_PREFIX)
    expect(bubble).to_contain_text(READ_ANSWER_QUOTE)
    _assert_no_error_banner(page)

    # The UI shows the three tool lines in order: the regex-shaped
    # first grep (Searching for <code>qwen.*3\.8</code>), the
    # plain-form retry (Searching for <code>qwen3.8</code>), the read
    # of the combined identity.
    lines = page.locator(".msg.brain .tool-call")
    expect(lines).to_have_count(3)
    expect(lines.nth(0)).to_contain_text("Searching for")
    expect(lines.nth(0).locator("code")).to_have_text(GREP_TEACH_PATTERN)
    expect(lines.nth(1)).to_contain_text("Searching for")
    expect(lines.nth(1).locator("code")).to_have_text(GREP_TEACH_PLAIN)
    expect(lines.nth(2)).to_contain_text("Reading")
    expect(lines.nth(2).locator("code")).to_have_text(DOC1_SP)

    # Three rounds on the wire: the tool frames arrive in order —
    # grep qwen.*3\.8 (the incident's regex-shaped first call), grep
    # qwen3.8 (the plain-form correction the teaching line triggered),
    # read the combined identity — and there is NO fourth tool frame:
    # the loop ended in one correction, not at the round cap.
    frames = _drain_frames(page)
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "grep", "argument": GREP_TEACH_PATTERN},
        {"type": "tool", "name": "grep", "argument": GREP_TEACH_PLAIN},
        {"type": "tool", "name": "read", "argument": DOC1_SP},
    ]
    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
    assert not [f for f in frames if f.get("type") == "error"]


# --------------------------------------------------------------------------
# 2. No regression to the plain search flow — the SAME session: after
#    the GREP-TEACH turn, the SEARCH_TRIGGER follow-up (the phase-68
#    search flow) still settles with the "Found …" answer
# --------------------------------------------------------------------------


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

    # Turn 1 — the GREP-TEACH flow (the incident's regex-shaped first
    # grep → the teaching line → the plain-form retry → the read → the
    # echo answer).
    _submit(page, GREP_TEACH_QUESTION)
    _wait_settled(page)
    teach_frames = _drain_frames(page)
    assert _tool_frames(teach_frames) == [
        {"type": "tool", "name": "grep", "argument": GREP_TEACH_PATTERN},
        {"type": "tool", "name": "grep", "argument": GREP_TEACH_PLAIN},
        {"type": "tool", "name": "read", "argument": DOC1_SP},
    ]
    expect(
        page.locator(".msg.brain .bubble").last
    ).to_contain_text(READ_ANSWER_PREFIX)

    # Turn 2 — the SAME session: the phase-68 search flow on
    # SEARCH_TRIGGER. The new flow must not have swallowed the existing
    # trigger: the follow-up settles with the search flow's answer
    # (grep the sentinel → "Found <first matched line>").
    _submit(page, SEARCH_QUESTION)
    _wait_settled(page)

    second_msg = page.locator(".msg.brain").last
    lines = second_msg.locator(".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 is the search flow's echo: "Found <first matched
    # line's content up to 80 chars>" — the sentinel line from DOC1.
    sentinel_line = next(
        line for line in DOC1_CONTENT.splitlines() if SEARCH_PATTERN in line
    )
    bubble = second_msg.locator(".bubble").last
    expect(bubble).to_contain_text(f"Found {sentinel_line[:80]}")
    _assert_no_error_banner(page)

    # Wire level for the follow-up: one grep of the sentinel pattern —
    # the search flow, unchanged.
    frames = _drain_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 not [f for f in frames if f.get("type") == "error"]