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


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():
    s = src("vrm-stage.js")
    assert "https://esm.sh/three@0.185.1" in s
    assert "https://esm.sh/@pixiv/three-vrm@3.5.5?deps=three@0.185.1" 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",
    ],
)
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"
        )