"""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())