tests/test_transport_seam.py

"""The transport seam. If the gr.HTML spike fails, only the stagePort changes.

These tests exist so that stays true as the phase progresses. Plans 01-07 and 01-08
add turn behaviour; these tests fail if it lands in a transport file instead of in
the shared turn-loop module.
"""

import re
from pathlib import Path

import pytest

# Resolved from this file rather than the process cwd: the quick loop is documented as
# running from the repo root, but a relative Path() would make that a silent
# precondition of every assertion below.
AVATAR = Path(__file__).resolve().parent.parent / "avatar"
CORE = ["vrm-stage.js", "lipsync.js", "audio-queue.js"]
SHARED = ["facade.js", "turn-loop.js"]
TRANSPORTS = ["avatar.js", "avatar-iframe.js"]
GRADIO_TOKENS = ["gradio", "gradio_api", "server.", "trigger("]
AMPLITUDE_TOKENS = ["AnalyserNode", "getByteFrequencyData", "getFloatTimeDomainData"]
TURN_SURFACE = ["startListening", "stopListening", "dispatchTurn", "requestSlower"]
# Plan 01-07. Mic capture and ASR run in the PARENT document under both transports -
# only rendering and audio playback live inside the iframe - so these modules must hang
# off the shared turn loop, never off a transport.
AUDIO_IN = ["mic.js", "asr.js"]
# Wave 5. The host glue binds the page's controls to window.Avatar. It is loaded by the
# shared boot template in avatar_component.py, so both transports get the same controls;
# it must import no avatar module, be imported by none, and implement no turn behaviour.
HOST = "host.js"
COMPONENT = AVATAR.parent / "src" / "japanese_avatar" / "ui" / "avatar_component.py"


def src(name: str) -> str:
    return (AVATAR / name).read_text(encoding="utf-8")


def avatar_surface() -> list[str]:
    """Parse AVATAR_SURFACE out of facade.js. Never duplicate the list in the test."""
    m = re.search(r"AVATAR_SURFACE\s*=\s*\[(.*?)\]", src("facade.js"), re.S)
    assert m, "AVATAR_SURFACE not found in avatar/facade.js"
    return re.findall(r"'([a-zA-Z]+)'", m.group(1))


@pytest.mark.parametrize("name", CORE)
def test_stage_core_is_gradio_free(name):
    lowered = src(name).lower()
    for tok in GRADIO_TOKENS:
        assert tok.lower() not in lowered, (
            f"{name} references {tok!r}; the stage core must stay transport-agnostic"
        )


@pytest.mark.parametrize("name", CORE)
def test_no_amplitude_lipsync(name):
    for tok in AMPLITUDE_TOKENS:
        assert tok not in src(name), f"{name} uses {tok!r}; AVTR-02 disqualifies amplitude lip-sync"


@pytest.mark.parametrize("name", SHARED)
def test_shared_modules_are_gradio_free(name):
    lowered = src(name).lower()
    for tok in GRADIO_TOKENS:
        assert tok.lower() not in lowered, f"{name} must not know about Gradio"


def test_lipsync_uses_audiocontext_clock():
    s = src("lipsync.js")
    assert "currentTime" in s
    assert "setTimeout" not in s
    assert "requestAnimationFrame" not in s  # the stage owns the loop; the player is a tick fn


def test_module_urls_are_deps_pinned():
    """The three runtime modules are the VENDORED, version-pinned builds (plan 01-10).

    Until 01-10 this asserted the esm.sh ``?deps=three@0.185.1`` URLs; the pin now lives
    in scripts/vendor_modules.py and the stage imports the files it wrote. Every one of
    the three must resolve relative to the module, so that the app
    (/gradio_api/file=avatar/...) and the standalone stage.html share one rule.
    """
    s = src("vrm-stage.js")
    for name in ("three.mjs", "GLTFLoader.mjs", "three-vrm.mjs"):
        assert f"resolveModuleUrl('./vendor/{name}')" in s, f"{name} is not vendored-resolved"
    assert "esm.sh/" not in s, "a CDN URL is back in the render path"
    assert "import.meta.url" in s
    assert "importmap" not in s


