tests/e2e/conftest.py

"""Browser-suite plumbing: a static server for the standalone harness, a real Gradio
app subprocess per transport for the parity proof, and the launch profiles the deployed
suite drives the public Space with.

The local suites must be green BEFORE anything deploys, because that is what makes a
deployed failure interpretable: if standalone passes and deployed fails, the problem is
the host, not three.js.

The deployed profiles are declared once here, as fixtures, so a test that needs a fake
microphone or a browser with no WebGPU says so by requesting one rather than by
repeating Chromium's launch flags. They are the same flags
``tests/e2e/test_asr_standalone.py`` proved locally.
"""

from __future__ import annotations

import contextlib
import functools
import http.server
import mimetypes
import os
import socket
import subprocess
import sys
import tempfile
import threading
import time
import urllib.error
import urllib.request
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parent.parent.parent
APP_BOOT_TIMEOUT_S = 180

# --------------------------------------------------------------------- deployed profiles

# Chromium's fake media stack: the permission prompt auto-accepts, the fake device is
# selected, and --use-file-for-fake-audio-capture (added per launch) feeds a 16-bit PCM
# WAV as the microphone. %noloop plays it once and then silence, so a hold longer than
# the clip captures the clip plus silence rather than the clip twice.
AUDIO_CAPTURE_ARGS = (
    "--autoplay-policy=no-user-gesture-required",
    "--use-fake-ui-for-media-stream",
    "--use-fake-device-for-media-stream",
)
NO_WEBGPU_ARGS = ("--disable-features=WebGPU", "--disable-gpu")

# Chromium 151 keeps navigator.gpu defined with WebGPU disabled by flag - the adapter is
# withheld but the API surface stays. test_asr_wasm_fallback asserts the branch where the
# API is ABSENT, so the property is deleted in the page as well. A bare statement, not an
# arrow function: add_init_script evaluates the string as a script.
HIDE_WEBGPU = "try { delete Navigator.prototype.gpu; } catch (e) {}"

# whisper-base q4 is 135.8 MB on first use. The model lives in the browser's Cache API,
# keyed by page origin, so a stable profile directory is what makes the second deployed
# run a disk read instead of a download. Separate from the local ASR suite's profile:
# the origins differ, so the caches would not be shared anyway.
DEPLOYED_ASR_PROFILE = Path(tempfile.gettempdir()) / "jla-deployed-chromium-profile"

# The Space sleeps after 48 h (gcTimeout 172800); the first navigation of a session may
# pay a cold start. Hugging Face answers 503 while the container is scheduled and built.
COLD_START_TIMEOUT_S = 300
COLD_START_POLL_S = 5
STAGE_ATTACHED_TIMEOUT_MS = 30_000
AVATAR_READY_TIMEOUT_MS = 90_000
FIRST_FRAME_TIMEOUT_MS = 30_000

AVATAR_READY = (
    "() => !!window.Avatar && !!window.Avatar.__debug && window.Avatar.__debug.ready === true"
)
# ready fires before the first frame renders, and the first frame compiles every MToon
# shader (>1.5 s headless). breathValue is written every tick and is 0 only before the
# first one, so it is the rendered-frame signal. Never a fixed sleep.
AVATAR_FIRST_FRAME = """
async () => {
  const d = window.Avatar ? await window.Avatar.getDebug() : null;
  return !!d && d.breathValue !== 0;
}
"""

