"""Phase 71 E2E (Playwright, mock-only): tool-scaffolding guardrails —
the deterministic strip + one bounded recovery, through the real UI.

Owner request (chat, 2026-09-03): the same incident as phase 70 — a
deflected round streamed the model's raw
``<|tool_call_start|>[read(path='…')]<|tool_call_end|>`` chat-template
markup into the UI although no tools were offered. The guardrail is
DETERMINISTIC ONLY (no model in detection or repair): a streaming
filter strips known scaffolding from ``delta.content`` server-side
(``app/rag/scaffolding.py``), and a reply whose visible content ends
up empty gets exactly ONE bounded recovery (``tools=None``, the
harness correction folded into the system prompt); a second empty
reply settles with the dedicated error frame.

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

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

MOCK-ONLY suite: ``E2E_REAL_LLM=1`` is not supported — the gate is the
deterministic scaffolding flows in ``tests/e2e/mock_llm.py`` (phase
71), which are independent of the ``<tools>`` marker:

* ``emit raw tool markup`` (``SCAFFOLD_TRIGGER``): request 1 streams
  ONLY the incident span as ``delta.content`` (split across the
  mock's 12-char chunks — the filter's boundary path), no structured
  ``tool_calls``, no reasoning; the follow-up request carrying
  ``CORRECTION_INSTRUCTION`` in the system prompt (the one bounded
  recovery) streams the clean ``SCAFFOLD_RECOVERY_ANSWER``.
* ``always emit raw tool markup`` (``SCAFFOLD_ALWAYS_TRIGGER``): the
  scaffolding-only span on EVERY request (the recovery included) —
  the terminal malformed-reply path.

Both triggers run on an EMPTY knowledge base: with zero chunks the
honesty gate is LOW (cosine 0.0 < the E2E 0.30 threshold, no FTS
hits), so every turn takes the DEFLECTED path — the exact path the
2026-09-03 incident hit — where the filter + recovery live in
``app/api/chat.py``.

Test → phase mapping (Playwright Mapping Rule):
1. ``test_recovery_strips_scaffolding_and_streams_clean_answer`` — the
   recovery case: the turn settles (the composer re-enables, ``done``
   on the wire), the final answer bubble carries the recovery's clean
   text, ``document.body.innerText`` contains NEITHER
   ``tool_call_start`` nor ``tool_call_end`` (nor the raw
   ``[read(path=…`` fragment), and no error banner — and wire-level,
   no ``delta`` frame ever carries a scaffolding fragment (the strip
   happens server-side, not in the UI).
2. ``test_terminal_scaffolding_lands_on_dedicated_error_app_stays_usable``
   — the terminal case: the existing error state renders with the
   dedicated copy ("The model returned a malformed reply — please try
   again."), no raw tokens in the DOM, no answer bubble, no ``done``
   and NO ``query_log`` row (the existing terminal-error semantics) —
   and the app stays usable: a follow-up plain question in the same
   session gets a normal deflected answer and the banner clears.
3. ``test_plain_turn_never_recovers_and_streams_byte_clean`` — no
   false positive: a plain deflected question streams its
   first-request answer byte-clean (the concatenated delta text is
   EXACTLY the mock's deterministic deflection answer — not the
   recovery's), with no error state and no recovery request visible
   (the turn settles on the first request).
"""
from __future__ import annotations

import json
import re
import time

from playwright.sync_api import Page, expect
from sqlalchemy import select, text

from app.db import SessionLocal
from app.models import QueryLog
from tests.e2e.mock_llm import SCAFFOLD_RECOVERY_ANSWER, compose_answer

# --------------------------------------------------------------------------
# Questions + the mock's deterministic expectations
# --------------------------------------------------------------------------

#: Carries ``SCAFFOLD_TRIGGER`` (and nothing else — the module-level
#: asserts below pin the trigger exclusions).
RECOVERY_Q = "Please emit raw tool markup in your reply to this."

#: Carries ``SCAFFOLD_ALWAYS_TRIGGER`` — the scaffolding-only span on
#: EVERY request, recovery included (the terminal path).
TERMINAL_Q = "Please always emit raw tool markup in every reply."

#: A plain question (the phase-67 deflection shape): no trigger at all.
PLAIN_Q = "How do I bake sourdough bread?"

for _q in (RECOVERY_Q, TERMINAL_Q, PLAIN_Q):
    for _other in (
        "read two documents",
        "search your 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",
        "use your tools",
    ):
        assert _other not in _q.lower(), f"{_other!r} unexpectedly in {_q!r}"
assert "emit raw tool markup" in RECOVERY_Q.lower()
assert "always emit raw tool markup" not in RECOVERY_Q.lower()
assert "always emit raw tool markup" in TERMINAL_Q.lower()
assert "emit raw tool markup" not in PLAIN_Q.lower()

#: The terminal malformed-reply copy (``app.api.chat`` — the phase-71
#: dedicated error frame; the mock's ALWAYS trigger is what makes the
#: recovery come back empty).
MALFORMED_ERROR_COPY = "The model returned a malformed reply — please try again."

