tests/e2e/test_latency_harness.py

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