# Installed before navigation. Subscribes to the facade's event bus the moment
# window.Avatar exists - which is before the VRM finishes mounting, so nothing a turn
# emits can be missed - and records every event with the page clock and, where the
# event carries one, the played AudioBuffer's duration. Also installs a viseme watcher
# that samples currentVisemes at animation-frame rate between the next speech-start
# and the speech-end that follows it, so "the mouth moved during THIS playback" is a
# number rather than a page-lifetime peak.
EVENT_TAP_JS = """
(() => {
  const NAMES = ['turn-start', 'turn', 'speech-start', 'speech-end', 'latency', 'error',
                 'transcript', 'asr-tier', 'listening', 'replay'];
  const VISEMES = ['aa', 'ih', 'ou', 'ee', 'oh'];
  window.__events = [];
  const plain = (d) => {
    try { return d === undefined ? null : JSON.parse(JSON.stringify(d)); }
    catch (e) { return null; }
  };
  const attach = () => {
    const a = window.Avatar;
    if (!a || typeof a.on !== 'function' || a.__eventTap) return false;
    a.__eventTap = true;
    for (const name of NAMES) {
      a.on(name, (d) => window.__events.push({
        event: name,
        t: performance.now(),
        audioDuration: d && typeof d.duration === 'number' ? d.duration : null,
        data: plain(d),
      }));
    }
    return true;
  };
  const timer = setInterval(() => { if (attach()) clearInterval(timer); }, 20);

  window.__watchVisemes = (timeoutMs) => new Promise((resolve) => {
    const max = {}; for (const v of VISEMES) max[v] = 0;
    const count = (name) => window.__events.filter((e) => e.event === name).length;
    const starts = count('speech-start');
    const ends = count('speech-end');
    let frames = 0;
    const t0 = performance.now();
    const step = async () => {
      const started = count('speech-start') > starts;
      if (started && window.Avatar) {
        try {
          const d = await window.Avatar.getDebug();
          for (const v of VISEMES) max[v] = Math.max(max[v], d.currentVisemes[v]);
          frames += 1;
        } catch (e) { /* mid-render; next frame answers */ }
      }
      if ((started && count('speech-end') > ends) || performance.now() - t0 >= timeoutMs) {
        resolve({ max, frames, started, timedOut: performance.now() - t0 >= timeoutMs });
      } else {
        requestAnimationFrame(step);
      }
    };
    requestAnimationFrame(step);
  });
})();
"""


def fake_audio_arg(wav: Path) -> str:
    """The fake-microphone flag. Chromium wants an absolute POSIX-style path."""
    return f"--use-file-for-fake-audio-capture={Path(wav).resolve().as_posix()}%noloop"


# ------------------------------------------------------------ strict autoplay (plan 01-11)
#
# Every profile above, and the session-wide one below, runs Chromium with
# --autoplay-policy=no-user-gesture-required. That flag is what let the phone bug ship: an
# AudioContext resumed seconds after the tap looked fine in every test and was refused on a
# real gesture-gated browser. The profile here launches WITHOUT it, under Chromium's strict
# default, and drives the page so that NOTHING grants user activation before the one
# trusted tap under test. Three facts, all measured on this fleet (Chromium 151) while this
# profile was built, shape it:
#
# 1. Chromium has two strict values. document-user-activation-required (the desktop
#    default) gates a top-level page: an ungestured context is created 'suspended', an
#    ungestured resume() is refused with a console warning and its promise stays PENDING.
#    Activation is STICKY: after any trusted input, a resume() issued at any later time is
#    allowed. user-gesture-required gates Web Audio only inside cross-origin frames; at the
#    top level it leaves the context constructed 'running', so it cannot serve as a strict
#    profile here. It is what the huggingface.co embed applies to the Space, and a note
#    for a future cross-origin-embed row.
# 2. Playwright's page.evaluate / wait_for_function / wait_for_selector run through CDP
#    with userGesture:true and GRANT activation - navigator.userActivation.hasBeenActive
#    flips true - after which the sticky policy lets a late resume() through. A strict test
#    therefore must not touch the page with those before the tap: reads go through a raw
#    CDP session with userGesture:false, and the tap is raw mouse input at coordinates read
#    the same way. After the tap activation exists in real life too, so Playwright's own
#    helpers are fine again.
# 3. A trusted tap grants activation in every launch mode (headless shell, new headless,
#    headed), and the resume() -> 'running' transition then takes ~0.6-0.7 s here - the
#    headless audio sink starting - so 'running' is waited for, never read once.
#
# iOS Safari is stricter than any Chromium policy: the resume() call must be on the
# gesture's own call stack. No flag models that; GESTURE_TAP_JS does, by recording whether
# each resume() was issued while a trusted input event was being dispatched.
STRICT_AUTOPLAY_POLICY = "document-user-activation-required"
CDP_POLL_S = 0.1
# A trusted tap's resume() reaches 'running' once the audio sink starts (~0.7 s measured).
UNLOCK_RUNNING_TIMEOUT_MS = 10_000
# The load-bearing bound is relative: 'running' must be observed BEFORE the server answers
# ('turn'), which is seconds later on the turn path. This absolute ceiling is only a sanity
# check - measured tap-to-running is 0.7 s on an idle fleet and up to ~3.5 s with two
# Gradio apps synthesising under SwiftShader rendering, so it is set well above both.
UNLOCK_BEFORE_RESPONSE_MS = 10_000
# A gesture relayed into the same-origin stage frame by postMessage lands a task later.
RELAYED_UNLOCK_MAX_MS = 1_000

