py/tests/robot_streamlit.py
"""Robot-UI drive against the LIVE Streamlit demo on :8501, RECORDED to saga.
This is the last mile the other two robots do not cover:
* `pkg-facet-wrapped/tests/robot_installed.py` drives the installed wheel's loader
page directly — it proves the ARTIFACT works, host-agnostically.
* `py/tests/test_app.py` spies each pane's `facett_component` mount — it proves the
PAYLOAD reaching the component is real, live server data.
* this file drives the RUNNING APP in a real browser and reads the state each
component publishes from inside its Streamlit iframe — it proves the whole chain
actually renders: korp-server → pane → Streamlit's `declare_component` → the
`streamlit:componentReady` / `RENDER_EVENT` handshake → the wasm.
That handshake is the reason this robot exists. The wheel's loader speaks BOTH a
host-agnostic vocabulary (`?component=&data=`) and Streamlit's postMessage protocol;
only a real Streamlit page exercises the second one, and a regression there is
invisible to every other test — the panes still hand over correct payloads, the app
still runs, and every canvas is blank.
Every mount is recorded to the saga tape and asserted from the read-back, so a run
leaves a durable, queryable record of what the live demo showed.
Usage (the app must already be serving)::
python tests/robot_streamlit.py [--url http://127.0.0.1:8501] [--no-saga]
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from pathlib import Path
from playwright.sync_api import sync_playwright
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from facett_demo.saga import SagaRecorder # noqa: E402
#: Components the live demo MUST show, and the minimum each must be carrying. A
#: mount that renders is not enough — an empty table renders perfectly.
EXPECTED: dict = {
# The map must be navigable AND actually drawing map data — a framed, empty
# Sweden is the exact failure this demo has had before.
"osm2d": lambda s: s.get("navigable") is True
and (s.get("drawn_points", 0) > 0 or s.get("visible_ways", 0) > 0)
and sum(s.get("lod_sizes", {}).values()) > 0,
# The graph must have laid out real nodes on its canvas.
"graphnav": lambda s: len(s.get("canvas", {}).get("nodes", [])) > 0,
"table": lambda s: isinstance(s.get("rows"), int) and s["rows"] > 0,
"stat-grid": lambda s: isinstance(s.get("rows"), int) and s["rows"] > 0,
}
#: The app mounts this many components on a first load (map + graph + 8 population
#: surfaces + 2 error matrices). Fewer means a pane silently rendered nothing.
MIN_MOUNTS = 12
def _component_frames(pg) -> list:
"""The component iframes in DOM order (the order the panes are laid out).
`page.frames` is CREATION order and keeps frames Streamlit has already detached
on a rerun, so it yields a scrambled, sometimes stale view. Going through the
live `<iframe>` elements gives the page as it actually is right now.
"""
out = []
for el in pg.query_selector_all("iframe"):
fr = el.content_frame()
if fr is not None:
out.append(fr)
return out
def _mounted_names(frames) -> list[str]:
"""What each frame currently reports as mounted ('' when not yet).
Tolerates frames that are mid-navigation (Streamlit swaps them on every rerun):
an unreadable frame is simply "not mounted yet", not an error.
"""
names = []
for fr in frames:
try:
names.append(fr.evaluate(
"() => typeof window.facet_mounted === 'function' ? window.facet_mounted() : ''"
))
except Exception:
names.append("")
return names
#: The graph and map groups of the gallery: EVERY member is mounted by `--gallery` and
#: judged on its state, and on its GPU lane when `--headful`. Owner ruling 2026-09-28:
#: a GPU always exists (an integrated one counts, SwiftShader/llvmpipe do not), so a
#: graph or map component that paints on the CPU in a headful Chrome is a defect.
#: The app's top tabs are facett-demo's groups; each opens on its `🎨 Komponenter` sub-tab.
GALLERY_GROUPS = ("🕸 Grafer", "🗺 Kartor")
COMPONENTS_TAB = "\U0001F3A8 Komponenter"
def _lane_verdict(name: str, state: dict, backend: str, adapter: dict) -> tuple[bool, str]:
"""(ok, why) for one mounted graph/map component, from what it PUBLISHES.
Three readings, all from the component's own state and the canvas' own adapter:
the canvas is WebGPU on a hardware adapter; no lane flag says `cpu`; no GPU
layout says it fell back. A component that publishes no lane at all is recorded
(`lane=?`) — it is drawn by egui through the same WebGPU canvas.
"""
if backend != "webgpu":
return False, f"backend={backend!r}"
if adapter.get("software") is True:
return False, f"software adapter {adapter.get('name')!r} (not a hardware GPU)"
blob = json.dumps(state)
flag = state.get("lane_flag")
if flag == "cpu":
return False, "lane_flag=cpu behind a WebGPU canvas"
cv = state.get("canvas") if isinstance(state.get("canvas"), dict) else {}
if cv.get("layout_fallback"):
return False, f"GPU layout fell back: {cv['layout_fallback']}"
if '"gpu_render_cpu_fallback": true' in blob or '"gpu_render_cpu_fallback":true' in blob:
return False, "gpu_render_cpu_fallback=true"
lanes = [f"lane_flag={flag}"] if flag else []
if cv.get("layout_lane"):
lanes.append(f"layout={cv['layout_lane']}")
return True, " ".join(lanes) or "lane=?"
def gallery_drive(pg, args, check, saga) -> None:
"""Mount every graph and map component in the Gallery, one at a time, and judge each.
`one at a time` mode keeps one canvas per group tab alive (five WebGPU devices at
most), which a laptop GPU survives; `the whole group` would be forty.
"""
pg.wait_for_timeout(4000)
if args.dev:
# The sidebar toggle: every canvas mounted after this shows its navgraph.
pg.get_by_text("dev mode · navgraph numbers").first.click()
pg.wait_for_timeout(3000)
for group in GALLERY_GROUPS:
pg.get_by_role("tab", name=group).first.click()
pg.wait_for_timeout(1500)
pg.get_by_role("tab", name=COMPONENTS_TAB).first.click()
pg.wait_for_timeout(1500)
panel = pg.locator('[role="tabpanel"]:visible').last
select = panel.locator('div[data-baseweb="select"]').first
select.click()
names = [o.inner_text().split(" (")[0].strip() for o in pg.get_by_role("option").all()]
pg.keyboard.press("Escape")
for name in names:
select.click()
pg.get_by_role("option", name=name).first.click()
state, err, backend, adapter, nav = None, "", "", {}, None
for _ in range(60):
pg.wait_for_timeout(1000)
el = panel.locator("iframe").first.element_handle()
fr = el.content_frame() if el else None
if fr is None:
continue
try:
if fr.evaluate("() => typeof window.facet_mounted === 'function'") and \
fr.evaluate("() => window.facet_mounted()") == name:
raw = fr.evaluate("() => window.facet_state()")
state = json.loads(raw) if raw and raw != "null" else None
err = fr.evaluate("() => window.facet_error()")
backend = fr.evaluate("() => window.facet_backend ? window.facet_backend() : ''")
ad = fr.evaluate("() => window.facet_adapter ? window.facet_adapter() : '{}'")
adapter = json.loads(ad) if ad else {}
if args.dev:
nraw = fr.evaluate("() => window.facet_navgraph ? window.facet_navgraph() : 'null'")
nav = json.loads(nraw) if nraw else None
if state is not None and (not args.dev or (nav and nav.get("count", 0) > 0)):
break
except Exception:
continue
ok = state is not None and not err
why = f"err={err!r}" if err else ("" if state is not None else "never mounted")
if ok and args.headful:
ok, why = _lane_verdict(name, state, backend, adapter)
elif ok:
why = f"backend={backend} (headless: lanes recorded, not judged)"
if ok and args.dev:
nodes = {n["id"]: n["number"] for n in (nav or {}).get("nodes", [])}
spine = nodes.get("facet-wrapped") == "1" and nodes.get(f"facet-wrapped.{name}") == "1.1"
ok = ok and spine
why += f" nav={len(nodes)} nodes" + ("" if spine else " (spine 1 / 1.1 missing)")
check(ok, f"gallery:{name}", f"{why} adapter={adapter.get('name', '')!r}")
if saga:
saga.mounted(name, json.dumps(state) if state else "", action="gallery-mount",
detail=f"group={group} assert={'pass' if ok else 'fail'}")
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--url", default="http://127.0.0.1:8501")
ap.add_argument("--timeout", type=int, default=180,
help="seconds to wait for every expected component to mount")
ap.add_argument("--no-saga", action="store_true", help="drive without recording")
# HEADFUL, on a real GPU. Headless Chrome offers only SwiftShader, cannot present a
# WebGPU canvas and rejects wgpu readbacks, so it reports working GPU lanes as
# broken (facet-wrapped-wheel.md row 13). The only truthful vehicle for the lane
# readouts below is a windowed Chrome on a display with a hardware adapter:
# WAYLAND_DISPLAY=wayland-0 python tests/robot_streamlit.py --headful \
# --chrome /usr/bin/google-chrome-stable
ap.add_argument("--headful", action="store_true",
help="windowed Chrome with WebGPU on; asserts the GPU lanes")
ap.add_argument("--chrome", default=None,
help="path to a Chrome/Chromium binary (default: playwright's chromium)")
ap.add_argument("--tab", default=None,
help="click this Streamlit tab first (e.g. Gallery: every component, "
"synthetic data, no korp-server needed)")
ap.add_argument("--expect", default=None,
help="comma-separated component names to require instead of EXPECTED")
ap.add_argument("--gallery", action="store_true",
help="mount EVERY graph and map component in the Gallery and judge each "
"one's state (and GPU lane with --headful), instead of the live panes")
ap.add_argument("--dev", action="store_true",
help="with --gallery: turn on dev mode and require each canvas' navgraph "
"(facet-wrapped = 1, the component = 1.1)")
args = ap.parse_args()
if args.expect is not None:
EXPECTED.clear()
EXPECTED.update({n: (lambda s: True) for n in args.expect.split(",") if n})
saga = None
if not args.no_saga:
saga = SagaRecorder(app="facet-wrapped-robot", strict=True)
saga.record("nav", route=args.url, action="goto", detail="live streamlit demo")
print(f"saga session {saga.session_id} → {saga.target} [{saga.workspace}]")
results: list[dict] = []
def check(ok: bool, label: str, extra: str = "") -> None:
results.append({"label": label, "ok": bool(ok), "detail": extra})
print(f"{'PASS' if ok else 'FAIL'} {label:26} {extra}")
with sync_playwright() as p:
if args.headful:
launch_args = ["--enable-unsafe-webgpu", "--enable-features=Vulkan",
"--no-first-run", "--no-default-browser-check"]
if os.environ.get("WAYLAND_DISPLAY"):
launch_args.append("--ozone-platform=wayland")
b = p.chromium.launch(headless=False, executable_path=args.chrome, args=launch_args)
else:
b = p.chromium.launch(executable_path=args.chrome,
args=["--enable-unsafe-swiftshader", "--use-gl=swiftshader"])
pg = b.new_page(viewport={"width": 1500, "height": 1000})
pg.goto(args.url, wait_until="load", timeout=90000)
if args.gallery:
gallery_drive(pg, args, check, saga)
else:
if args.tab:
pg.wait_for_timeout(4000)
pg.get_by_role("tab", name=args.tab).first.click()
# WAIT for the app to finish mounting rather than sleeping a fixed time, and
# require the picture to be STABLE before reading it.
#
# Two distinct races made a fixed settle lie. The map pane is by far the
# slowest (korp streams tens of thousands of ways), so an early read shows it
# missing; and Streamlit tears down and recreates component iframes on every
# rerun, so a read taken mid-rerun sees only some of them. Both look exactly
# like "a pane rendered nothing". So: poll until every expected component is
# mounted AND the (count, names) picture repeats unchanged — then read.
waited, prev, last = 0.0, None, []
while waited < args.timeout:
frames = _component_frames(pg)
names = _mounted_names(frames)
now = (len(frames), sorted(names))
settled = (
set(EXPECTED) <= set(names) and len(frames) >= MIN_MOUNTS and now == prev
)
last = names
if settled:
break
prev = now
pg.wait_for_timeout(2000)
waited += 2.0
missing = sorted(set(EXPECTED) - set(last))
why = ""
if missing:
# A pane that mounted nothing said WHY on the page (the honest error /
# empty panels this demo is built around). Quote it, so a failure names
# its upstream cause instead of just "component absent" — most often the
# korp-server being restarted under the drive, which is not a facett bug
# but must never be silently retried into a green run either.
body = pg.inner_text("body")
why = " | ".join(
ln.strip() for ln in body.splitlines()
if any(k in ln for k in ("failed", "unavailable", "unreachable",
"UNAVAILABLE", "honest empty", "no geometry"))
)[:300]
check(not missing and waited < args.timeout, "all-mounted-and-stable",
f"{waited:.0f}s, {len(last)} frames, missing={missing}"
+ (f" — app said: {why}" if why else ""))
check(pg.title() == "facett-demo", "page-title", pg.title())
frames = _component_frames(pg)
check(len(frames) >= MIN_MOUNTS, "component-iframes", f"{len(frames)} (min {MIN_MOUNTS})")
seen: dict[str, int] = {}
for i, fr in enumerate(frames):
try:
api = fr.evaluate("() => typeof window.facet_mounted === 'function'")
except Exception as e: # a frame that navigated away mid-read
check(False, f"frame[{i}]-readable", str(e)[:100])
continue
if not api:
check(False, f"frame[{i}]-loader", "index.html did not install the facet API")
continue
name = fr.evaluate("() => window.facet_mounted()")
err = fr.evaluate("() => window.facet_error()")
raw = fr.evaluate("() => window.facet_state()")
state = json.loads(raw) if raw and raw != "null" else None
ok = bool(name) and not err and state is not None
if ok and name in EXPECTED:
try:
ok = bool(EXPECTED[name](state))
except Exception as e:
ok, err = False, f"assertion error: {e}"
seen[name] = seen.get(name, 0) + 1
# THE GPU LANE READOUTS — the reason the headful mode exists. Every
# component publishes its lane; a canvas that painted on the CPU painter
# behind a WebGPU context is the select-without-install defect, and a
# graph layout that fell back names why. Headless cannot judge these
# (it has no hardware adapter), so they are recorded there and asserted
# only when --headful.
lane = ""
if isinstance(state, dict):
api = state.get("facet_canvas_api") or state.get("canvas_api") or ""
flag = state.get("lane_flag")
cv = state.get("canvas") if isinstance(state.get("canvas"), dict) else {}
llane, lfb = cv.get("layout_lane"), cv.get("layout_fallback")
bits = [f"api={api}" if api else "", f"lane={flag}" if flag else "",
f"layout={llane}" if llane else "", f"fallback={lfb!r}" if lfb else ""]
lane = " ".join(b for b in bits if b)
if args.headful:
if api and api != "webgpu":
ok, err = False, f"canvas api is {api!r}, not webgpu"
if flag == "cpu":
ok, err = False, "lane_flag=cpu behind a WebGPU canvas (renderer not installed?)"
if lfb:
ok, err = False, f"GPU layout fell back: {lfb}"
check(ok, f"frame[{i}]:{name or '(unmounted)'}", (lane + " " if lane else "") +
(f"err={err!r} " if err else "") + (json.dumps(state)[:80] if state else ""))
if saga:
saga.mounted(name or "(unmounted)", raw or "", action="streamlit-mount",
detail=f"frame={i} assert={'pass' if ok else 'fail'}")
if err:
saga.record("error", component=name, action="streamlit-mount",
detail=str(err), level="error")
for want in EXPECTED:
check(want in seen, f"present:{want}", f"x{seen.get(want, 0)}")
b.close()
if saga:
print("\n-- saga tape --")
saga.record("log", action="finish",
detail=f"{sum(1 for r in results if r['ok'])}/{len(results)} checks ok")
captured = saga.status()["captured"]
accepted = saga.flush()
tape = saga.read_session()
check(accepted == captured, "saga:accepted-all", f"{accepted}/{captured}")
check(len(tape) == captured, "saga:readback-count", f"read {len(tape)}")
action = "gallery-mount" if args.gallery else "streamlit-mount"
min_mounts = 1 if args.gallery else MIN_MOUNTS
mounts_on_tape = [e for e in tape if e["action"] == action and e["kind"] == "state"]
check(len(mounts_on_tape) >= min_mounts, "saga:mounts-recorded",
f"{len(mounts_on_tape)} of >= {min_mounts}")
check(all(e["state_json"] and json.loads(e["state_json"]) is not None for e in mounts_on_tape),
"saga:state-json-parses")
print(f" session_id={saga.session_id}")
saga.close()
else:
print("\n-- saga tape -- SKIPPED (--no-saga): nothing was recorded")
passed = sum(1 for r in results if r["ok"])
print(f"\n{passed}/{len(results)} checks passed against the LIVE app at {args.url}")
return 0 if passed == len(results) else 1
if __name__ == "__main__":
raise SystemExit(main())