"""VOIC-05's measurement: N warm turns against the deployed Space, written to docs/LATENCY.md. This is a HARNESS, not an assertion suite. Its job is to produce a number with enough context to reproduce it; the only thing it fails on is not being able to produce one. 01-RESEARCH.md is emphatic that a predicted latency must not become an acceptance criterion, so there is no threshold here. If the p50 exceeds the ~2.5 s re-plan trigger (Open Question 6) the harness says so loudly, in the test output and in the document, and still passes. The numbers live in git (docs/LATENCY.md) because Space disk is ephemeral and Phase 4 owns persistence; that is option 1 of the three RESEARCH lists, chosen deliberately over a SQLite file on the Space and over a CommitScheduler telemetry stream. Method, fixed so the run is reproducible: load the Space, wait for the first rendered frame, one throwaway turn to warm the synthesiser, then N=30 typed turns cycling a fixed list of five sentences in a fixed order (the ``short`` and ``long`` fixture texts among them, so the sample is not all one size), then 5 replays. Per turn the page's own performance marks are read (``turn:dispatch`` -> ``turn:response`` -> ``turn:speech-start``) alongside the server's per-stage timings. Percentiles are nearest-rank: with N=30 the p95 method matters, and nearest-rank never invents a value that was not observed. Run:: .venv/Scripts/python.exe -m pytest tests/e2e/test_latency_harness.py -q \ --space-url https://wolfdavid-japanese-learning-avatar.hf.space A loopback ``--space-url`` (a local rehearsal) writes to pytest's tmp_path instead of docs/LATENCY.md, so a developer machine's numbers can never overwrite the Space's. """ from __future__ import annotations import math import os import platform import re import time from datetime import UTC, datetime from pathlib import Path from urllib.parse import urlparse import pytest from tests.e2e.test_avatar_loop import _controls_rearmed, _text_turn, read_debug pytestmark = [pytest.mark.deployed, pytest.mark.slow] REPO_ROOT = Path(__file__).resolve().parent.parent.parent LATENCY_MD = REPO_ROOT / "docs" / "LATENCY.md" SPACE_REPO = "WolfDavid/japanese-learning-avatar" WARM_TURNS = 30 REPLAY_TURNS = 5 # Fixed list, fixed order. Indexes 0 and 2 are replaced at run time by the short and long # fixture texts from tests/fixtures/synth_meta.json (5 and 36 moras), and the harness # asserts they are what this list says, so a fixture edit cannot silently change the # sample. The other three span the range between. SENTENCES = ( "こんにちは", "はじめまして、よろしくお願いします。", "今日はいい天気ですから、公園を散歩してから、買い物に行きました。", "駅はどこですか。", "日本語を勉強しています。", ) WARMUP_TEXT = "こんにちは" # Mora counts for the three non-fixture sentences, counted by hand (the engine's own # count for the fixture sentences comes from synth_meta.json at run time). HAND_COUNTED_MORAS = {1: "17 (hand)", 3: "8 (hand)", 4: "14 (hand)"} # The Space's CPU synthesises the 36-mora sentence in ~10 s (plan 01-09); the bounds are # wide because a slow turn is a data point here, not a failure. START_TIMEOUT_MS = 120_000 END_TIMEOUT_MS = 90_000 REARM_TIMEOUT_MS = 90_000 REPLAY_START_TIMEOUT_MS = 10_000 REPLAN_TRIGGER_MS = 2_500 VOIC05_TARGET_MS = 1_500 STAGE_KEYS = ("audio_query_ms", "synthesis_ms", "timeline_ms", "encode_ms", "server_total_ms") # The last occurrence of each turn mark. Marks accumulate across turns, so the latest of # each is this turn's, read after its speech-end. TURN_MARKS = """ () => { const last = (name) => { const e = performance.getEntriesByName(name, 'mark'); return e.length ? e[e.length - 1].startTime : null; }; return { dispatch: last('turn:dispatch'), response: last('turn:response'), speechStart: last('turn:speech-start'), replayDispatch: last('replay:dispatch'), replaySpeechStart: last('replay:speech-start'), }; } """ # Sections of docs/LATENCY.md that a HUMAN fills (plan 01-10 Task 3). When the file # already has them, they are carried forward verbatim so re-running the harness can # never erase the owner's cold-start, mobile or lip-sync answers. HUMAN_SECTIONS = ("Cold start", "Mobile", "Lip-sync verification (AVTR-02)") def nearest_rank(values: list[float], percentile: float) -> float: """Nearest-rank percentile: the ceil(P/100 * N)-th smallest observed value.""" ordered = sorted(values) rank = max(1, math.ceil(percentile / 100 * len(ordered))) return ordered[rank - 1] def summarise(values: list[float]) -> dict[str, float]: return { "p50": nearest_rank(values, 50), "p95": nearest_rank(values, 95), "min": min(values), "max": max(values), "n": len(values), } def is_public_space(space_url: str) -> bool: host = urlparse(space_url).hostname or "" return host not in {"127.0.0.1", "localhost", "::1"} def hub_metadata() -> dict[str, str]: """Revision SHA, hardware and stage from the Hub; DISABLE_GPU if a token allows.""" meta = { "sha": "unavailable (Hub API not reachable)", "hardware": "unavailable", "stage": "unavailable", "disable_gpu": "not read (no token)", } try: from huggingface_hub import HfApi api = HfApi() info = api.space_info(SPACE_REPO) meta["sha"] = info.sha or meta["sha"] runtime = getattr(info, "runtime", None) if runtime is not None: hw = getattr(runtime, "hardware", None) meta["hardware"] = str( getattr(hw, "current", None) or getattr(hw, "requested", None) or hw ) meta["stage"] = str(getattr(runtime, "stage", "unavailable")) try: variables = api.get_space_variables(SPACE_REPO) meta["disable_gpu"] = str(variables["DISABLE_GPU"].value) except Exception as err: # noqa: BLE001 - a missing token is a fact to record meta["disable_gpu"] = f"not read ({type(err).__name__})" except Exception as err: # noqa: BLE001 - recorded, not fatal meta["sha"] = f"unavailable ({type(err).__name__})" return meta def existing_sections(path: Path) -> dict[str, str]: """``heading -> body`` for every ``## `` section of an existing document.""" if not path.exists(): return {} text = path.read_text(encoding="utf-8") parts = re.split(r"^## ", text, flags=re.M) sections: dict[str, str] = {} for part in parts[1:]: heading, _, body = part.partition("\n") sections[heading.strip()] = body.rstrip("\n") return sections def row(label: str, s: dict[str, float]) -> str: return f"| {label} | {s['p50']:.0f} | {s['p95']:.0f} | {s['min']:.0f} | {s['max']:.0f} |" def render_document( *, measured_at: str, space_url: str, meta: dict[str, str], browser_version: str, client: str, load: dict[str, float], warm: list[dict], replays: list[dict], carried: dict[str, str], ) -> tuple[str, dict[str, dict[str, float]]]: stats = { "speech": summarise([t["dispatch_to_speech_ms"] for t in warm]), "response": summarise([t["dispatch_to_response_ms"] for t in warm]), "replay": summarise([r["dispatch_to_speech_ms"] for r in replays]), } for key in STAGE_KEYS: stats[key] = summarise([t["timings"][key] for t in warm]) p50 = stats["speech"]["p50"] synth_share = stats["synthesis_ms"]["p50"] / p50 * 100 if p50 else 0.0 decode_ms = stats["speech"]["p50"] - stats["response"]["p50"] trigger_fired = p50 > REPLAN_TRIGGER_MS replay_lower = stats["replay"]["p50"] < p50 lines = [ "# Turn latency (VOIC-05)", "", f"**Measured:** {measured_at} ", f"**Space:** {SPACE_REPO} **Revision:** `{meta['sha']}` " f"**URL measured:** {space_url} ", f"**Hardware:** {meta['hardware']} (stage `{meta['stage']}`) " f"**DISABLE_GPU:** {meta['disable_gpu']} ", f"**Browser:** Chromium {browser_version} (Playwright, headless) **Client:** {client} ", f"**Method:** N={len(warm)} warm text turns after 1 throwaway warm-up " f"({WARMUP_TEXT}), {len(replays)} replay turns, five sentences of 5–36 moras in a fixed " "cycle (`tests/e2e/test_latency_harness.py`). Measured from the page's own " "`performance.mark` entries; server stages from `__debug.lastStageTimings`. ", "p95 computed by **nearest-rank** (the ceil(0.95·N)-th smallest observed value; " "p50 likewise), so no percentile is an interpolated value that was never observed.", "", "Generated by the harness; the **Cold start**, **Mobile** and **Lip-sync** sections " "are the owner's and are carried forward verbatim when the harness re-runs.", "", f"## Warm turns (N={len(warm)})", "", "| Stage | p50 (ms) | p95 (ms) | min | max |", "|---|---|---|---|---|", row("dispatch -> speech-start (THE NUMBER)", stats["speech"]), row("dispatch -> response", stats["response"]), row("server: audio_query", stats["audio_query_ms"]), row("server: synthesis", stats["synthesis_ms"]), row("server: timeline", stats["timeline_ms"]), row("server: encode", stats["encode_ms"]), row("server: total", stats["server_total_ms"]), "", "Per sentence (dispatch -> speech-start, ms; each sentence was spoken " f"{len(warm) // len(SENTENCES)} times):", "", "| # | Sentence | Moras | p50 | min | max | synthesis p50 |", "|---|---|---|---|---|---|---|", ] for idx, sentence in enumerate(SENTENCES): mine = [t for t in warm if t["sentence_index"] == idx] if not mine: continue s = summarise([t["dispatch_to_speech_ms"] for t in mine]) synth = summarise([t["timings"]["synthesis_ms"] for t in mine]) moras = mine[0].get("moras", "") lines.append( f"| {idx + 1} | {sentence} | {moras} | {s['p50']:.0f} | {s['min']:.0f} | " f"{s['max']:.0f} | {synth['p50']:.0f} |" ) lines += [ "", f"## Replay (N={len(replays)}, client-side, zero network)", "", "| Stage | p50 (ms) | p95 (ms) | min | max |", "|---|---|---|---|---|", row("dispatch -> speech-start", stats["replay"]), "", ( f"Replay p50 {stats['replay']['p50']:.0f} ms vs warm-turn p50 {p50:.0f} ms: " + ( "the cache is free, the round trip is the cost." if replay_lower else "**replay is NOT faster than a turn - a real finding, recorded, not hidden.**" ) ), "", ] # Human-owned sections: carried forward if the owner has filled them, else templates. cold_default = "\n".join( [ "", "| Measurement | Value | How obtained |", "|---|---|---|", "| Space SLEEPING -> first HTTP 200 | **PENDING - Task 3 (owner)** | Pause + Restart " "(or factory rebuild - say which) from Space Settings, stopwatch from page load |", "| First turn after boot (synthesiser load dominates) | **PENDING - Task 3 (owner)** | " "first typed turn after the restart, until audible speech |", f"| Page load -> avatar `ready` (client-side, no backend) | this run: ready " f"{load['ready_seconds']:.1f} s, first rendered frame " f"{load['first_frame_seconds']:.1f} s" " (Space already RUNNING) | `wait_for_avatar_ready`; plan 01-05 recorded 4.1–8.3 s " "warm with tutor.vrm 3.0–6.8 s of it, 01-09 4.0–16.8 s; Space rebuild-to-RUNNING " "106 s (01-05); warm wake 0.2–2.7 s |", "", "Revision live during the cold-start test: **PENDING - Task 3 (owner)**.", ] ) mobile_default = "\n".join( [ "", "| Device | OS | Browser | Renders? | Observed smoothness | Push-to-talk | " "Turn latency |", "|---|---|---|---|---|---|---|", "| **PENDING - Task 3 (owner)** | | | | | | |", "", "Open the Space on a real phone via the huggingface.co Space page, not only the direct " "subdomain; record device, OS version, browser version, renders / smooth-or-slideshow, " "warmth, whether the mic permission and push-to-talk worked, and a rough turn time.", ] ) lipsync_default = "\n".join( [ "", "**PENDING - Task 3 (owner).** Date, sentence used (~20 s), duration, and the four " "judgements: (1) do あ/い/う visibly differ; (2) does the mouth close on ん, っ and " "pauses; (3) is the END of the sentence as well synced as the start (drift over the " "last third is the documented failure mode); (4) does the mouth freeze on です / した " "(devoiced vowels)? Plus a path or link to the screen recording.", ] ) defaults = { "Cold start": cold_default, "Mobile": mobile_default, "Lip-sync verification (AVTR-02)": lipsync_default, } for heading in HUMAN_SECTIONS: body = carried.get(heading) if body is None or "PENDING - Task 3" in body: body = defaults[heading] lines += [f"## {heading}", body, ""] under_target = p50 < VOIC05_TARGET_MS lines += [ "## Interpretation", "", f"- **p50 dispatch -> speech-start is {p50:.0f} ms** " f"({'under' if under_target else 'over'} the ~{VOIC05_TARGET_MS} ms VOIC-05 perceived-" f"response target; {'under' if not trigger_fired else 'OVER'} the {REPLAN_TRIGGER_MS} ms " "re-plan trigger).", ] if trigger_fired: lines += [ "", f"> **RE-PLAN TRIGGER FIRED.** p50 {p50:.0f} ms exceeds {REPLAN_TRIGGER_MS} ms. Per " '`01-RESEARCH.md` Open Question 6 - *"treat p50 > ~2.5 s as a re-plan trigger, not a ' 'phase failure"* - this is a decision for the roadmap, not a failing test: the ' 'criterion for Phase 1 is *"p50/p95 measured and recorded"*, which this file is.', ] lines += [ f"- **What dominates:** VOICEVOX synthesis on the Space's CPU - server `synthesis_ms` p50 " f"{stats['synthesis_ms']['p50']:.0f} ms is {synth_share:.0f} % of the p50 turn; " f"`audio_query` ({stats['audio_query_ms']['p50']:.1f} ms), `timeline` " f"({stats['timeline_ms']['p50']:.2f} ms) and `encode` ({stats['encode_ms']['p50']:.2f} ms) " "are noise. Network + Gradio round trip above the server total: " f"{stats['response']['p50'] - stats['server_total_ms']['p50']:.0f} ms at p50; browser " f"decode + schedule after the response: {decode_ms:.0f} ms at p50.", "- **Cheapest improvement (recorded, NOT implemented in Phase 1):** the same sentence " "synthesises 3–6x faster on a developer laptop than on the Space (plan 01-09), so the " "container's CPU share is the lever - check `cpu_num_threads` on the Space's " "`Synthesizer` and whether the ZeroGPU container's CPU allocation is the ceiling; " "beyond that, VOICEVOX CORE 0.17's streaming synthesis would let speech start before the " "whole utterance is rendered, and per-sentence synthesis of long replies would bound the " "first-sample latency by the first sentence rather than the whole reply.", "", "## Notes", "", "- Phase 1 has no LLM. These numbers are NOT predictive of Phase 3, which adds one.", "- Space sleeps after 48 h (gcTimeout 172800), so a cold start is the default first-visit " "experience.", "- Numbers live in git because Space disk is ephemeral. Continuous telemetry is " "deliberately deferred to Phase 4/6, which owns the privacy story.", "- Headless Chromium renders through SwiftShader on the client; the client-side decode + " "schedule share above is therefore an upper bound on what a GPU-rendered visitor sees.", "", "## Raw samples (dispatch -> speech-start, ms, in run order)", "", "| Turn | Sentence # | dispatch->speech | dispatch->response | synthesis | lastTurnMs |", "|---|---|---|---|---|---|", ] for i, t in enumerate(warm, start=1): lines.append( f"| {i} | {t['sentence_index'] + 1} | {t['dispatch_to_speech_ms']:.0f} | " f"{t['dispatch_to_response_ms']:.0f} | {t['timings']['synthesis_ms']:.0f} | " f"{t['last_turn_ms']} |" ) lines += [ "", "Replays (dispatch -> speech-start, ms): " + ", ".join(f"{r['dispatch_to_speech_ms']:.0f}" for r in replays), "", ] return "\n".join(lines), stats def _measure_turn(page, speech_events, text: str, index: int, moras: str) -> dict: turn = _text_turn( page, speech_events, text, start_timeout_ms=START_TIMEOUT_MS, end_timeout_ms=END_TIMEOUT_MS, ) marks = page.evaluate(TURN_MARKS) debug = read_debug(page) assert marks["dispatch"] is not None and marks["speechStart"] is not None, marks assert marks["dispatch"] < marks["response"] < marks["speechStart"], marks timings = debug["lastStageTimings"] assert timings and all(k in timings for k in STAGE_KEYS), timings _controls_rearmed(page, timeout_ms=REARM_TIMEOUT_MS) return { "sentence_index": index, "text": text, "moras": moras, "dispatch_to_speech_ms": marks["speechStart"] - marks["dispatch"], "dispatch_to_response_ms": marks["response"] - marks["dispatch"], "last_turn_ms": debug["lastTurnMs"], "timings": {k: float(timings[k]) for k in STAGE_KEYS}, "audio_duration_s": turn["duration"], "wall_speech_start_s": turn["speech_start_seconds"], } def _measure_replay(page, speech_events) -> dict: before = len(speech_events.named(page, "speech-end")) page.click("#replay-button") speech_events.wait_for( page, "speech-start", timeout_ms=REPLAY_START_TIMEOUT_MS, at_least=before + 1 ) speech_events.wait_for(page, "speech-end", timeout_ms=END_TIMEOUT_MS, at_least=before + 1) marks = page.evaluate(TURN_MARKS) debug = read_debug(page) assert marks["replayDispatch"] is not None and marks["replaySpeechStart"] is not None, marks _controls_rearmed(page, timeout_ms=REARM_TIMEOUT_MS) return { "dispatch_to_speech_ms": marks["replaySpeechStart"] - marks["replayDispatch"], "last_replay_ms": debug["lastReplayMs"], } @pytest.mark.deployed @pytest.mark.slow def test_measure_warm_turns( page, browser, space_url, synth_meta, speech_events, wait_for_avatar_ready, tmp_path ): """Produce docs/LATENCY.md. Fails only if a number cannot be produced.""" short, long = synth_meta["cases"]["short"], synth_meta["cases"]["long"] assert SENTENCES[0] == short["text"] and SENTENCES[2] == long["text"], ( "the fixed sentence list no longer matches the fixture texts" ) moras = {0: str(short["mora_count"]), 2: str(long["mora_count"]), **HAND_COUNTED_MORAS} public = is_public_space(space_url) out = LATENCY_MD if public else tmp_path / "LATENCY.md" # A local rehearsal may shorten the loop to prove the harness itself; the committed # document is always the full N, because the override is ignored for a public URL. turns = WARM_TURNS if public else int(os.environ.get("LATENCY_TURNS", WARM_TURNS)) carried = existing_sections(LATENCY_MD) measured_at = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") meta = ( hub_metadata() if public else { "sha": "local rehearsal (not the Space)", "hardware": "developer machine", "stage": "local", "disable_gpu": "1 (local env)", } ) speech_events.install(page) load = wait_for_avatar_ready(page, space_url) print( f"\n[latency] ready {load['ready_seconds']:.1f}s, " f"first frame {load['first_frame_seconds']:.1f}s" ) warmup = _measure_turn(page, speech_events, WARMUP_TEXT, 0, moras[0]) print(f"[latency] warm-up turn (discarded): {warmup['dispatch_to_speech_ms']:.0f} ms") t0 = time.monotonic() warm: list[dict] = [] for i in range(turns): index = i % len(SENTENCES) sample = _measure_turn(page, speech_events, SENTENCES[index], index, moras.get(index, "")) warm.append(sample) print( f"[latency] turn {i + 1:2d}/{turns} s{index + 1}: " f"{sample['dispatch_to_speech_ms']:.0f} ms to speech " f"(response {sample['dispatch_to_response_ms']:.0f}, " f"synthesis {sample['timings']['synthesis_ms']:.0f}, " f"lastTurnMs {sample['last_turn_ms']})" ) replays = [_measure_replay(page, speech_events) for _ in range(REPLAY_TURNS)] print( f"[latency] replays: {[round(r['dispatch_to_speech_ms']) for r in replays]} ms; " f"{turns} turns + {REPLAY_TURNS} replays in {time.monotonic() - t0:.0f}s" ) document, stats = render_document( measured_at=measured_at, space_url=space_url, meta=meta, browser_version=browser.version, client=f"{platform.system()} {platform.release()}, public internet, one visitor", load=load, warm=warm, replays=replays, carried=carried, ) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(document, encoding="utf-8", newline="\n") p50, p95 = stats["speech"]["p50"], stats["speech"]["p95"] print( f"\n[latency] ===== dispatch -> speech-start: p50 {p50:.0f} ms, p95 {p95:.0f} ms " f"(N={len(warm)}, nearest-rank); replay p50 {stats['replay']['p50']:.0f} ms; " f"synthesis p50 {stats['synthesis_ms']['p50']:.0f} ms; revision {meta['sha']} =====\n" f"[latency] written to {out}" ) if p50 > REPLAN_TRIGGER_MS: print( f"[latency] WARNING: p50 {p50:.0f} ms exceeds the {REPLAN_TRIGGER_MS} ms re-plan " "trigger (01-RESEARCH.md Open Question 6). This is a re-plan trigger, not a phase " "failure; recorded in the document." ) if stats["replay"]["p50"] >= p50: print( "[latency] NOTE: replay p50 is not lower than the warm-turn p50 - " "recorded as a finding." ) # The harness must always be able to say it produced the number. assert len(warm) == turns and len(replays) == REPLAY_TURNS assert out.exists() and "p95" in document and "THE NUMBER" in document