# Installed before navigation, in every frame. Wraps AudioContext.prototype.resume so each
# call records whether it was issued inside the dispatch of a trusted input event, and
# whether it was on that dispatch's CALL STACK - the distinction iOS enforces.
#
# "Inside the dispatch" (inDispatch) is a capture listener on document (runs before any
# handler on the target) and a bubble listener on window (runs after every handler), with
# a macrotask fallback in case a handler stops propagation. That alone is not enough: an
# `await` continuation queued by a handler runs as a microtask before the bubble listener,
# still "inside" the dispatch but no longer on the gesture's stack. So `sync` is measured
# exactly: addEventListener is wrapped for the gesture event types and every page listener
# runs inside a depth counter; a resume() with the counter above zero was called
# synchronously from a trusted-input listener, and a continuation - which runs after the
# listener returned - sees zero. Records carry both the page clock and the wall clock,
# because a child frame has its own performance.now().
GESTURE_TAP_JS = """
(() => {
  const GESTURES = ['pointerdown', 'mousedown', 'touchstart', 'keydown', 'click'];
  const rec = { inDispatch: false, syncDepth: 0, resumeCalls: [], dispatches: [] };
  window.__gesture = rec;
  const enter = (e) => {
    rec.inDispatch = true;
    rec.dispatches.push({ type: e.type, trusted: e.isTrusted, t: performance.now(),
                          wall: Date.now() });
    setTimeout(() => { rec.inDispatch = false; }, 0);
  };
  const leave = () => { rec.inDispatch = false; };
  for (const name of GESTURES) {
    document.addEventListener(name, enter, true);
    window.addEventListener(name, leave, false);
  }
  const wrappers = new WeakMap();
  const proto = EventTarget.prototype;
  const add = proto.addEventListener;
  const remove = proto.removeEventListener;
  proto.addEventListener = function addTapped(type, listener, ...rest) {
    if (!GESTURES.includes(type) || typeof listener !== 'function') {
      return add.call(this, type, listener, ...rest);
    }
    let wrapped = wrappers.get(listener);
    if (!wrapped) {
      wrapped = function gestureListener(event) {
        if (!event.isTrusted) return listener.call(this, event);
        rec.syncDepth += 1;
        try {
          return listener.call(this, event);
        } finally {
          rec.syncDepth -= 1;
        }
      };
      wrappers.set(listener, wrapped);
    }
    return add.call(this, type, wrapped, ...rest);
  };
  proto.removeEventListener = function removeTapped(type, listener, ...rest) {
    const wrapped = typeof listener === 'function' ? wrappers.get(listener) : null;
    return remove.call(this, type, wrapped || listener, ...rest);
  };
  const Ctor = window.AudioContext || window.webkitAudioContext;
  if (Ctor && Ctor.prototype && typeof Ctor.prototype.resume === 'function') {
    const original = Ctor.prototype.resume;
    Ctor.prototype.resume = function resumeTapped(...args) {
      const stack = String(new Error().stack || '').split('\\n').slice(1, 6);
      rec.resumeCalls.push({
        t: performance.now(), wall: Date.now(), inDispatch: rec.inDispatch,
        sync: rec.syncDepth > 0, stateBefore: this.state,
        stack: stack.map((line) => line.trim()).join(' | '),
      });
      return original.apply(this, args);
    };
  }
})();
"""