def test_only_facade_assigns_window_avatar():
    """One facade builder. Two transports cannot drift if only one of them writes the object."""
    for p in AVATAR.glob("*.js"):
        assigns = re.search(r"window\.Avatar\s*=", p.read_text(encoding="utf-8"))
        if p.name == "facade.js":
            assert assigns, "facade.js must be the one that installs window.Avatar"
        else:
            assert not assigns, f"{p.name} assigns window.Avatar; only facade.js may"


@pytest.mark.parametrize("name", TRANSPORTS)
def test_transports_only_do_plumbing(name):
    """A transport builds a stagePort and delegates. It never implements turn behaviour."""
    s = src(name)
    assert "installFacade" in s and "createTurnLoop" in s, (
        f"{name} must use the shared facade/turn loop"
    )
    for member in TURN_SURFACE:
        assert f"function {member}" not in s and f"{member}(text" not in s, (
            f"{name} implements {member}; turn behaviour belongs in avatar/turn-loop.js "
            "or the iframe fallback silently loses it"
        )


@pytest.mark.parametrize(
    "member",
    [
        "mount",
        "speak",
        "replay",
        "setListening",
        "setThinking",
        "on",
        "getDebug",
        "startListening",
        "stopListening",
        "dispatchTurn",
        "requestSlower",
        "unlockAudio",
    ],
)
def test_surface_member_is_declared_once(member):
    """Every promised method must resolve in facade.js's surface list."""
    assert member in avatar_surface()


@pytest.mark.parametrize("member", TURN_SURFACE)
def test_turn_surface_lives_in_the_shared_module(member):
    assert member in src("turn-loop.js"), (
        f"{member} must be defined in avatar/turn-loop.js so BOTH transports get it"
    )


@pytest.mark.parametrize(
    "key",
    [
        "ready",
        "mountCount",
        "vrmMetaTitle",
        "threeInstanceCount",
        "currentVisemes",
        "blinkValue",
        "clockOffset",
        "transport",
    ],
)
def test_debug_contract_keys_present(key):
    joined = src("facade.js") + src("turn-loop.js") + src("vrm-stage.js")
    assert key in joined, f"__debug.{key} is required by 01-VALIDATION.md Observable Signals"


@pytest.mark.parametrize("name", AUDIO_IN)
def test_audio_input_modules_are_gradio_free(name):
    lowered = src(name).lower()
    for tok in GRADIO_TOKENS:
        assert tok.lower() not in lowered, (
            f"{name} references {tok!r}; mic capture and ASR must stay host-agnostic"
        )


@pytest.mark.parametrize("name", AUDIO_IN)
def test_audio_input_modules_hang_off_the_turn_loop(name):
    """Imported by turn-loop.js, by neither transport.

    If a transport imported these directly, the other transport would silently lose
    push-to-talk - the exact failure mode the shared turn loop exists to prevent.
    """
    assert f"./{name}" in src("turn-loop.js"), (
        f"avatar/turn-loop.js must import {name} so BOTH transports get it at once"
    )
    for transport in TRANSPORTS:
        assert name not in src(transport), (
            f"{transport} imports {name}; mic/ASR wiring belongs in avatar/turn-loop.js"
        )


@pytest.mark.parametrize("name", AUDIO_IN)
def test_audio_input_modules_do_not_do_amplitude_lipsync(name):
    """RMS in mic.js gates ASR. It must never become a lip-sync source (AVTR-02)."""
    for tok in AMPLITUDE_TOKENS:
        assert tok not in src(name), f"{name} uses {tok!r}; AVTR-02 disqualifies amplitude lip-sync"


def test_blocklist_does_not_swallow_ordinary_japanese():
    """The bare polite form is a thing a learner says out loud.

    Blocking it would be worse than the hallucination it prevents, and this guard exists
    because that entry is exactly what a later "helpful" edit would add.
    """
    entries = re.findall(r"^\s*'([^']+)',\s*$", src("mic.js"), re.M)
    assert "ご視聴ありがとうございました" in entries, (
        "the subtitle-boilerplate blocklist is missing its canonical entry"
    )
    assert "ありがとうございました" not in entries, (
        "the bare polite form is blocklisted; it is ordinary Japanese, not a hallucination"
    )