#: The mock's deterministic deflection phrase (the phase-67 pin).
DEFLECT_PHRASE = r"haven't done anything like that"

#: The raw scaffolding fragments that must NEVER reach the user — the
#: span's tokens and the incident's argument fragment (``SCAFFOLD_SPAN``
#: in the mock).
RAW_TOKENS = ("tool_call_start", "tool_call_end", "[read(path=", "<|")


def _expected_deflect_answer(question: str) -> str:
    """The mock's deterministic DEFLECT_MODE answer for *question*.

    Derived from the mock itself (``compose_answer`` on a synthetic
    body: the mode marker in the system prompt, the question as the
    user message, no tuning/KB sections — the suite truncates
    ``steering_notes`` / ``kb_overview``, so the real prompt carries
    none), so the byte-clean pin can never drift from the mock.
    """
    return compose_answer(
        {
            "messages": [
                {"role": "system", "content": "DEFLECT_MODE marker"},
                {"role": "user", "content": question},
            ]
        }
    )


# --------------------------------------------------------------------------
# DB reset (EMPTY knowledge base → every turn is deterministically
# deflected: the LOW gate, the incident's path) + query_log reads
# --------------------------------------------------------------------------


def _reset_db_empty() -> None:
    """Truncate the KB (plus the prompt-shaping tables): an EMPTY
    knowledge base, so the honesty gate is LOW for every question
    (no chunks → cosine 0.0 < 0.30, fts_hits 0) — the deflected path
    where the phase-71 filter + recovery live, and the prompts stay
    byte-stable regardless of leftovers from other suites."""
    with SessionLocal() as db:
        db.execute(
            text("TRUNCATE chunks, documents, query_log, steering_notes, kb_overview")
        )
        db.commit()


def _query_log_rows() -> list[QueryLog]:
    with SessionLocal() as db:
        return list(db.scalars(select(QueryLog)).all())


# --------------------------------------------------------------------------
# Page hooks (the SSE capture — the house pattern from
# test_agent_document_tools.py / test_llm_retry.py)
# --------------------------------------------------------------------------

#: Captures the raw SSE ``data:`` payloads of the /api/chat stream
#: (a response clone read in the background) — wire-level assertions
#: 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, terminal: str = "done") -> list[dict]:
    """One turn's SSE frames: wait for that turn's *terminal* frame
    (``done`` — or ``error`` for the terminal case), 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: the terminal frame is
    observed only in the batch that carries the turn's full frame
    sequence."""
    deadline = time.monotonic() + 30.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") == terminal for f in parsed):
            return parsed
        if time.monotonic() > deadline:
            raise AssertionError(
                f"SSE hook captured no `{terminal}` frame (frames so far: "
                f"{len(parsed)}) — hook install failed?"
            )
        time.sleep(0.05)


def _delta_text(frames: list[dict]) -> str:
    """The turn's answer text exactly as the wire carried it."""
    return "".join(f.get("text", "") for f in frames if f.get("type") == "delta")


def _assert_no_raw_tokens_on_wire(frames: list[dict]) -> None:
    for frame in frames:
        if frame.get("type") != "delta":
            continue
        text = frame.get("text", "")
        for raw in RAW_TOKENS:
            assert raw not in text, f"raw scaffolding {raw!r} on the wire: {text!r}"


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_raw_tokens(page: Page) -> None:
    """The user-visible contract: no raw scaffolding fragment anywhere
    in the rendered page (``document.body.innerText``)."""
    body = page.locator("body").inner_text()
    for raw in RAW_TOKENS:
        assert raw not in body, f"raw scaffolding {raw!r} leaked into the DOM"


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 recovery case: the span is stripped server-side, the ONE
#    bounded recovery answers, and no raw token ever reaches the DOM
# --------------------------------------------------------------------------