# Armed on a page whose avatar is ready, BEFORE the gesture, through the CDP session so it
# grants nothing. Timestamps the trusted tap on `selector` (capture phase on document, so
# before the host's handler runs), samples audioState at animation-frame rate, and records
# the audioState the facade reports at each turn event. getDebug() resolves to the live
# merged object, so every scalar is copied out the moment the promise settles.
UNLOCK_PROBE_JS = """
(selector) => {
  const rec = { selector, clickAt: null, clickWall: null, runningAt: null, done: false,
                events: [], samples: [] };
  window.__unlock = rec;
  const plain = (d) => {
    try { return d === undefined ? null : JSON.parse(JSON.stringify(d)); }
    catch (e) { return null; }
  };
  const stamp = (e) => {
    if (rec.clickAt !== null) return;
    const target = e.target && e.target.closest ? e.target.closest(selector) : null;
    if (target) { rec.clickAt = performance.now(); rec.clickWall = Date.now(); }
  };
  document.addEventListener('pointerdown', stamp, true);
  document.addEventListener('click', stamp, true);
  for (const name of ['turn-start', 'turn', 'speech-start', 'speech-end', 'error']) {
    window.Avatar.on(name, (d) => {
      const t = performance.now();
      window.Avatar.getDebug().then((dbg) => rec.events.push({
        name, t, audioState: dbg.audioState, readAt: performance.now(), data: plain(d),
      }));
    });
  }
  const step = async () => {
    if (rec.done) return;
    const t = performance.now();
    try {
      const dbg = await window.Avatar.getDebug();
      rec.samples.push({ t, audioState: dbg.audioState });
      if (dbg.audioState === 'running' && rec.runningAt === null) rec.runningAt = t;
    } catch (e) { /* mid-render; the next frame answers */ }
    requestAnimationFrame(step);
  };
  requestAnimationFrame(step);
  return true;
}
"""

USER_ACTIVATION_JS = (
    "navigator.userActivation ? { isActive: navigator.userActivation.isActive, "
    "hasBeenActive: navigator.userActivation.hasBeenActive } : null"
)
AVATAR_READY_EXPR = (
    "!!window.Avatar && !!window.Avatar.__debug && window.Avatar.__debug.ready === true"
)
AVATAR_FIRST_FRAME_EXPR = (
    "window.Avatar ? window.Avatar.getDebug().then((d) => !!d && d.breathValue !== 0) : false"
)
AUDIO_STATE_EXPR = (
    "window.Avatar ? window.Avatar.getDebug().then((d) => "
    "d.audioState === undefined ? null : d.audioState) : null"
)


def _goto_cold_start_tolerant(page, url: str) -> float:
    """Navigate, retrying while a sleeping Space answers 503. Returns the start time."""
    t0 = time.monotonic()
    deadline = t0 + COLD_START_TIMEOUT_S
    last: object = None
    while True:
        try:
            response = page.goto(url, timeout=STAGE_ATTACHED_TIMEOUT_MS + 30_000)
            if response is not None and response.ok:
                return t0
            last = response.status if response is not None else "no response"
        except Exception as err:  # noqa: BLE001 - the retry IS the cold-start handling
            last = err
        if time.monotonic() > deadline:
            raise AssertionError(
                f"{url} never answered 2xx within {COLD_START_TIMEOUT_S}s; last: {last!r}"
            )
        time.sleep(COLD_START_POLL_S)


