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

"""Phase 72 E2E (Playwright, mock-only): the ls-teaching
self-correction loop through the real UI.

Story: ``.agent/user_stories/agent-document-tools.md`` (this phase
repairs the model-facing contract the phase-70 tools reshaped — the
2026-09-03 incident: the harness-prior ``ls(path='.')`` misuse met the
terse refusal, and the model re-reasoned the same paragraphs over and
over before answering from the seed documents alone).
Run in isolation (DB must be up: ``podman compose up -d db``):

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

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the gate is the
deterministic LS-TEACH flow in ``tests/e2e/mock_llm.py``
(``LS_TEACH_TRIGGER`` — "list the files in this directory" — + the
HIGH prompt's ``<tools>`` section): the incident's misuse (``ls`` with
``{"path": "."}``, id ``call_0``) → the agent's teaching refusal
(``No source named '.' — check the ls output. (…)``) → the corrected
no-arg ``ls()`` (id ``call_1``) → the deterministic
``These are the indexed documents: <first catalog line>`` 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 catalog line is deterministic):

* ``Homelab/aws-route53.md`` — the CATALOG-FIRST document, indexed
  WITHOUT chunks (catalog-only; never in the retrieval context, so
  the single-read flow's ``read`` of it is NOT deduped as already-in-
  context). Its FIRST line is longer than 80 chars, so the mock's
  first-80-chars quote (the single-read regression turn) stays
  newline-free.
* ``Homelab/example-record-file.json`` — 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 and
  FTS-matches too → grounded, the ``<tools>`` section rides along).
  It is the seed context only — the single-read flow reads the
  catalog-FIRST document, not the seed.

Test → phase mapping (Playwright Mapping Rule):
1. ``test_ls_misuse_self_corrects_to_noarg_listing`` — the grounded
   LS-TEACH turn: the turn settles (composer re-enables, ``done``
   observed), the answer bubble carries the first catalog line — the
   first document's ``source:`` / ``path:`` / title fields (the
   catalog reached the model and landed in the answer), the UI shows
   the two tool lines (``🔎 Listing documents in <code>.</code>``
   then ``🔎 Listing documents``), and no error banner. Wire level:
   the ``tool`` frames arrive in order — first ``ls`` with
   ``argument: "."``, then ``ls`` with ``argument: null`` — and there
   is NO third ``tool`` frame (the loop ended in one correction, not
   at the round cap).
2. ``test_plain_tool_flow_not_swallowed_by_new_trigger`` — in the SAME
   session, the LS-TEACH turn settles and a follow-up question
   carrying ``TOOLS_TRIGGER`` (the single-read flow) still settles
   with the read flow's answer (``ls`` → ``read`` on the first
   catalog line's combined identity → ``Read <source/path>. <quote>``)
   — 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 tests.e2e.mock_llm import (
    LS_TEACH_TRIGGER,
    TOOLS_TRIGGER,
    embed_text,
)

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

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

SEED_SOURCE = "Homelab"
DOC1_PATH = "aws-route53.md"
DOC1_TITLE = "AWS Route 53 Notes"
DOC1_SP = f"{SEED_SOURCE}/{DOC1_PATH}"

DOC2_PATH = "example-record-file.json"
DOC2_TITLE = "Example Record File"
DOC2_SP = f"{SEED_SOURCE}/{DOC2_PATH}"

#: The FIRST catalog line (catalog order = (source, path) — DOC1 sorts
#: first): the mock's LS-TEACH answer quotes exactly this line.
FIRST_CATALOG_LINE = (
    f"source: {SEED_SOURCE} | path: {DOC1_PATH} | title: {DOC1_TITLE}"
)

#: The catalog-first document (catalog order = (source, path) —
#: DOC1 sorts first): the single-read flow reads THIS document, so it
#: must NOT be the seed (a seed read dedupes to "Already in your
#: context.", which the mock's single-read flow does not model — it
#: would loop to the round cap). Indexed WITHOUT chunks: catalog-only,
#: never in the retrieval context. Its FIRST line is longer than 80
#: chars, so the mock's first-80-chars quote (the single-read
#: regression turn) stays newline-free.
DOC1_CONTENT = (
    "The aws route53 hosted zone for reeselink keeps every record in "
    "reseelink.json — the exact JSON shape of reeselink.json is "
    "documented in the record file below.\n"
    + (
        "The aws route53 hosted zone for reeselink keeps every record in "
        "reseelink.json — the record file shape of reeselink.json is "
        "the contract every sync job relies on.\n"
    )
    * 10
    + "\n## Sync job\n\n"
    "A cron job pushes reeselink.json to the aws route53 hosted zone "
    "every fifteen minutes; the diff is applied through the route53 api.\n"
)
assert "\n" not in DOC1_CONTENT[:80]  # the quote must stay one line

#: The retrievable document (the grounded seed context, the cf.
#: test_harness_aligned_tools.py pattern): the repeated record-file
#: lines carry the trigger question's key tokens — well past the E2E
#: 0.30 cosine threshold, plus FTS hits. Its FIRST line is longer
#: than 80 chars too, so the retrieval seed context is one clean
#: line.
DOC2_CONTENT = (
    "The ReeseLink hosted zone record file reeselink.json holds every "
    "aws route53 record for reeselink — the note documents the exact "
    "JSON shape of reeselink.json for the record file.\n"
    + (
        "The aws route53 record file reeselink.json keeps every record "
        "for the reeselink hosted zone — the exact JSON shape of the "
        "record file is the contract every sync job relies on.\n"
    )
    * 10
    + "\n## Sync job\n\n"
    "A cron job pushes reeselink.json to the aws route53 hosted zone "
    "every fifteen minutes; the diff is applied through the route53 api.\n"
)
assert "\n" not in DOC2_CONTENT[:80]  # the seed context stays one line

#: Carries ``LS_TEACH_TRIGGER`` and is on-topic (grounded — HIGH, the
#: ``<tools>`` section rides along); it carries NO other mock marker.
LS_TEACH_QUESTION = (
    "List the files in this directory — what do my aws route53 notes "
    "say about the reeselink.json record file?"
)
assert LS_TEACH_TRIGGER in LS_TEACH_QUESTION.lower()
for _other in (
    "use your tools",
    "read two documents",
    "search your documents",
    "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 LS_TEACH_QUESTION.lower(), _other

#: Carries ``TOOLS_TRIGGER`` (the single-read flow) and nothing else —
#: the no-regression follow-up question in the same session.
READ_QUESTION = (
    "Use your tools: what is the exact JSON shape of reeselink.json "
    "for my aws route53 hosted zone?"
)
assert TOOLS_TRIGGER in READ_QUESTION.lower()
for _other in (
    LS_TEACH_TRIGGER,
    "read two documents",
    "search your documents",
    "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 READ_QUESTION.lower(), _other

#: The mock's single-read answer (the read document reached the model
#: and landed in the answer) — DOC1 is the first catalog line, so the
#: flow reads ``Homelab/aws-route53.md`` and quotes its first 80 chars.
READ_ANSWER_PREFIX = f"Read {DOC1_SP}."
READ_ANSWER_QUOTE = DOC1_CONTENT[:80]


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

    DOC1 (catalog-first) 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 and FTS-matches too →
    grounded). DOC2 is the seed context only — the single-read flow
    reads the catalog-FIRST document (DOC1), which is not in context.
    """
    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),
        )
    )
    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),
    )
    db.add(doc2)
    db.flush()
    # One chunk carrying the mock's own embedding → genuine token
    # overlap between the trigger question and DOC2 (the only
    # retrievable document).
    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_harness_aligned_tools.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")
    # 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."""
    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 LS-TEACH turn: the incident's ls(path='.') misuse →
