#!/usr/bin/env python3 """The mobile/viewport layout audit (goal 41 T12). Boots the REAL app through the e2e harness (start_bundle: real binary, real statum wire, fake saddle daemon), seeds representative data for every page family, then renders each route in headless Chromium at three widths (360 = phone, 768 = tablet, 1280 = desktop) and reports every element that overflows its box horizontally, plus page-level horizontal scrolling. Rendering is FULL hydration, not static HTML: the audit seeds the harness login's auth frame into localStorage via the Chrome DevTools Protocol (Page.addScriptToEvaluateOnNewDocument) BEFORE statum.js runs, so every stm-* template evaluates against real server state, CSS applies, and client engines (sub-nav, log feed, filters) boot. Usage: python3 test/mobile_audit.py [app_binary] [--json OUT.json] [--widths 360,768,1280] [--routes a,b] [--keep] Exit code 0 = zero findings. chromium-browser/chromium is required; the script tells you when it is missing (the t12mobile bundle gates on the same probe and skips its browser part there). """ import json import os import shutil import subprocess import sys import tempfile import time import urllib.request HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import harness # noqa: E402 (module-level import: sets REPO/APP defaults) # The harness's default scratch dir is a top-level /tmp path the # janitor reaps mid-run (a known fleet gotcha) — default to a # repo-local one unless the caller set their own. if "WEBSTACK_WORK" not in os.environ: os.environ["WEBSTACK_WORK"] = os.path.join( harness.REPO, "build", "ws-mobile") CHROME_CANDIDATES = ("chromium-browser", "chromium", "google-chrome", "google-chrome-stable") # Every route the app ships, with the fixtures each needs. Detail-page # ids are resolved from the seeded data at runtime (see seed()). ROUTE_PAGES = ("/", "/tasks", "/goals", "/projects", "/rules", "/stacks", "/discoveries", "/chats", "/saddles", "/users", "/models", "/profile") DETAIL_PAGES = ("/tasks/{task_id}", "/tasks/{task_id}/log", "/goals/{goal_id}", "/projects/{project_id}", "/stacks/{stack_id}", "/chats/{chat_id}") DEFAULT_WIDTHS = (360, 768, 1280) def find_chromium(): for name in CHROME_CANDIDATES: path = shutil.which(name) if path: return path return None # --------------------------------------------------------------------------- # The in-page probe: runs inside the rendered page (Runtime.evaluate), # after statum has hydrated. Reports every horizontally-overflowing # element (scrollWidth > clientWidth + tolerance), page-level overflow, # tiny touch targets in the sticky chrome, and dialog widths. # --------------------------------------------------------------------------- PROBE_JS = r""" (() => { const out = {href: location.pathname, ready: !!(window.statum && window.statum.version), pageOverflowX: null, offenders: [], touch: [], dialogs: []}; const de = document.documentElement; out.pageOverflowX = de.scrollWidth - de.clientWidth; const label = (el) => { if (el.id) return el.tagName.toLowerCase() + '#' + el.id; let sel = el.tagName.toLowerCase(); if (el.classList && el.classList.length) sel += '.' + [...el.classList].slice(0, 3).join('.'); return sel; }; const path = (el) => { const parts = []; let node = el; while (node && node.nodeType === 1 && parts.length < 4) { parts.unshift(label(node)); node = node.parentElement; } return parts.join(' > '); }; const VISIBLE = (el) => { const cs = getComputedStyle(el); if (cs.display === 'none' || cs.visibility === 'hidden') return false; const r = el.getBoundingClientRect(); return r.width > 0 && r.height > 0; }; const all = document.querySelectorAll('*'); // A box that sticks out past the viewport's right edge is a page // scroll contributor (in-flow or abs-positioned alike). Boxes fully // inside — including .table-wrap/.subnav/terminal that scroll // internally — are fine. const docW = de.clientWidth; const contained = (el) => { for (let a = el.parentElement; a; a = a.parentElement) { const ax = getComputedStyle(a).overflowX; if (ax !== 'visible') return true; // clipped/scrolling ancestor } return false; }; for (const el of all) { if (!VISIBLE(el)) continue; if (contained(el)) continue; const r = el.getBoundingClientRect(); if (r.right > docW + 0.5 && r.width > 0) { const cs = getComputedStyle(el); out.offenders.push({ path: path(el), tag: el.tagName.toLowerCase(), right: Math.round(r.right), width: Math.round(r.width), over: Math.round(r.right - docW), pos: cs.position, white: cs.whiteSpace, }); } } // Cap the offender list: 40 richest entries (biggest overflow first). out.offenders.sort((a, b) => b.over - a.over); out.offenders = out.offenders.slice(0, 40); // Touch targets: interactive controls in the sticky chrome + any // chips — must be >= 44px tall at phone widths to be tappable. const controls = document.querySelectorAll( 'button, [role=button], a.badge, .chip, summary, .dialog-close, ' + '.subnav a, .tabs label, label.choice, .btn'); for (const el of controls) { if (!VISIBLE(el)) continue; // Checkboxes/radios ride inside label.choice — the LABEL is the // touch target; the input itself is intentionally small. if (el.tagName === 'INPUT') continue; const r = el.getBoundingClientRect(); if (r.height > 0 && r.height < 44 && el.offsetParent !== null) { const cs = getComputedStyle(el); // Inline text links inside paragraphs are reading flow, not // controls — skip pure anchors without a class. if (el.tagName === 'A' && !el.className.trim()) continue; out.touch.push({path: path(el), h: Math.round(r.height), display: cs.display}); } } out.touch = out.touch.slice(0, 30); // Dialog boxes: closed dialogs have no layout — measure one by // opening it if the page carries the app's dialog helper, then // close it again. The fit contract is spry's width: min(92vw, 24rem) // for .dialog-sm; an OPEN dialog must sit fully inside the viewport. if (window.openAppDialog) { const dlg = document.querySelector('dialog'); if (dlg && !dlg.open) { try { window.openAppDialog(dlg.id); const r = dlg.getBoundingClientRect(); out.dialogs.push({id: dlg.id || 'first', left: Math.round(r.left), right: Math.round(r.right), fits: r.right <= de.clientWidth + 0.5 && r.left >= -0.5}); dlg.close(); } catch (e) { /* the page's own open hooks may differ */ } } } return out; })() """ class Chrome: """Minimal CDP client over --headless --remote-debugging-port. Deliberately dependency-light: python3-websocket-client only (no selenium/playwright). One tab per audit; pages are navigated with Page.navigate + a settled-DOM wait loop. """ def __init__(self, binary, port): self.log = open(os.path.join(tempfile.gettempdir(), "mobile-audit-chrome.log"), "w") self.proc = subprocess.Popen( [binary, "--headless=new", "--no-sandbox", "--disable-gpu", "--disable-dev-shm-usage", "--remote-debugging-port=%d" % port, "--user-data-dir=%s" % tempfile.mkdtemp(prefix="audit-chrome-"), "--no-first-run", "--remote-allow-origins=*", "about:blank"], stdout=self.log, stderr=subprocess.STDOUT) self.port = port self.ws = None self.msg_id = 0 self._await_http(port) @staticmethod def _await_http(port, timeout=30): deadline = time.time() + timeout url = "http://127.0.0.1:%d/json/version" % port while time.time() < deadline: try: with urllib.request.urlopen(url, timeout=2) as response: if response.status == 200: return True except OSError: time.sleep(0.25) raise SystemExit("chromium devtools endpoint never came up on %d" % port) def connect(self): import websocket # python3-websocket-client with urllib.request.urlopen( "http://127.0.0.1:%d/json/list" % self.port) as response: targets = json.loads(response.read().decode()) page = next(t for t in targets if t["type"] == "page") self.ws = websocket.create_connection(page["webSocketDebuggerUrl"], timeout=30) self.touch_media = False self._send("Target.createTarget" if False else "Page.enable") self._send("Runtime.enable") self._send("Network.enable") def _send(self, method, params=None, timeout=30): self.msg_id += 1 self.ws.send(json.dumps({"id": self.msg_id, "method": method, "params": params or {}})) deadline = time.time() + timeout while time.time() < deadline: self.ws.settimeout(max(0.5, deadline - time.time())) try: message = json.loads(self.ws.recv()) except websocket.WebSocketTimeoutException: break if message.get("id") == self.msg_id: return message return {"error": "timeout waiting for %s" % method} def _eval(self, expression, await_promise=False): result = self._send("Runtime.evaluate", {"expression": expression, "returnByValue": True, "awaitPromise": await_promise}, timeout=60) value = (result.get("result") or {}).get("result", {}) return value.get("value") def add_init_script(self, script): self._send("Page.addScriptToEvaluateOnNewDocument", {"source": script}) def set_viewport(self, width, height=900): # Coarse pointer at phone widths: headless reports pointer:fine # by default, which would hide every @media (pointer: coarse) # rule from the audit. Emulate a finger on phones so the touch # -target findings (and their fixes) are actually exercised. self.touch_media = width <= 500 self._send("Emulation.setDeviceMetricsOverride", {"width": width, "height": height, "deviceScaleFactor": 1, "mobile": width <= 500}) def goto(self, url, settle_s=3.0, timeout_s=45): """Navigate and wait for statum hydration + network quiet.""" # Touch input at phone widths: headless reports NO pointer type # at all (both pointer:fine and pointer:coarse match nothing), # so Emulation.setEmulatedMedia cannot exercise the touch # rules. setTouchEmulationEnabled flips any-pointer: coarse, # which the CSS keys on (phones are any-pointer: coarse too). if self.touch_media: self._send("Emulation.setTouchEmulationEnabled", {"enabled": True, "maxTouchPoints": 5}) self._send("Page.navigate", {"url": url}, timeout=60) deadline = time.time() + timeout_s hydrated = False while time.time() < deadline: state = self._eval( "document.readyState === 'complete' && " "!!(window.statum && window.statum.version) && " "(window.__auditHalt ? 'halt' : true)") if state == "halt": return False if state is True: hydrated = True break time.sleep(0.2) if not hydrated: print(" (warning: %s never fully hydrated)" % url) # Let afterEvaluate hooks, subnav engine and fonts settle. time.sleep(settle_s) return True def probe(self): return self._eval(PROBE_JS) def screenshot(self, path): self._send("Page.captureScreenshot", {"format": "png"}, timeout=60) def close(self): try: if self.ws: self.ws.close() self.proc.terminate() self.proc.wait(timeout=10) except OSError: pass self.log.close() # --------------------------------------------------------------------------- # Fixtures: representative data for every page family, planted through # the same wire/sqlite paths the bundles use. # --------------------------------------------------------------------------- def seed(ctx): """Returns the format-map for DETAIL_PAGES routes. Every write COMMITS immediately and every read opens a fresh connection: the wire act() calls below write through the app process, and a python-held open transaction would both hide their commits from this snapshot and block the app's writer. """ auth = ctx.auth now = "strftime('%s','now')" def sql(statement, args=()): conn = harness.db() try: conn.execute(statement, args) conn.commit() finally: conn.close() def one(statement): conn = harness.db() try: return conn.execute(statement).fetchone()[0] finally: conn.close() # -- a second project (links) ------------------------------------ sql("insert into projects (name, git_url, default_branch, summary," " created_at) values ('audit-second'," " 'https://example.com/very/long/path/to/second/repo.git', 'main'," " 'A summary line for the second audit project.', %s)" % now) # -- tasks in every lifecycle shape ------------------------------- for statement in ( "insert into tasks (project_id, title, status, priority," " created_at) values (1, 'Audit queued task', 'queued', 5," " strftime('%s','now'))", "insert into tasks (project_id, title, status, priority," " created_at) values (1, 'Audit working task', 'working', 5," " strftime('%s','now'))", "insert into tasks (project_id, title, status, priority," " created_at) values (1, 'Audit blocked task', 'blocked', 5," " strftime('%s','now'))", "insert into tasks (project_id, title, status, priority, summary," " completed_at, created_at) values (1, 'Audit done task with a" " very long completion summary heading', 'done', 5," " 'Done: the audit harness landed cleanly\n\nBody paragraphs" " render under the headline.'," " strftime('%s','now'), strftime('%s','now','-60 seconds'))", "insert into tasks (project_id, title, status, priority," " created_at) values (1, 'Audit failed task', 'failed', 5," " strftime('%s','now'))"): sql(statement) task_id = one("select id from tasks where title = 'Audit working task'") blocked_id = one("select id from tasks where title = 'Audit blocked task'") # The blocked task owns a pending block (the pile surface) ... sql("insert into task_blocks (task_id, type, question, options," " created_at) values (?, 'question'," " 'Which datastore should the audit harness back its fixtures" " with, given the fleet runs on cheap hardware?'," " '[\\\"SQLite\\\",\\\"Postgres\\\"]', %s)" % now, (blocked_id,)) # ... and one answered block (the history fold). sql("insert into task_blocks (task_id, type, question, answer," " answered_at, answered_by, created_at) values" " (?, 'question', 'Hold or fold?', 'Hold — the audit says the" " layout holds.', %s, 'user:admin', %s)" % (now, now), (blocked_id,)) sql("insert into task_comments (task_id, author, body, created_at)" " values (?, 'user:admin', 'An audit comment with a quite long" " unbroken SSH-style token" " ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIExampleVeryLongKeyBody" "ThatDoesNotWrap user@host', %s)" % now, (task_id,)) # A working session on the working task + a rich event log. sql("insert into sessions (task_id, saddle_id, started_at)" " values (?, ?, %s)" % now, (task_id, ctx.fake_id)) session_id = one("select max(id) from sessions") events = [ ("comment", "user:admin", {"body": "Long comment body for the" " audit feed: " + "wrap-check-word " * 12}), ("tool_call", "model", {"tool": "run_command", "args": {"command": "meson setup build && ninja -C build" " && python3 test/web_stack_test.py ./build/clanker-saddle" " --bundle t7log --parallel 8"}}), ("tool_result", "model", {"result": "ok: the exit code was 0 and" " the output line was quite long indeed, well past the wrap point" " of a narrow terminal viewport 123456789012345678901234567890"}), ("progress", "model", {"text": "Overseeing the audit fixture" " fan-out with a long progress line."}), ("verdict", "model", {"verdict": "accept"}), ("turn_text", "model", {"text": "A turn of narrative text that" " spans a few sentences so the log feed has to wrap it in the" " narrow viewport."}), ("usage", "model", {"prompt_tokens": 12000, "completion_tokens": 900, "cached_tokens": 11000}), ] for seq, (kind, actor, payload) in enumerate(events, start=1): sql("insert into events (session_id, seq, kind, actor, scope_type," " scope_id, payload, created_at) values (?, ?, ?, ?, 'task', ?," " ?, %s - ?)" % now, (session_id, seq, kind, actor, task_id, json.dumps(payload), seq)) # The blocked task carries an open block event so its pile card renders. sql("insert into events (session_id, seq, kind, actor, scope_type," " scope_id, payload, created_at) values (NULL, 9001, 'question'," " 'model', 'task', ?, ?, %s)" % now, (blocked_id, json.dumps({"question": "Which datastore..."}))) # -- a goal with the full review surface -------------------------- # The honest wire flow (the goalui pattern): script the fake agent # to raise a plan review, create the goal, START it — the app then # mints the goal_run task and the agent parks blocked on the plan # question, which gives /goals/ its live review surface. ctx.fake.goal_script = [( "plan", "# Audit plan\n\n## Steps\n\n1. A very long numbered step" " that has to wrap inside the plan review card on a narrow screen" " without spilling sideways\n2. Second step")] goal_create = harness.slot_public(auth, "/goals", "goals")["create"] goal_id = None for attempt in range(3): harness.act(auth, goal_create, { "project_id": "1", "title": "Audit goal with a long title that" " must wrap on a narrow phone viewport without spilling", "brief": "Drive the mobile audit end to end across the fleet's" " surfaces with representative fixture depth.", "branch_name": "goal/audit-t12"}) goal_id = harness.wait_for( lambda: (one("select count(*) from goals where title like" " 'Audit goal%'") and "yes") or False, timeout=10, what="the audit goal row") and one( "select id from goals where title like 'Audit goal%'") if goal_id: break # A dispatch tick can hold the writer; retry the create once. time.sleep(1) harness.act(auth, harness.goal_state(goal_id)["start"], {}) harness.wait_for( lambda: (one("select count(*) from tasks where goal_id = %d" " and kind = 'goal_run'" % goal_id) and "yes") or False, timeout=60, what="the audit goal's run task") run_task = one("select id from tasks where goal_id = %d" " and kind = 'goal_run'" % goal_id) harness.wait_for( lambda: (one("select status from tasks where id = %d" % run_task) == "blocked" and "yes") or False, timeout=60, what="the audit goal agent parks on its plan review") # One answered plan revision (the history fold surface). sql("insert into task_blocks (task_id, type, question, answer," " answered_at, answered_by, created_at) values (?, 'plan'," " '# Earlier plan revision\n\nOlder body.', 'Add a step for" " the mobile audit, please.', %s, 'user:admin', %s)" % (now, now), (run_task,)) # -- stacks -------------------------------------------------------- stacks_create = harness.slot_public(auth, "/stacks", "stacks")["create"] harness.act(auth, stacks_create, {"name": "audit-stack"}) stack_id = one("select id from stacks where name = 'audit-stack'") # -- rules --------------------------------------------------------- rules_create = harness.slot_public(auth, "/rules", "rules")["create"] harness.act(auth, rules_create, { "name": "audit-rule", "body": "Always build with meson before running the suite; the" " canonical builddir is ./builddir for canon_deploy."}) # -- chats (advisor) ------------------------------------------------- sql("insert into advisor_chats (user_id, title, status, created_at," " updated_at) values (1, 'Audit chat about layout widths'," " 'active', %s, %s)" % (now, now)) chat_id = one("select max(id) from advisor_chats") for kind, content in ( ("user", "What widths should the audit check for horizontal" " overflow?"), ("assistant", "360 for phones, 768 for tablets, 1280 for" " desktops — the audit probes all three."), ("question", "Which bundler do you use for the audit wire?")): sql("insert into advisor_messages (chat_id, kind, content," " created_at) values (?, ?, ?, %s)" % now, (chat_id, kind, content)) # -- discoveries ---------------------------------------------------- sql("insert into discoveries (body, source_task_id, created_at," " updated_at) values (?, ?, %s, %s)" % (now, now), ("**Chromium headless tip** — the CDP probe pattern: seed the" " statum auth frame into localStorage via" " Page.addScriptToEvaluateOnNewDocument before statum.js runs" " and every page hydrates fully. body/" "wouldyouwrapthislongunbrokentoken" "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIAnotherExtremelyLongKey" "ForWrapTesting user@host", task_id)) return {"task_id": task_id, "goal_id": goal_id, "project_id": 2, "stack_id": stack_id, "chat_id": chat_id} def run_audit(app, widths, routes=None, json_out=None, keep=False): chromium = find_chromium() if not chromium: print("SKIP no chromium binary found (looked for: %s)" % ", ".join(CHROME_CANDIDATES)) return None ctx = harness.start_bundle("t12mobile") harness.CTX = ctx # goal_state() reads the module's current context try: ids = seed(ctx) except Exception: # Never orphan the app process on a seeding failure — a later # free_port() TOCTOU then kills the NEXT run's app at bind. try: ctx.app.terminate() ctx.app.wait(timeout=10) except OSError: pass raise auth = ctx.auth # Re-derive the auth frame exactly as the browser would hold it: # the login POST's set directive carries snapshot_frame{content}. frames = harness.directives("/login") content = json.loads(frames[0]["snapshot_frame"]["content"]) login_action = content["public"]["login"] status, body = harness.http("POST", login_action["uri"], {"username": "admin", "password": "adminpass123"}) auth_frame = None for directive in json.loads(body): if directive["type"] == "set": frame = json.loads(directive["snapshot_frame"]["content"]) if frame["slot"]["type"] == "auth": auth_frame = directive["snapshot_frame"] if auth_frame is None: raise SystemExit("no auth frame in login response") port = harness.free_port() chrome = Chrome(chromium, port) try: chrome.connect() # Seed localStorage BEFORE any page script runs. The frame is # the exact JSON statum.js persists (persistEntry: the frame, # not the snapshot). chrome.add_init_script( "try { localStorage.setItem('statum:auth', %s); } catch (e) {}" % json.dumps(json.dumps(auth_frame))) chrome.set_viewport(widths[0]) all_routes = [r.format(**ids) for r in (routes or (list(ROUTE_PAGES) + list(DETAIL_PAGES)))] findings = {} for route in all_routes: findings[route] = {} for width in widths: chrome.set_viewport(width) ok = chrome.goto(harness.WEB + route) if not ok: findings[route][width] = {"error": "navigation halted"} continue findings[route][width] = chrome.probe() or {} findings[route][width]["url"] = route findings[route][width]["width"] = width if json_out: with open(json_out, "w") as handle: json.dump(findings, handle, indent=1) return findings finally: chrome.close() if not keep: try: ctx.app.terminate() ctx.app.wait(timeout=10) except OSError: pass try: ctx.fake.shutdown() except AttributeError: pass def summarise(findings): """Prints the per-page/width report; returns a flat finding list.""" rows = [] dialog_bad = [] print("\n%-34s %6s %8s %s" % ("route", "width", "pageDX", "offenders")) for route in sorted(findings): for width in sorted(findings[route]): data = findings[route][width] if "error" in data: print("%-34s %6d ERROR %s" % (route, width, data["error"])) rows.append((route, width, "error", data["error"])) continue page_dx = data.get("pageOverflowX") or 0 offenders = data.get("offenders") or [] touch = data.get("touch") or [] for dialog in data.get("dialogs") or []: if not dialog.get("fits", True): dialog_bad.append( (route, width, "dialog %s spills (right=%s)" % (dialog.get("id"), dialog.get("right")))) mark = "" if (page_dx <= 0 and not offenders) else " <-- FIX" print("%-34s %6d %8d %3d%s" % (route, width, page_dx, len(offenders), mark)) for offender in offenders: rows.append((route, width, "overflow", "%s right=%d w=%d (%+d) pos=%s ws=%s" % (offender["path"], offender.get("right", 0), offender.get("width", 0), offender["over"], offender.get("pos"), offender.get("white")))) for target in touch: rows.append((route, width, "touch", "%s h=%dpx" % (target["path"], target["h"]))) for route, width, note in dialog_bad: rows.append((route, width, "dialog", note)) print(" DIALOG %s @%s %s" % (route, width, note)) return rows def main(): app = sys.argv[1] if len(sys.argv) > 1 and not sys.argv[1].startswith("-") \ else os.path.join(harness.REPO, "builddir", "clanker-saddle") sys.argv = [a for a in sys.argv if a != app or sys.argv.index(a) == 0] widths = DEFAULT_WIDTHS routes = None json_out = os.environ.get("MOBILE_AUDIT_JSON") keep = False args = sys.argv[1:] i = 0 while i < len(args): if args[i] == "--widths": widths = [int(w) for w in args[i + 1].split(",")] i += 2 elif args[i] == "--routes": routes = args[i + 1].split(",") i += 2 elif args[i] == "--json": json_out = args[i + 1] i += 2 elif args[i] == "--keep": keep = True i += 1 else: i += 1 if not os.path.exists(app): raise SystemExit("app binary not found at %s (build first)" % app) findings = run_audit(app, widths, routes, json_out, keep) if findings is None: return 0 rows = summarise(findings) overflow = [r for r in rows if r[2] == "overflow"] dialogs = [r for r in rows if r[2] == "dialog"] touch = [r for r in rows if r[2] == "touch"] print("\nSUMMARY: %d overflow findings, %d dialog-fit findings," " %d sub-44px touch targets" % (len(overflow), len(dialogs), len(touch))) for row in overflow: print(" OVERFLOW %s @%s %s" % (row[0], row[1], row[3])) for row in dialogs: print(" DIALOG %s @%s %s" % (row[0], row[1], row[3])) for row in touch: print(" TOUCH %s @%s %s" % (row[0], row[1], row[3])) return 1 if (overflow or dialogs) else 0 if __name__ == "__main__": sys.exit(main())