def test_gate_thresholds_match_the_measured_fixtures():
    """The three constants the gate lives or dies by, pinned to docs/ASR-TIERS.md.

    Measured: cafe_noise_30s.wav modulates 1.957, speech_ja.wav modulates 10.716. A
    threshold above ~10 would reject real speech; below ~2.0 would pass the noise fixture.
    """
    s = src("mic.js")
    assert "MIN_DURATION_MS: 300" in s
    assert "MIN_RMS: 0.01" in s
    assert "MIN_MODULATION: 2.5" in s
    assert "BLOCKLIST_MAX_MS: 1500" in s
    assert "REARM_TAIL_MS = 200" in s


def test_no_remote_asr_path_exists():
    """Tier D would burn the exact visitor GPU quota SC-4 tests the loop's survival
    without, so no hosted transcription path may exist anywhere in avatar/."""
    for name in ["asr.js", "mic.js", "turn-loop.js"]:
        lowered = src(name).lower()
        for tok in ["zerogpu", "spaces.gpu", "/api/asr", "@spaces"]:
            assert tok not in lowered, (
                f"{name} references {tok!r}; browser ASR must cost zero GPU quota"
            )


def test_asr_module_url_is_pinned():
    s = src("asr.js")
    assert s.count("https://esm.sh/@huggingface/transformers@4.2.0") == 1, (
        "exactly one pinned runtime URL, same discipline as vrm-stage.js"
    )
    assert "importmap" not in s


def test_asr_webgpu_fallback_is_real_not_aspirational():
    """The fallback must probe for an ADAPTER, not merely for navigator.gpu.

    Measured on this stack: when navigator.gpu exists but requestAdapter() resolves to
    null, calling pipeline(..., {device:'webgpu'}) throws AND poisons the ONNX Runtime
    backend registry for the whole page, so the subsequent WASM re-instantiation fails
    with the same WebGPU error. Catch-and-retry alone is therefore not a working
    fallback; the doomed call must never be made. Deleting this assertion re-introduces
    a bug that only shows up on the machines the fallback exists for.
    """
    s = src("asr.js")
    assert "navigator.gpu" in s
    assert "webgpu" in s
    assert "requestAdapter" in s, (
        "avatar/asr.js checks navigator.gpu but never requests an adapter; that is the "
        "exact naive check that ships a broken app to every machine without a GPU"
    )
    assert re.search(r"catch\s*\(", s), "no catch around the WebGPU init path"
    assert s.count("language: 'ja'") == 1


@pytest.mark.parametrize("key", ["asrTier", "asrModel", "lastTranscript", "micRejectedCount"])
def test_push_to_talk_debug_keys_present(key):
    assert key in src("turn-loop.js"), (
        f"__debug.{key} is contributed by avatar/turn-loop.js, so both transports report it"
    )


@pytest.mark.parametrize("member", ["startListening", "stopListening"])
def test_push_to_talk_is_implemented_not_deferred(member):
    """The two methods this plan owns must no longer be notWiredYet stubs."""
    s = src("turn-loop.js")
    assert f"notWiredYet('{member}'" not in s, (
        f"{member} is still a deferred stub in avatar/turn-loop.js"
    )
    assert f"async {member}(" in s, f"{member} must be implemented in avatar/turn-loop.js"


@pytest.mark.parametrize("name", TRANSPORTS)
def test_transports_gained_push_to_talk_without_gaining_code(name):
    """The whole point of the seam: the iframe fallback got this feature for free."""
    s = src(name)
    for token in ["startListening", "stopListening", "mic.", "asr.", "createMic", "createAsr"]:
        assert token not in s, (
            f"{name} mentions {token!r}; push-to-talk must live only in avatar/turn-loop.js"
        )


@pytest.mark.parametrize("member", ["dispatchTurn", "requestSlower"])
def test_turn_dispatch_is_implemented_not_deferred(member):
    """The wave-5 methods are real, and nothing in the file is a deferred stub any more."""
    s = src("turn-loop.js")
    assert "notWiredYet" not in s, "a deferred stub survives in avatar/turn-loop.js"
    assert f"{member}(" in s, f"{member} must be implemented in avatar/turn-loop.js"