#    the teaching refusal → the corrected no-arg ls() → the catalog
#    answer — the loop settles in ONE correction (two tool rounds),
#    pinned on the SSE wire
# --------------------------------------------------------------------------


def test_ls_misuse_self_corrects_to_noarg_listing(
    page: Page, app_url: str, mock_llm: int, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    _reset_db_fixture()
    page.goto(app_url)
    _install_sse_hook(page)

    _submit(page, LS_TEACH_QUESTION)
    _wait_settled(page)

    # Self-correction: the answer quotes the FIRST catalog line — the
    # first document's source: / path: / title fields reached the model
    # and landed in the answer (the catalog round settled the turn).
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text("These are the indexed documents:")
    expect(bubble).to_contain_text(FIRST_CATALOG_LINE)
    _assert_no_error_banner(page)

    # The UI shows the two tool lines in order: the scoped misuse
    # (🔎 Listing documents in <code>.</code>) then the corrected
    # unscoped listing (🔎 Listing documents — no <code>).
    lines = page.locator(".msg.brain .tool-call")
    expect(lines).to_have_count(2)
    expect(lines.nth(0)).to_contain_text("Listing documents in")
    expect(lines.nth(0).locator("code")).to_have_text(".")
    expect(lines.nth(1)).to_contain_text("Listing documents")
    expect(lines.nth(1).locator("code")).to_have_count(0)

    # Two rounds on the wire: the tool frames arrive in order — first
    # ls with argument "." (the incident's misuse), then ls with
    # argument null (the correction) — and there is NO third tool
    # frame: the loop ended in one correction, not at the round cap.
    frames = _drain_frames(page)
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "ls", "argument": "."},
        {"type": "tool", "name": "ls", "argument": None},
    ]
    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 flow — the SAME session: after the