def test_recovery_strips_scaffolding_and_streams_clean_answer(
    page: Page, app_url: str, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    _reset_db_empty()
    page.goto(app_url)
    _install_sse_hook(page)

    _submit(page, RECOVERY_Q)
    _wait_settled(page)

    # The final answer bubble carries the RECOVERY's clean text — the
    # mock's deterministic recovery answer, proof the one bounded
    # recovery ran (request 2, with the correction in the system
    # prompt) and its answer is what the user saw.
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(SCAFFOLD_RECOVERY_ANSWER)
    # The deflected message state is intact — the turn took the LOW
    # path, the incident's path.
    expect(page.locator(".msg.brain").last).to_have_class(re.compile(r"is-deflected"))
    _assert_no_raw_tokens(page)
    _assert_no_error_banner(page)

    # Wire level: the strip happens SERVER-side — no `delta` frame ever
    # carries a scaffolding fragment, and the concatenated delta text is
    # EXACTLY the recovery answer (request 1's span produced zero delta
    # frames). No tool frames, no retry frames, no error frame; the
    # turn settled with `done` (deflected).
    frames = _drain_frames(page, terminal="done")
    assert [f for f in frames if f.get("type") == "delta"], (
        "the recovery answer must have streamed delta frames"
    )
    _assert_no_raw_tokens_on_wire(frames)
    assert _delta_text(frames) == SCAFFOLD_RECOVERY_ANSWER
    assert not [f for f in frames if f.get("type") == "tool"]
    assert not [f for f in frames if f.get("type") == "retry"]
    assert not [f for f in frames if f.get("type") == "error"]
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is True

    # The recovered turn settles durably: exactly one query_log row
    # (deflected — the incident's path).
    rows = _query_log_rows()
    assert len(rows) == 1, rows
    assert rows[0].question == RECOVERY_Q
    assert rows[0].deflected is True


# --------------------------------------------------------------------------
# 2. The terminal case: scaffolding twice → the dedicated error frame,
#    no done, no query_log row — and the app stays usable
# --------------------------------------------------------------------------


def test_terminal_scaffolding_lands_on_dedicated_error_app_stays_usable(
    page: Page, app_url: str, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    _reset_db_empty()
    page.goto(app_url)
    _install_sse_hook(page)

    _submit(page, TERMINAL_Q)

    # BOTH requests (original + the one bounded recovery) came back
    # scaffolding-only: no clean content ever streamed, so the turn
    # settles with the EXISTING terminal error state — the banner
    # (role=alert) with the DEDICATED malformed-reply copy — and the
    # send button re-enabled (the banner path settles the state
    # machine, cf. the phase-67 exhaustion test).
    expect(page.locator("#kb-banner")).to_have_attribute(
        "role", "alert", timeout=60_000
    )
    expect(page.locator("#kb-banner")).to_contain_text(MALFORMED_ERROR_COPY)
    expect(page.locator("#send-btn")).to_be_enabled(timeout=30_000)
    expect(page.locator("#send-label")).to_have_text("Send", timeout=30_000)

    # No answer bubble was ever rendered (every streamed frame was
    # stripped server-side) and no raw token is anywhere in the DOM.
    expect(page.locator("#messages > .msg.brain")).to_have_count(0)
    _assert_no_raw_tokens(page)

    # Wire: the terminal error frame is LAST — no done, and NO delta
    # frame at all (both requests' content was pure scaffolding).
    frames = _drain_frames(page, terminal="error")
    assert frames[-1]["type"] == "error"
    assert MALFORMED_ERROR_COPY in frames[-1]["detail"]
    assert not [f for f in frames if f.get("type") == "done"]
    assert not [f for f in frames if f.get("type") == "delta"]

    # Terminal semantics (byte-for-byte the existing LLMError shape):
    # the turn writes no query_log row.
    assert _query_log_rows() == []

    # The app stays usable: a follow-up plain question (no trigger) in
    # the SAME session gets a normal streamed deflected answer, the
    # banner clears, and the wire is clean.
    _submit(page, PLAIN_Q)
    _wait_settled(page)
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(re.compile(DEFLECT_PHRASE, re.IGNORECASE))
    _assert_no_error_banner(page)
    _assert_no_raw_tokens(page)
    follow_frames = _drain_frames(page, terminal="done")
    assert next(f for f in follow_frames if f["type"] == "done")["deflected"] is True
    assert not [f for f in follow_frames if f.get("type") == "error"]
    _assert_no_raw_tokens_on_wire(follow_frames)


# --------------------------------------------------------------------------
# 3. No false positive: a plain turn streams its first-request answer
#    byte-clean — no strip, no recovery request, no error state
# --------------------------------------------------------------------------


def test_plain_turn_never_recovers_and_streams_byte_clean(
    page: Page, app_url: str, db_ready: None
) -> None:
    page.set_default_timeout(30_000)
    _reset_db_empty()
    page.goto(app_url)
    _install_sse_hook(page)

    _submit(page, PLAIN_Q)
    _wait_settled(page)

    # Wire level: the plain deflected answer streams byte-clean from
    # the FIRST request — the concatenated delta text is EXACTLY the
    # mock's deterministic deflection answer (derived from the mock
    # itself above) and NOT the recovery answer (a recovery request
    # would have streamed that text instead).
    frames = _drain_frames(page, terminal="done")
    delta_text = _delta_text(frames)
    assert delta_text == _expected_deflect_answer(PLAIN_Q)
    assert SCAFFOLD_RECOVERY_ANSWER not in delta_text
    _assert_no_raw_tokens_on_wire(frames)
    # No recovery request is visible: no error frame, no retry frame
    # (the recovery is not a phase-67 retry), and the turn settled on
    # the first request with `done`.
    assert not [f for f in frames if f.get("type") == "error"]
    assert not [f for f in frames if f.get("type") == "retry"]
    done = next(f for f in frames if f.get("type") == "done")
    assert done["deflected"] is True

    # UI level: the deflected answer rendered, no error banner, no raw
    # token in the DOM.
    bubble = page.locator(".msg.brain .bubble").last
    expect(bubble).to_contain_text(re.compile(DEFLECT_PHRASE, re.IGNORECASE))
    _assert_no_error_banner(page)
    _assert_no_raw_tokens(page)