def test_thinking_engages_before_the_first_await():
    """VOIC-05's cheapest latency mitigation: the pose switches at dispatch, not at response.

    Verified by line number inside dispatchTurn: setThinking(true) precedes the first
    await, which is the server call.
    """
    lines = src("turn-loop.js").splitlines()
    start = next(i for i, line in enumerate(lines) if "async function dispatchTurn(" in line)
    body = lines[start:]
    think_at = next(i for i, line in enumerate(body) if "setThinking(true)" in line)
    await_at = next(i for i, line in enumerate(body) if "await " in line)
    assert think_at < await_at, (
        f"setThinking(true) is on dispatchTurn line {think_at} but the first await is on "
        f"line {await_at}; the thinking pose must engage before the server round trip"
    )
    assert "await bridge." in body[await_at], body[await_at]


def test_replay_is_networkless_by_construction():
    """replay() re-emits the cached buffer. No fetch, no dynamic import, anywhere in the loop."""
    s = src("turn-loop.js")
    assert "fetch(" not in s
    assert "import(" not in s
    assert "XMLHttpRequest" not in s


def test_turn_loop_publishes_the_latency_numbers():
    s = src("turn-loop.js")
    for key in ("lastTurnMs", "lastStageTimings", "lastSubtitle", "replayCount", "turnCount"):
        assert f"{key}:" in s, f"__debug.{key} is not initialised in avatar/turn-loop.js"
    for mark in ("turn:dispatch", "turn:response", "turn:speech-start"):
        assert f"performance.mark('{mark}')" in s, f"performance.mark({mark!r}) missing"
    assert "0.75" in s, "the slower speed must be the VOICEVOX speedScale 0.75"
    assert "greeting" in s


def test_host_glue_is_neither_a_transport_nor_the_turn_loop():
    """host.js is page glue: loaded from the shared boot template, importing nothing here."""
    host = src(HOST)
    assert not re.search(r"^\s*import\s", host, re.M), (
        "host.js must not import any avatar module; it is handed the facade"
    )
    for member in TURN_SURFACE:
        assert f"function {member}" not in host and f"async {member}(" not in host, (
            f"host.js implements {member}; turn behaviour belongs in avatar/turn-loop.js"
        )
    for member in ["dispatchTurn", "requestSlower", "replay", "startListening", "stopListening"]:
        assert f".{member}(" in host, f"host.js never calls Avatar.{member}()"
    for name in [*TRANSPORTS, *SHARED, *CORE, *AUDIO_IN]:
        assert HOST not in src(name), f"{name} references {HOST}; only the boot template may"
    component = COMPONENT.read_text(encoding="utf-8")
    assert component.count(f"import('/gradio_api/file=avatar/{HOST}')") == 1, (
        "the shared boot template must load host.js exactly once, for both transports"
    )
    assert "bindHost(avatar, document)" in component


# ------------------------------------------------------------------ plan 01-11: audio unlock
#
# The owner heard nothing on a phone. The only resume() on the turn path lived in
# playBuffer, AFTER the server round trip and therefore outside the tap's user activation;
# a gesture-gated browser leaves the context suspended, and every test had been masking it
# with --autoplay-policy=no-user-gesture-required. These guards pin the fix statically; the
# strict-policy browser tests prove it behaviourally at three layers.

# The four turn-loop methods a gesture reaches, by the signature line each starts on.
GESTURE_ENTRY_POINTS = [
    "async function dispatchTurn(",
    "async replay()",
    "async requestSlower()",
    "async startListening()",
]
UNLOCK_CALL = "stagePort.unlockAudio()"


def _function_body(source: str, signature: str) -> list[str]:
    """The lines of the function whose signature line contains ``signature``, from the
    signature to its closing brace, found by brace counting from the signature's ``{``."""
    lines = source.splitlines()
    start = next(i for i, line in enumerate(lines) if signature in line)
    depth = 0
    opened = False
    for i in range(start, len(lines)):
        depth += lines[i].count("{") - lines[i].count("}")
        if "{" in lines[i]:
            opened = True
        if opened and depth == 0:
            return lines[start : i + 1]
    raise AssertionError(f"unbalanced braces after {signature!r}")