#    LS-TEACH turn, the TOOLS_TRIGGER follow-up (the single-read flow)
#    still settles with the read flow's answer
# --------------------------------------------------------------------------


def test_plain_tool_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()
    page.goto(app_url)
    _install_sse_hook(page)

    # Turn 1 — the LS-TEACH flow (the incident's misuse → the
    # correction → the catalog answer).
    _submit(page, LS_TEACH_QUESTION)
    _wait_settled(page)
    teach_frames = _drain_frames(page)
    assert _tool_frames(teach_frames) == [
        {"type": "tool", "name": "ls", "argument": "."},
        {"type": "tool", "name": "ls", "argument": None},
    ]
    expect(
        page.locator(".msg.brain .bubble").last
    ).to_contain_text(FIRST_CATALOG_LINE)

    # Turn 2 — the SAME session: the single-read flow on
    # TOOLS_TRIGGER. The new flow must not have swallowed the existing
    # trigger: the follow-up settles with the read flow's answer.
    _submit(page, READ_QUESTION)
    _wait_settled(page)

    second_msg = page.locator(".msg.brain").last
    # The UI shows the single-read flow's two lines: the unscoped ls
    # then the read of the first catalog line's COMBINED identity.
    lines = second_msg.locator(".tool-call")
    expect(lines).to_have_count(2)
    expect(lines.nth(0)).to_contain_text("Listing documents")
    expect(lines.nth(0).locator("code")).to_have_count(0)
    expect(lines.nth(1)).to_contain_text("Reading ")
    expect(lines.nth(1).locator("code")).to_have_text(DOC1_SP)

    # The answer quotes the read document (the mock's deterministic
    # echo: "Read <source/path>. <first 80 chars>").
    bubble = second_msg.locator(".bubble").last
    expect(bubble).to_contain_text(READ_ANSWER_PREFIX)
    expect(bubble).to_contain_text(READ_ANSWER_QUOTE)
    _assert_no_error_banner(page)

    # Wire level for the follow-up: ls (null) → read (the combined
    # identity) — the single-read flow, unchanged.
    frames = _drain_frames(page)
    assert _tool_frames(frames) == [
        {"type": "tool", "name": "ls", "argument": None},
        {"type": "tool", "name": "read", "argument": DOC1_SP},
    ]
    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"]