class StrictPage:
    """A page under the strict autoplay policy plus a raw CDP session, so a test can read
    it and tap it without any Playwright helper granting user activation first."""

    def __init__(self, page, cdp) -> None:
        self.page = page
        self.cdp = cdp

    def eval(self, js: str):
        """Evaluate WITHOUT granting activation (Runtime.evaluate, userGesture:false).

        Accepts an expression, or a zero-argument arrow function which is invoked, so the
        suite's existing ``() => ...`` predicates can be reused. Promises are awaited and
        the value must be JSON-serialisable.
        """
        stripped = js.strip()
        expression = f"({stripped})()" if stripped.startswith(("() =>", "async () =>")) else js
        result = self.cdp.send(
            "Runtime.evaluate",
            {
                "expression": expression,
                "returnByValue": True,
                "awaitPromise": True,
                "userGesture": False,
            },
        )
        if "exceptionDetails" in result:
            detail = result["exceptionDetails"]
            text = detail.get("exception", {}).get("description") or detail.get("text")
            raise RuntimeError(f"page threw during CDP evaluate: {text}")
        return result["result"].get("value")

    def wait(self, js: str, *, timeout_ms: int, what: str = ""):
        """Poll ``eval(js)`` until truthy. A page that throws mid-load counts as falsy."""
        deadline = time.monotonic() + timeout_ms / 1000
        while True:
            try:
                value = self.eval(js)
            except RuntimeError:
                value = None
            if value:
                return value
            if time.monotonic() > deadline:
                raise AssertionError(f"timed out after {timeout_ms} ms waiting for {what or js}")
            time.sleep(CDP_POLL_S)

    def user_activation(self) -> dict | None:
        return self.eval(USER_ACTIVATION_JS)

    def real_click(self, selector: str) -> dict:
        """TRUSTED input, which is the whole point of the strict-policy tests.

        The element's centre is read through the CDP session and the tap is
        ``page.mouse.click`` - raw Input.dispatchMouseEvent, which Chromium treats as a
        real pointer and grants user activation for. Not ``page.click``: its actionability
        checks evaluate in the page with userGesture:true before the pointer moves, which
        would activate the document a moment early. Never ``page.evaluate('el.click()')``:
        that dispatches a synthetic event with ``isTrusted === false`` and grants NO
        activation, so a strict policy would refuse the resume and the test would fail for
        the wrong reason.
        """
        rect = self.eval(
            f"(() => {{ const el = document.querySelector({selector!r}); if (!el) return null;"
            " const r = el.getBoundingClientRect();"
            " return { x: r.x + r.width / 2, y: r.y + r.height / 2, w: r.width, h: r.height }; })()"
        )
        assert rect and rect["w"] > 0 and rect["h"] > 0, f"{selector} has no box to tap: {rect}"
        self.page.mouse.click(rect["x"], rect["y"])
        return rect

    def open_ready(self, url: str, *, timeout_ms: int = AVATAR_READY_TIMEOUT_MS) -> dict:
        """The gesture-free twin of ``wait_for_avatar_ready``: load, wait for ready and
        then for the first RENDERED frame, reading only through the CDP session."""
        t0 = _goto_cold_start_tolerant(self.page, url)
        self.wait(
            "!!document.querySelector('#vrm-stage')",
            timeout_ms=STAGE_ATTACHED_TIMEOUT_MS,
            what="#vrm-stage attached",
        )
        self.wait(AVATAR_READY_EXPR, timeout_ms=timeout_ms, what="window.Avatar ready")
        ready_s = time.monotonic() - t0
        self.wait(AVATAR_FIRST_FRAME_EXPR, timeout_ms=FIRST_FRAME_TIMEOUT_MS, what="first frame")
        return {"ready_seconds": ready_s, "first_frame_seconds": time.monotonic() - t0}


@pytest.fixture
def chromium_strict_autoplay(playwright):
    """``with chromium_strict_autoplay() as strict:`` - a fresh Chromium under
    STRICT_AUTOPLAY_POLICY, deliberately not built from ``browser_type_launch_args`` (which
    carries the relaxed flag the rest of the suite depends on), with the event tap and the
    gesture tap installed in every frame before navigation, wrapped as a StrictPage."""

    @contextlib.contextmanager
    def launch():
        browser = playwright.chromium.launch(args=[f"--autoplay-policy={STRICT_AUTOPLAY_POLICY}"])
        try:
            context = browser.new_context()
            page = context.new_page()
            page.add_init_script(EVENT_TAP_JS)
            page.add_init_script(GESTURE_TAP_JS)
            cdp = context.new_cdp_session(page)
            yield StrictPage(page, cdp)
        finally:
            browser.close()

    return launch


@pytest.fixture
def real_click():
    """``real_click(strict, selector)`` - see StrictPage.real_click for why it is raw input."""

    def click(strict: StrictPage, selector: str) -> dict:
        return strict.real_click(selector)

    return click