def _statements(body: list[str]) -> list[str]:
    """The body's lines after the signature, minus blank lines and comment-only lines."""
    out = []
    for line in body[1:]:
        stripped = line.strip()
        if not stripped or stripped.startswith(("//", "/*", "*")):
            continue
        out.append(stripped)
    return out


@pytest.mark.parametrize("name", TRANSPORTS)
def test_stage_port_unlocks_audio_under_both_transports(name):
    """unlockAudio is a stagePort method on BOTH transports, and the iframe stage answers
    the message the iframe transport sends for it."""
    assert "unlockAudio" in src(name), f"{name} builds a stagePort with no unlockAudio"
    assert "unlockAudio" in avatar_surface(), "unlockAudio is not in AVATAR_SURFACE"
    assert "'avatar:unlockAudio'" in src("avatar-iframe.js")
    assert "'avatar:unlockAudio'" in src("stage.html"), (
        "avatar/stage.html has no handler for the iframe transport's unlock message"
    )


def test_inline_unlock_is_synchronous():
    """The inline transport must CALL resume() synchronously, never await it: the call has
    to happen in the gesture's stack, and a blocked resume() promise never settles."""
    body = _function_body(src("avatar.js"), "unlockAudio()")
    joined = "\n".join(body)
    assert "audioCtx.resume()" in joined
    assert "await" not in joined, "unlockAudio must not await; the CALL is the unlock"
    assert "return audioCtx.state" in joined


@pytest.mark.parametrize("signature", GESTURE_ENTRY_POINTS)
def test_turn_loop_unlocks_audio_before_its_first_await(signature):
    """Line-order proof, in the spirit of test_thinking_engages_before_the_first_await:
    the unlock is the FIRST statement of every gesture-reachable entry point, so it runs
    inside the gesture's call stack - anything after an await does not."""
    body = _function_body(src("turn-loop.js"), signature)
    statements = _statements(body)
    assert statements and statements[0].startswith(UNLOCK_CALL), (
        f"{signature} does not start with {UNLOCK_CALL}; first statement is {statements[:1]}"
    )
    awaits = [i for i, line in enumerate(statements) if "await " in line]
    if awaits:
        assert awaits[0] > 0, f"{signature} awaits before it unlocks"


def test_turn_loop_unlock_count():
    assert src("turn-loop.js").count(UNLOCK_CALL) >= 4


def test_host_unlocks_audio_in_every_gesture_handler():
    """Belt and braces: the host unlocks first thing in all six gesture handlers, so a
    gesture that never reaches the loop (Send with an empty box) still unlocks."""
    host = src(HOST)
    assert host.count("avatar.unlockAudio()") >= 6, (
        f"host.js calls avatar.unlockAudio() {host.count('avatar.unlockAudio()')} time(s); "
        "every gesture handler (send, Enter, hello, replay, slower, ptt) must call it first"
    )
    for handler in ("'click'", "'keydown'", "'pointerdown'"):
        assert handler in host


def test_audio_queue_refuses_to_fake_speech_on_a_stopped_clock():
    """playBuffer must check the context is running BEFORE it emits speech-start, and must
    report the blocked state as an 'error' rather than scheduling against a dead clock."""
    s = src("audio-queue.js")
    assert "state !== 'running'" in s
    assert s.index("state !== 'running'") < s.index("emit('speech-start'"), (
        "audio-queue.js emits speech-start before it checks audioCtx.state"
    )
    assert "where: 'audio'" in s
    # A blocked resume() promise never settles in Chromium; the wait must be bounded.
    assert "Promise.race" in s and "RESUME_GRACE_MS" in s


@pytest.mark.parametrize("name", ["avatar.js", "stage.html", "facade.js"])
def test_audio_state_is_published(name):
    """audioState is a published field on getDebug() from both transports and the
    standalone harness, so the strict-policy tests assert a number, not a silence."""
    assert "audioState" in src(name), f"{name} does not publish audioState"