scripts/playwright_llm_up.py

"""Browser validation for the LLM-UP runtime paths — a REAL successful turn.

The sibling harnesses (playwright_composer.py, playwright_c1c2.py) deliberately
exercise the LLM-DOWN error path (fast conn-refused). This one needs BOTH:

- the llama.cpp Gemma server reachable (real turns), and
- ``uvicorn loosecanvas.main:app`` on APP_PORT (default 8000),

and asserts the three things that could only be confirmed with Gemma running:

(a) Streaming + composer lifecycle — on Send the composer textarea + Send button
    DISABLE and ``.lc-turn-status`` shows "Thinking…"; tokens stream into the
    Assistant chatbot (the bot bubble grows); on success the composer RE-ENABLES,
    its text is CLEARED, and a non-empty answer is present (main.py on_turn).
(b) Build-from-text (P2-05 A4) — a SUCCESSFUL build clears the source textarea
    (main.py on_build_text success yield sets text_content to "").
(c) Last-turn spotlight (P2-06) — the nodes a real turn touches get the .flash
    Cytoscape class. .flash is a canvas class (not DOM), lives ~1800 ms, so a
    40 ms poller installed BEFORE navigation accumulates every flashed id into
    window.__flashSeen and the 1800 ms window is never missed. The assertion is
    grounded against the real graph delta: if the turn added/changed elements but
    nothing flashed, that is a real spotlight failure; if the turn touched no
    nodes at all (the model answered conversationally), that is recorded as a
    warning, not a failure, and a second, more directive turn is attempted.

    .venv/Scripts/python.exe scripts/playwright_llm_up.py
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path

from playwright.sync_api import Page, sync_playwright

BASE = f"http://127.0.0.1:{os.environ.get('APP_PORT', '8000')}/"
OUT = Path(__file__).resolve().parent.parent / ".playwright-out"
OUT.mkdir(exist_ok=True)
PLACEHOLDER = "Ask about the graph"
TURN_TIMEOUT_MS = 180_000
BUILD_TIMEOUT_MS = 240_000

# A turn whose answer needs a graph mutation: both nodes are visible in
# annotated_graph, so the model should call create_edge → add_element (edge) →
# both endpoints flash. The conversational answer streams in parallel.
TURN_PROMPT = (
    "How are 'learning rate' and 'gradient descent' related? "
    "Add an edge between them that captures the relationship."
)
# Fallback if the first turn touches no nodes: an explicit creation almost always
# emits add_element.
TURN_PROMPT_FALLBACK = (
    "Create a new concept node called 'Momentum' and connect it to "
    "'gradient descent' with an edge labelled 'accelerates'."
)
BUILD_TEXT = (
    "Photosynthesis is the process by which plants convert sunlight into chemical "
    "energy. Chlorophyll is the pigment that absorbs sunlight. The light reactions "
    "produce ATP. The Calvin cycle uses ATP to fix carbon dioxide into glucose. "
    "Glucose provides the energy plants need to grow."
)

# Installed before navigation: window.__cy is only exposed when __PLAYWRIGHT_TEST
# is set at mount time, and the 40 ms poller accumulates the transient signals
# (flash ids, composer-disabled, "Thinking…", growing bot text) so fast-LLM
# timing never loses them.
INIT_SCRIPT = r"""
window.__PLAYWRIGHT_TEST = true;
window.__flashSeen = [];
window.__composerDisabledSeen = false;
window.__thinkingSeen = false;
window.__maxBotLen = 0;
window.__botSamples = [];
(function () {
  function composer() {
    return [...document.querySelectorAll('textarea')].find(
      e => e.placeholder && e.placeholder.indexOf('Ask about the graph') === 0
    );
  }
  function botLen() {
    let best = 0;
    const sel = '.message.bot, [data-testid="bot"], .bot .message, .message-row.bot-row';
    document.querySelectorAll(sel).forEach(function (b) {
      const t = (b.textContent || '').trim();
      if (t.length > best) best = t.length;
    });
    return best;
  }
  setInterval(function () {
    try {
      const c = composer();
      if (c && c.disabled) window.__composerDisabledSeen = true;
      const st = document.querySelector('.lc-turn-status');
      if (st && /Thinking/.test(st.textContent || '')) window.__thinkingSeen = true;
      const bl = botLen();
      if (bl > window.__maxBotLen) { window.__maxBotLen = bl; window.__botSamples.push(bl); }
      if (window.__cy) {
        window.__cy.nodes('.flash').forEach(function (n) {
          const id = n.id();
          if (window.__flashSeen.indexOf(id) === -1) window.__flashSeen.push(id);
        });
      }
    } catch (e) { /* element churn between yields */ }
  }, 40);
})();
"""


def _element_ids(page: Page) -> list[str]:
    return page.evaluate(
        "() => window.__cy ? window.__cy.elements().map(e => e.id()) : []"
    )


def _reset_pollers(page: Page) -> None:
    page.evaluate(
        "() => { window.__flashSeen = []; window.__composerDisabledSeen = false; "
        "window.__thinkingSeen = false; window.__maxBotLen = 0; window.__botSamples = []; }"
    )


def _run_turn(page: Page, prompt: str) -> dict[str, object]:
    """Send one real turn, wait for success, return the captured signals."""
    box = page.get_by_placeholder(PLACEHOLDER)
    before_ids = set(_element_ids(page))
    _reset_pollers(page)
    box.fill(prompt)
    page.get_by_role("button", name="Send").click()

    errored = False
    try:
        # Success is the only state where the composer is enabled AND empty.
        page.wait_for_function(
            "() => { const c = [...document.querySelectorAll('textarea')]"
            ".find(e => e.placeholder && e.placeholder.indexOf('Ask about the graph') === 0);"
            " return c && !c.disabled && c.value === ''; }",
            timeout=TURN_TIMEOUT_MS,
        )
    except Exception:  # noqa: BLE001 - timed out waiting for success
        errored = True

    page.wait_for_timeout(
        400
    )  # let the 40 ms poller flush the tail of the flash window
    after_ids = set(_element_ids(page))
    return {
        "prompt": prompt,
        "errored_or_timeout": errored,
        "composer_disabled_seen": page.evaluate("() => window.__composerDisabledSeen"),
        "thinking_seen": page.evaluate("() => window.__thinkingSeen"),
        "max_bot_len": page.evaluate("() => window.__maxBotLen"),
        "bot_samples": page.evaluate("() => window.__botSamples"),
        "flash_seen": page.evaluate("() => window.__flashSeen"),
        "composer_value_after": box.input_value(),
        "composer_enabled_after": box.is_enabled(),
        "added_or_changed_ids": sorted(after_ids - before_ids),
        "node_count_after": page.evaluate(
            "() => window.__cy ? window.__cy.nodes().length : 0"
        ),
    }


def main() -> int:  # noqa: C901, PLR0912, PLR0915 - linear validation script
    console: list[str] = []
    report: dict[str, object] = {"base": BASE}
    failures: list[str] = []
    warnings: list[str] = []

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(viewport={"width": 1400, "height": 900})
        page.on("console", lambda m: console.append(f"{m.type}: {m.text}"))
        page.on("pageerror", lambda e: console.append(f"pageerror: {e}"))
        page.add_init_script(INIT_SCRIPT)

        page.goto(BASE, wait_until="domcontentloaded")
        page.wait_for_selector("#lc-canvas", timeout=30_000)
        page.get_by_placeholder(PLACEHOLDER).wait_for(timeout=15_000)

        # Load the fixture so cy + window.__cy exist and a turn has nodes to touch.
        page.get_by_role("button", name="Load graph").click()
        page.wait_for_function(
            "() => window.__cy && window.__cy.nodes().length > 0", timeout=20_000
        )

        # ── (a) + (c): a real turn ───────────────────────────────────────────
        turn = _run_turn(page, TURN_PROMPT)
        report["turn_1"] = turn
        page.screenshot(path=str(OUT / "llm_up_turn1.png"))

        # If the model answered conversationally (touched no nodes), the spotlight
        # has nothing to fire on — retry once with an explicit creation prompt so
        # path (c) is genuinely exercised.
        if not turn["flash_seen"] and not turn["added_or_changed_ids"]:
            warnings.append(
                "turn 1 touched no nodes (model answered conversationally); "
                "retrying with an explicit creation prompt to exercise the spotlight"
            )
            turn2 = _run_turn(page, TURN_PROMPT_FALLBACK)
            report["turn_2"] = turn2
            page.screenshot(path=str(OUT / "llm_up_turn2.png"))
            spotlight_turn = turn2
        else:
            spotlight_turn = turn

        # (a) streaming + composer lifecycle — assert on the FIRST turn.
        if turn["errored_or_timeout"]:
            failures.append(
                "(a) turn did not reach success (composer never re-enabled+cleared); "
                f"composer_value_after={turn['composer_value_after']!r}"
            )
        if not turn["composer_disabled_seen"]:
            failures.append("(a) composer was never observed disabled during the turn")
        if turn["composer_value_after"] != "":
            failures.append(
                f"(a) composer not cleared after success (={turn['composer_value_after']!r})"
            )
        if not turn["composer_enabled_after"]:
            failures.append("(a) composer left disabled after the turn")
        if not turn["max_bot_len"] or int(turn["max_bot_len"]) <= 0:  # type: ignore[arg-type]
            warnings.append(
                "(a) did not observe a growing Assistant bubble via DOM "
                "(selector miss?); relying on contract-exact composer clear instead"
            )
        if not turn["thinking_seen"]:
            warnings.append("(a) did not catch the 'Thinking…' status (fast LLM)")

        # (c) spotlight — grounded against the real graph delta.
        flash = spotlight_turn["flash_seen"]
        changed = spotlight_turn["added_or_changed_ids"]
        report["spotlight_flash_ids"] = flash
        report["spotlight_changed_ids"] = changed
        if flash:
            pass  # spotlight fired
        elif changed:
            failures.append(
                f"(c) turn added/changed elements {changed} but NOTHING flashed — "
                "spotlight (P2-06) is broken"
            )
        else:
            warnings.append(
                "(c) spotlight not exercised: even the fallback turn touched no nodes "
                "(model behaviour, not a code defect) — rerun to confirm"
            )

        # ── (b) build-from-text clears the source textarea on success ─────────
        try:
            page.get_by_text("Build from text", exact=False).first.click()
            page.wait_for_timeout(300)
            content = page.locator('textarea[placeholder*="Paste prose"]')
            content.wait_for(timeout=5_000)
            content.fill(BUILD_TEXT)
            nodes_before_build = page.evaluate(
                "() => window.__cy ? window.__cy.nodes().length : -1"
            )
            page.get_by_role("button", name="Build graph from text").click()
            build_ok = True
            try:
                page.wait_for_function(
                    "() => { const t = document.querySelector("
                    "'textarea[placeholder*=\"Paste prose\"]');"
                    " return t && t.value === ''; }",
                    timeout=BUILD_TIMEOUT_MS,
                )
            except Exception:  # noqa: BLE001
                build_ok = False
            page.wait_for_timeout(500)
            content_after = content.input_value()
            nodes_after_build = page.evaluate(
                "() => window.__cy ? window.__cy.nodes().length : -1"
            )
            status_after = (
                page.locator(".lc-turn-status").first.inner_text() or ""
            ).strip()
            report["build"] = {
                "ok": build_ok,
                "content_after": content_after,
                "nodes_before_build": nodes_before_build,
                "nodes_after_build": nodes_after_build,
                "status_after": status_after,
            }
            page.screenshot(path=str(OUT / "llm_up_build.png"))
            if not build_ok or content_after != "":
                failures.append(
                    f"(b) successful build did not clear the source textarea "
                    f"(content_after={content_after!r}, status={status_after!r})"
                )
            elif nodes_after_build <= 0:
                warnings.append(
                    "(b) textarea cleared but canvas shows no nodes after build "
                    f"(nodes_after={nodes_after_build}) — verify the build succeeded"
                )
        except Exception as exc:  # noqa: BLE001
            failures.append(f"(b) build-from-text path crashed: {exc}")

        browser.close()

    report["console_errors"] = [
        c for c in console if c.startswith(("error:", "pageerror:"))
    ][-15:]
    report["warnings"] = warnings
    report["failures"] = failures
    report["ok"] = not failures
    (OUT / "llm_up_report.json").write_text(
        json.dumps(report, indent=2), encoding="utf-8"
    )
    print(json.dumps(report, indent=2))
    return 0 if not failures else 1


if __name__ == "__main__":
    sys.exit(main())