@pytest.fixture
def audio_unlock_probe():
    """The proof that one tap unlocks the audio clock, packaged so the both-transports and
    deployed layers assert exactly the same things.

    ``precondition(strict)`` asserts the ungestured page is really gated; ``arm`` installs
    UNLOCK_PROBE_JS through CDP; after the tap, ``wait`` blocks until speech-end (or an
    error) and returns the record plus every frame's gesture tap; ``assert_unlocked`` is
    the shared check and returns the numbers it derived, for the record.
    """

    class Probe:
        @staticmethod
        def precondition(strict: StrictPage) -> dict:
            """Load-bearing: without this a relaxed profile, or a document something has
            already activated, would let the rest pass vacuously."""
            state = strict.eval(AUDIO_STATE_EXPR)
            activation = strict.user_activation()
            assert activation is not None, "navigator.userActivation is missing"
            assert activation["hasBeenActive"] is False, (
                f"the document was activated before the tap ({activation}); something "
                "evaluated with a user gesture - reads before the tap must go through CDP"
            )
            assert state == "suspended", (
                f"audioState reads {state!r} before any gesture. None: the page does not "
                "publish audioState (a deployed build before plan 01-11 - push first). "
                "'running': the profile is not strict and this test would pass vacuously"
            )
            return {"audioState": state, "activation": activation}

        @staticmethod
        def arm(strict: StrictPage, selector: str) -> None:
            assert strict.eval(f"({UNLOCK_PROBE_JS})({selector!r})") is True

        @staticmethod
        def wait(strict: StrictPage, *, end_timeout_ms: int) -> dict:
            page = strict.page
            page.wait_for_function(
                "() => window.__unlock.events.some((e) => e.name === 'speech-end' || "
                "e.name === 'error')",
                timeout=end_timeout_ms,
            )
            record = page.evaluate("() => { window.__unlock.done = true; return window.__unlock; }")
            frames = {}
            for frame in page.frames:
                try:
                    frames[frame.url] = frame.evaluate("() => window.__gesture")
                except Exception:  # noqa: BLE001 - a detached or opaque frame has no tap
                    continue
            return {"record": record, "gesture": frames}

        @staticmethod
        def assert_unlocked(report: dict, *, relayed: bool = False) -> dict:
            """relayed=True is the iframe transport: the resume() happens inside the stage
            frame on receipt of a postMessage, so it is asserted by wall-clock proximity
            to the tap rather than by being on the tap's own call stack."""
            rec = report["record"]
            names = [e["name"] for e in rec["events"]]
            errors = [e for e in rec["events"] if e["name"] == "error"]
            assert errors == [], f"the turn emitted error events: {errors}"
            assert "speech-start" in names and "speech-end" in names, names
            assert rec["clickAt"] is not None, f"no trusted tap on {rec['selector']} was seen"
            turn = next((e for e in rec["events"] if e["name"] == "turn"), None)
            start = next(e for e in rec["events"] if e["name"] == "speech-start")
            assert rec["runningAt"] is not None, (
                f"audioState never read 'running'; samples {rec['samples'][:5]}..."
            )
            running_after_click_ms = rec["runningAt"] - rec["clickAt"]
            assert running_after_click_ms >= -1, "running before the tap; the profile is not strict"
            if turn is not None:
                assert rec["runningAt"] < turn["t"], (
                    f"audioState became 'running' {rec['runningAt'] - turn['t']:.0f} ms AFTER "
                    "the server answered - the unlock did not happen in the gesture"
                )
            assert running_after_click_ms <= UNLOCK_BEFORE_RESPONSE_MS, running_after_click_ms
            assert start["audioState"] == "running", (
                f"audioState was {start['audioState']!r} at speech-start"
            )

            # The property iOS enforces: the resume() CALL was issued during the trusted
            # input's dispatch AND on its call stack, not in a continuation queued by it.
            # Inline: the page's own context. Iframe: relayed into the stage frame.
            calls = [c for g in report["gesture"].values() for c in g["resumeCalls"]]
            in_dispatch = [c for c in calls if c["inDispatch"] and c["sync"]]
            if relayed:
                near = [
                    c
                    for c in calls
                    if rec["clickWall"] - 1 <= c["wall"] <= rec["clickWall"] + RELAYED_UNLOCK_MAX_MS
                ]
                assert near, (
                    f"no AudioContext.resume() within {RELAYED_UNLOCK_MAX_MS} ms of the tap in "
                    f"any frame; calls {calls}"
                )
            else:
                assert in_dispatch, (
                    "no AudioContext.resume() was issued INSIDE the trusted input's dispatch; "
                    f"calls seen: {calls}"
                )
            return {
                "running_after_click_ms": round(running_after_click_ms),
                "response_after_click_ms": round(turn["t"] - rec["clickAt"]) if turn else None,
                "speech_start_after_click_ms": round(start["t"] - rec["clickAt"]),
                "resume_calls": len(calls),
                "resume_calls_in_dispatch": len(in_dispatch),
                "audio_duration": next(
                    (
                        e["data"]["duration"]
                        for e in rec["events"]
                        if e["name"] == "speech-end" and e["data"]
                    ),
                    None,
                ),
            }

    return Probe


@pytest.fixture
def chromium_with_audio(playwright):
    """A browser whose microphone is a WAV file.

    Returns a context-manager factory: ``with chromium_with_audio(wav) as page:``. Each
    call launches its own Chromium, because the capture file is a launch argument and
    %noloop plays it exactly once per capture stream. ``persistent=True`` reuses
    DEPLOYED_ASR_PROFILE so a test that loads the ASR model pays the download once.
    """

    @contextlib.contextmanager
    def launch(wav: Path, *, no_webgpu: bool = False, persistent: bool = False):
        args = [*AUDIO_CAPTURE_ARGS, fake_audio_arg(wav)]
        if no_webgpu:
            args.extend(NO_WEBGPU_ARGS)
        browser = None
        if persistent:
            DEPLOYED_ASR_PROFILE.mkdir(parents=True, exist_ok=True)
            context = playwright.chromium.launch_persistent_context(
                user_data_dir=str(DEPLOYED_ASR_PROFILE), args=args
            )
        else:
            browser = playwright.chromium.launch(args=args)
            context = browser.new_context()
        try:
            if no_webgpu:
                context.add_init_script(HIDE_WEBGPU)
            page = context.new_page()
            page.add_init_script(EVENT_TAP_JS)
            yield page
        finally:
            context.close()
            if browser is not None:
                browser.close()

    return launch


@pytest.fixture
def chromium_no_webgpu(chromium_with_audio):
    """The same, with WebGPU disabled by flag AND navigator.gpu removed from the page."""
    return functools.partial(chromium_with_audio, no_webgpu=True)


@pytest.fixture
def request_counter():
    """``with request_counter(page) as seen:`` - every request URL issued in the window.

    Nothing is excluded. If a replay triggers so much as a favicon fetch, the list says
    so and the assertion can then be scoped deliberately rather than silently.
    """

    @contextlib.contextmanager
    def counting(page):
        seen: list[str] = []

        def record(request):
            seen.append(request.url)

        page.on("request", record)
        try:
            yield seen
        finally:
            page.remove_listener("request", record)

    return counting


@pytest.fixture
def speech_events():
    """Read the events EVENT_TAP_JS recorded: ``[{event, t, audioDuration, data}]``.

    ``install(page)`` is only needed for pages that did not come from
    ``chromium_with_audio``, which installs the tap itself.
    """

    class Events:
        @staticmethod
        def install(page) -> None:
            page.add_init_script(EVENT_TAP_JS)

        @staticmethod
        def read(page) -> list[dict]:
            return page.evaluate("() => window.__events || []")

        @staticmethod
        def clear(page) -> None:
            page.evaluate("() => { window.__events = []; }")

        @staticmethod
        def named(page, name: str) -> list[dict]:
            return [e for e in Events.read(page) if e["event"] == name]

        @staticmethod
        def wait_for(page, name: str, *, timeout_ms: int, at_least: int = 1) -> list[dict]:
            page.wait_for_function(
                "([name, n]) => (window.__events || []).filter((e) => e.event === name)"
                ".length >= n",
                arg=[name, at_least],
                timeout=timeout_ms,
            )
            return Events.named(page, name)

    return Events


@pytest.fixture
def wait_for_avatar_ready():
    """``wait(page, url)``: load the Space and return once the avatar has RENDERED a frame.

    Retries the initial navigation for up to COLD_START_TIMEOUT_S, because a sleeping
    Space serves 503 while it is scheduled and built. Waits for ``ready`` and then for
    the first rendered frame, since ready fires before the shaders compile.
    """

    def wait(page, url: str, *, timeout_ms: int = AVATAR_READY_TIMEOUT_MS) -> dict:
        t0 = _goto_cold_start_tolerant(page, url)
        page.wait_for_selector("#vrm-stage", state="attached", timeout=STAGE_ATTACHED_TIMEOUT_MS)
        page.wait_for_function(AVATAR_READY, timeout=timeout_ms)
        ready_s = time.monotonic() - t0
        page.wait_for_function(AVATAR_FIRST_FRAME, timeout=FIRST_FRAME_TIMEOUT_MS)
        rendered_s = time.monotonic() - t0
        return {"ready_seconds": ready_s, "first_frame_seconds": rendered_s}

    return wait


# ------------------------------------------------------------------------ local plumbing


def _free_port() -> int:
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


@pytest.fixture(scope="session")
def browser_type_launch_args(browser_type_launch_args):
    """Chromium's autoplay policy leaves an ungestured AudioContext suspended forever.

    A suspended context never advances currentTime, so the viseme player would never
    tick and 'speech-end' would never fire - the test would be measuring the autoplay
    policy rather than the lip-sync. The flag is the standard way to opt out.
    """
    return {
        **browser_type_launch_args,
        "args": [
            *browser_type_launch_args.get("args", []),
            "--autoplay-policy=no-user-gesture-required",
        ],
    }


@pytest.fixture(scope="session")
def static_server() -> str:
    """Serve the repo root so /avatar/stage.html and /avatar/assets/tutor.vrm resolve.

    The vendored runtime is ``avatar/vendor/*.mjs`` and module scripts are MIME-checked
    strictly; SimpleHTTPRequestHandler defers to ``mimetypes``, which on Windows reads
    ``.mjs`` as text/plain from the registry. Registered here exactly as the app does in
    ``avatar_component.py``, so the standalone layer serves what the app serves.
    """
    mimetypes.add_type("text/javascript", ".mjs")
    handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory=str(REPO_ROOT))
    server = http.server.ThreadingHTTPServer(("127.0.0.1", _free_port()), handler)
    server.daemon_threads = True
    threading.Thread(target=server.serve_forever, daemon=True).start()
    try:
        yield f"http://127.0.0.1:{server.server_port}"
    finally:
        server.shutdown()
        server.server_close()


def _wait_for_http(url: str, proc: subprocess.Popen, timeout_s: int) -> None:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        if proc.poll() is not None:
            out = proc.stdout.read() if proc.stdout else ""
            raise RuntimeError(f"app.py exited with {proc.returncode}:\n{out}")
        try:
            with urllib.request.urlopen(url, timeout=2) as response:
                if response.status == 200:
                    return
        except (urllib.error.URLError, TimeoutError, ConnectionError, OSError):
            time.sleep(0.5)
    raise RuntimeError(f"app.py did not answer {url} within {timeout_s}s")


@pytest.fixture(scope="session")
def gradio_apps():
    """Start app.py once per AVATAR_TRANSPORT value and return its base URL.

    One env var is the entire difference between the two deployments, which is the
    property the parity suite exists to prove.
    """
    started: dict[str, tuple[str, subprocess.Popen]] = {}

    def start(transport: str) -> str:
        if transport not in started:
            port = _free_port()
            env = {
                **os.environ,
                "AVATAR_TRANSPORT": transport,
                "GRADIO_SERVER_NAME": "127.0.0.1",
                "GRADIO_SERVER_PORT": str(port),
                "GRADIO_ANALYTICS_ENABLED": "False",
            }
            proc = subprocess.Popen(  # noqa: S603
                [sys.executable, "app.py"],
                cwd=str(REPO_ROOT),
                env=env,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
            )
            url = f"http://127.0.0.1:{port}/"
            try:
                _wait_for_http(url, proc, APP_BOOT_TIMEOUT_S)
            except Exception:
                proc.kill()
                raise
            started[transport] = (url, proc)
        return started[transport][0]

    try:
        yield start
    finally:
        for _url, proc in started.values():
            proc.terminate()
            try:
                proc.wait(timeout=15)
            except subprocess.TimeoutExpired:
                proc.kill()