"""The two London players share one watch-page contract.

The catalog opens poc/interactive-live.html (command lane) or
poc/video-player.html (H.264 lane) inside the same iframe and with the
same postMessage protocol; the header must offer the same controls. These
tests pin that contract at the source level (the browser canary covers
the runtime).
"""
import pathlib
import re
import unittest

ROOT = pathlib.Path(__file__).resolve().parent
LIVE = (ROOT / "poc" / "interactive-live.html").read_text(encoding="utf-8")
VIDEO = (ROOT / "poc" / "video-player.html").read_text(encoding="utf-8")


def header(page: str) -> str:
    return page[page.index('<header id="hdr">'):page.index("</header>")]


def buttons(block: str):
    """(id, title, glyph) for every header button, in document order."""
    return re.findall(
        r'<button[^>]*\bid="([a-z]+)"[^>]*\btitle="([^"]*)"[^>]*>([^<]*)</button>',
        block)


class HeaderParityTests(unittest.TestCase):
    def test_both_watch_pages_use_the_canonical_site_icon_bundle(self):
        origin = "https://war3replays.com/"
        for filename in (
                "favicon.svg?v=1", "favicon-48x48.png?v=1",
                "favicon-32x32.png?v=1", "favicon-16x16.png?v=1",
                "favicon.ico?v=1", "apple-touch-icon.png?v=1",
                "safari-pinned-tab.svg?v=1", "site.webmanifest?v=1"):
            with self.subTest(filename=filename):
                self.assertIn(origin + filename, LIVE)
                self.assertIn(origin + filename, VIDEO)

    def test_same_controls_in_the_same_order_as_the_live_player(self):
        live = buttons(header(LIVE))
        video = buttons(header(VIDEO))
        # The match-overlay button is injected at runtime by cast-overlay.js
        # (prepended into .controls) on both pages.
        self.assertEqual(video, live)
        self.assertEqual(["back", "backstream", "spoilbtn", "wheelbtn", "mute",
                          "fs", "watchalone"],
                         [b[0] for b in live])
        # The stream buttons stay hidden until their mode applies.
        for page in (LIVE, VIDEO):
            self.assertIn('id="backstream" title="back to replay stream"', page)
            self.assertIn('id="watchalone" title="watch on your own"', page)
            self.assertEqual(2, len(re.findall(
                r'id="(?:backstream|watchalone)"[^>]*\bhidden>', page)))
        self.assertIn('<button class="ctl" id="fs" title="fullscreen">⛶</button>', LIVE)
        self.assertNotIn(">Fullscreen</button>", LIVE)
        self.assertNotIn('class="brand"', LIVE)

    def test_title_stats_and_volume_match(self):
        for needle in ('<span id="ttl">war3replays <small>/ watch</small></span>',
                       '<span class="stat hud-stat">rtt <b id="rtt">–</b></span>',
                       '<span class="stat hud-stat">fps <b id="fps">–</b></span>',
                       'id="volume" type="range" min="0" max="100" step="1" value="50"'):
            self.assertIn(needle, LIVE)
            self.assertIn(needle, VIDEO)
        self.assertIn('controls:document.querySelector("header .controls")', LIVE)
        self.assertIn('$("fps").textContent=drawnFps.toFixed(0);', LIVE)
        self.assertIn('document.title="war3replays";', LIVE)

    def test_developer_hud_stays_hidden_by_default(self):
        self.assertIn('document.querySelector("header .metrics").style.display="none";', LIVE)
        self.assertIn('<div class="metrics">', header(LIVE))
        for page in (LIVE, VIDEO):
            self.assertIn('.hud-stat { display:none; }', page)
            self.assertIn('html[data-hud="on"] .hud-stat { display:inline; }', page)


class RttPolicyTests(unittest.TestCase):
    def test_rtt_is_telemetry_and_never_rejects_a_london_session(self):
        self.assertNotIn("LowLatencyAdmission", LIVE)
        self.assertNotIn("maxRttMs", LIVE)
        self.assertNotIn("No nearby replay server", LIVE)
        self.assertNotIn('status("latency too high")', LIVE)
        self.assertIn(
            '$("rtt").textContent=Math.round(pair.currentRoundTripTime*1000);',
            LIVE)
        self.assertIn("rtt:measuredRtt,receiveBps,renderFps,p95GapMs,", LIVE)


RELAY = (ROOT / "native" / "w3cs_webrtc_relay.cpp").read_text(encoding="utf-8")
RENDER_WORKER = (ROOT / "web" / "render-worker.js").read_text(encoding="utf-8")


class CommandBootstrapOrderingTests(unittest.TestCase):
    """The relay can emit its first snapshot as soon as a session is selected.
    Every receive path must bind before the page sends that selection."""

    def test_session_starts_after_the_consumer_is_ready(self):
        self.assertEqual(1, LIVE.count('type:"session",replay:replayId'))
        self.assertIn("result.startSession=makeSessionStarter();", LIVE)
        self.assertIn('case "started": startSession(); break;', LIVE)
        self.assertIn("if(client)startSession();", LIVE)

    def test_worker_captures_transferred_channels_immediately(self):
        self.assertIn("attachBufferedReceiveChannel(channel, name", RENDER_WORKER)
        self.assertIn("bufferBeforeListener: name === \"reliable\" || "
                      "name === \"recovery\"", RENDER_WORKER)
        self.assertIn("discardRealChannel(name);", RENDER_WORKER)

    def test_early_worker_failure_waits_for_the_page_dispatcher(self):
        self.assertIn("const workerEarlyMessages=[];", LIVE)
        self.assertIn("else if(workerEarlyMessages.length<64)", LIVE)
        self.assertIn("workerEarlyMessages.splice(0)", LIVE)

    def test_reference_websockets_ack_every_binary_plane(self):
        helper = LIVE[LIVE.index("const acknowledgeWebSocketBytes="):]
        helper = helper[:helper.index("const makeSessionStarter=")]
        self.assertIn('socket.addEventListener("message",event=>{', helper)
        self.assertIn("socket.send(String(bytes));", helper)
        connect = LIVE[LIVE.index("const connectWebSockets=async()=>{"):]
        connect = connect[:connect.index("const connectWebRtc=async()=>{")]
        for path, dataset in (("resource", "resourceAcked"),
                              ("frame", "frameAcked"),
                              ("recovery", "recoveryAcked")):
            self.assertIn(
                f'openPlaneSocket("{path}","{path}","{dataset}")',
                connect)
        self.assertIn(
            "const socket=acknowledgeWebSocketBytes(new WebSocket(", helper)
        self.assertIn("const PLANE_OPEN_TIMEOUT_MS=5000;", helper)
        self.assertIn(
            "await opened(socket,name,{timeoutMs:PLANE_OPEN_TIMEOUT_MS});",
            helper)
        self.assertIn("try{socket.close();}catch{}", helper)

    def test_reference_planes_open_only_after_signaling_authenticates(self):
        connect = LIVE[LIVE.index("const connectWebSockets=async()=>{"):]
        connect = connect[:connect.index("const connectWebRtc=async()=>{")]
        signaling_open = connect.index('await opened(signaling,"signaling");')
        resource_constructed = connect.index(
            'openPlaneSocket("resource","resource","resourceAcked")')
        plane_opens = connect.index(
            'const openedPlanes=await Promise.allSettled([')
        self.assertLess(signaling_open, plane_opens)
        self.assertLess(plane_opens, resource_constructed)
        before_signaling = connect[:signaling_open]
        self.assertNotIn('socket("resource")', before_signaling)
        self.assertNotIn('socket("frame")', before_signaling)
        self.assertNotIn('socket("recovery")', before_signaling)

    def test_reference_websocket_mode_reports_its_real_quality_path(self):
        connect = LIVE[LIVE.index("const connectWebSockets=async()=>{"):]
        connect = connect[:connect.index("const connectWebRtc=async()=>{")]
        self.assertIn(
            'document.documentElement.dataset.framePlane="websocket";',
            connect)
        self.assertIn(
            'document.documentElement.dataset.frameTransport="websocket";',
            connect)
        self.assertIn("Promise.allSettled", connect)
        self.assertIn('if(result.status==="fulfilled")try{result.value.close();}',
                      connect)
        interval = LIVE[LIVE.index("setInterval(async()=>{"):]
        self.assertIn(
            'document.documentElement.dataset.frameTransport!=="websocket"',
            interval)

    def test_reference_mode_stays_explicit_and_auto_keeps_webrtc_fallback(self):
        self.assertIn(
            'const transportMode=params.get("transport")||"webrtc";', LIVE)
        chooser = LIVE[LIVE.index("const connectWithSeat=async()=>{"):]
        chooser = chooser[:chooser.index("// This block must sit BEFORE")]
        self.assertIn('return transportMode==="ws"\n'
                      '        ?await connectWebSockets():await connectWebRtc();',
                      chooser)

    def test_relay_drain_accepts_selected_websocket_planes(self):
        readiness = RELAY[RELAY.index(
            "constexpr bool command_sender_planes_ready("):]
        readiness = readiness[:readiness.index("uint32_t read_u32(")]
        self.assertIn("(frame_websocket_selected && frame_websocket_open)",
                      readiness)
        drain = RELAY[RELAY.index("bool drain() {"):]
        drain = drain[:drain.index("PacketBatch frame;")]
        self.assertIn("command_sender_planes_ready(", drain)
        self.assertIn("frame_ws_preferred_, websocket_open(frame_ws_)", drain)
        self.assertNotIn("!channel_is_open(frame_dc_)", drain)


class InjectorLifecycleTests(unittest.TestCase):
    """A CRIU claim owns its Xvfb. Detach Xlib before the session TERM
    trap destroys that server, and never reuse a Display after fatal I/O."""

    def test_claim_display_is_detached_before_session_term(self):
        body = RELAY[RELAY.index("  void stop_session() {"):]
        body = body[:body.index("  void release_resource_cache() {")]
        self.assertLess(body.index("injector_.reopen(display_name_);"),
                        body.index("kill(-session_pid_, SIGTERM);"))
        after_term = body[body.index("kill(-session_pid_, SIGTERM);"):]
        self.assertNotIn("injector_.reopen(display_name_);", after_term)

    def test_fatal_display_io_disables_all_later_xlib_cleanup(self):
        injector = RELAY[RELAY.index("class InputInjector {"):]
        injector = injector[:injector.index("struct NativeKey {")]
        self.assertIn("self->display_lost_ = true;", injector)
        self.assertIn("bool usable() const { return display_ && !display_lost_; }",
                      injector)
        self.assertIn("if (!display_lost_) release_arrows();", injector)
        self.assertIn("if (!display_lost_) XCloseDisplay(previous);", injector)
        # Teardown uses keycodes cached while the display is healthy. It must
        # not enter XKB mapping code after the server can have disappeared.
        release = injector[injector.index("  void release_arrows() {"):]
        release = release[:release.index("  void add_geometry(")]
        self.assertIn("for (const KeyCode code : arrow_keycodes_)", release)
        self.assertNotIn("XKeysymToKeycode", release.split("*/", 1)[1])
        self.assertIn("index < arrows.size() && usable()", injector)
        self.assertIn("if (!usable()) break;", injector)

    def test_datachannel_close_cannot_touch_xlib_from_gstreamer(self):
        body = RELAY[RELAY.index("static gboolean channel_closed_main("):]
        body = body[:body.index("static void control_message(")]
        self.assertIn("self->deactivate_viewer_stream();", body)
        self.assertIn("self->stop_session();", body)
        callback = body[body.index("static void channel_closed("):]
        self.assertIn("g_idle_add(channel_closed_main, user_data);", callback)
        self.assertNotIn("deactivate_viewer_stream", callback)
        self.assertNotIn("stop_session", callback)


class LaunchFailureHandoffTests(unittest.TestCase):
    """A London launch that fails before gameplay must reach the viewer's
    page and hand the watch to the video player (seen live 2026-08-22: a
    sha outside the verified catalog left the page on "Starting the game
    engine..." for good and the seat busy until the viewer left)."""

    def test_relay_control_messages_fall_back_to_signaling(self):
        body = RELAY[RELAY.index("void send_interactive_control("):]
        body = body[:body.index("void send_cursor(")]
        self.assertIn("if (control_via_signal_ || !channel_is_open(control_lane()))", body)

    def test_relay_reports_exit_status_and_signals_a_failed_session(self):
        body = RELAY[RELAY.index("void reap_finished_session()"):]
        body = body[:body.index("bool input_boost_active()")]
        self.assertIn('"{\\"t\\":\\"sessionEnded\\",\\"status\\":" +', body)
        self.assertIn("if (exit_code != 0)", body)
        self.assertIn('signal_error("WC3 session ended with status "', body)

    def test_session_end_before_gameplay_hands_off(self):
        handler = LIVE[LIVE.index('message.t==="sessionEnded"'):]
        handler = handler[:handler.index('message.t==="engineStalled"')]
        self.assertIn("if(handOffToFallback(`session-ended:${exitStatus}`))return;", handler)
        self.assertIn("WC3 session ended before gameplay (status ${exitStatus})", handler)
        self.assertIn('finishReplay("session-ended");', handler)

    def test_launch_watchdog_hands_off_when_no_frame_arrives(self):
        self.assertIn('params.has("launchTimeout")', LIVE)
        self.assertIn(":(spectatorMode?60000:45000);", LIVE)
        self.assertIn("if(visibleGameFrame||gameplayReached)return;", LIVE)
        self.assertIn('handOffToFallback(spectatorMode?"spectate-timeout":"launch-timeout")', LIVE)

    def test_relay_errors_after_the_ended_card_are_ignored(self):
        body = LIVE[LIVE.index("const failed = error => {"):]
        body = body[:body.index('$("retry").onclick')]
        self.assertIn('if(document.body.classList.contains("ended"))return;', body)
        self.assertIn('handOffToFallback((spectatorMode?"spectate-failed:":"session-failed:")', body)


class LiveWatchReportingTests(unittest.TestCase):
    """The host page lists this watch on the homepage once it knows the
    worker binding and the first game frame (WHOS-WATCHING-NOW-SPEC.md)."""

    def test_binding_is_reported_on_connect_token_and_gameplay(self):
        self.assertIn('const streamBinding=()=>({type:"w3-stream-info",feeder:"london",', LIVE)
        self.assertIn("host:location.hostname,seat:activeSeat,session:sessionId,", LIVE)
        self.assertIn("spectateToken:hostSpectateToken,profile:gameProfile,", LIVE)
        self.assertIn("  if(!spectatorMode&&activeSeat>=1)postToParent(streamBinding());", LIVE)
        self.assertIn("  bindControl(control);\n  postStreamInfo();", LIVE)
        gameplay = LIVE[LIVE.index("if(s.frame>=400 && s.draws>100){"):]
        gameplay = gameplay[:gameplay.index("configureReplayView();")]
        self.assertLess(gameplay.index("gameplayReached=true;"),
                        gameplay.index('send({t:"gameplayReady"'))
        self.assertLess(gameplay.index('send({t:"gameplayReady"'),
                        gameplay.index("postStreamInfo();"))
        handler = LIVE[LIVE.index('message.t==="spectate"'):]
        handler = handler[:handler.index('message.t==="engineStalled"')]
        self.assertIn('hostSpectateToken=String(message.token||"");', handler)
        self.assertIn("postStreamInfo();", handler)

    def test_position_posts_follow_the_cast_clock(self):
        self.assertIn('postToParent({type:"w3-position",game:`arc${replayId}`,', LIVE)
        self.assertIn("sec:castOverlay.gameMs()/1000});", LIVE)
        self.assertIn("},5000);", LIVE)

    def test_back_to_stream_button_needs_the_from_parameter(self):
        self.assertIn('const cameFromStream=/^[0-9a-f]{16}$/.test(params.get("from")||"")', LIVE)
        self.assertIn('$("backstream").hidden=false;', LIVE)
        self.assertIn('postToParent({type:"w3-back-to-stream",', LIVE)
        self.assertIn("location.href=`https://war3replays.com/stream/${cameFromStream}`;", LIVE)


class CriticalControlMultipathTests(unittest.TestCase):
    """Recovery lifecycle messages must escape a stalled commands SCTP lane.

    The render worker can acknowledge the first recovery while the module is
    suspended in connectWithSeat(). These source contracts complement the live
    browser canary by pinning the ordering and every independent control lane.
    """

    def test_multipath_sender_exists_before_connection_can_start_streaming(self):
        self.assertLess(LIVE.index("send = event => {"),
                        LIVE.index("await connectWithSeat());"))
        sender = LIVE[LIVE.index("send = event => {"):
                      LIVE.index("// The host page reveals this iframe")]
        self.assertIn('control.send(JSON.stringify(outbound));', sender)
        self.assertIn(
            'signaling.send(JSON.stringify({type:"control",event:outbound}));',
            sender)
        self.assertIn(
            'renderWorker.postMessage({type:"wtInput",json:JSON.stringify(outbound),',
            sender)
        # Each lane is isolated. A throwing DataChannel cannot prevent the
        # following signaling attempt.
        self.assertGreaterEqual(sender.count("try{"), 3)

    def test_worker_readiness_messages_use_the_common_multipath_sender(self):
        dispatch = LIVE[LIVE.index('case "control":'):
                        LIVE.index('case "cursor":',
                                   LIVE.index('case "control":'))]
        self.assertIn('?JSON.parse(data.payload):data.payload;', dispatch)
        self.assertIn('!Array.isArray(event))send(event);', dispatch)
        self.assertNotIn('control.send(data.payload)', dispatch)

    def test_direct_renderer_readiness_uses_the_common_multipath_sender(self):
        proxy = LIVE[LIVE.index("const clientControlProxy={"):]
        proxy = proxy[:proxy.index("const client = renderWorker?null")]
        self.assertIn('?JSON.parse(payload):payload;', proxy)
        self.assertIn('!Array.isArray(event))send(event);', proxy)
        constructor = LIVE[LIVE.index(
            "const client = renderWorker?null:new LiveCommandClient({"):]
        constructor = constructor[:constructor.index("dependencyTimeoutMs")]
        self.assertIn("controlChannel:clientControlProxy", constructor)
        self.assertNotIn("controlChannel:control", constructor)

        fallback = LIVE[LIVE.index("const handleStreamFallback="):]
        fallback = fallback[:fallback.index("const clientControlProxy={")]
        self.assertIn('send({t:"streamError",message,', fallback)
        self.assertNotIn("control.send", fallback)
        self.assertEqual(LIVE.count("control.send("), 1)

    def test_host_recovery_requests_use_the_same_multipath_sender(self):
        request = LIVE[LIVE.index("const requestRecovery=detail=>{"):
                       LIVE.index("const handleClientReady=")]
        self.assertIn("    send(event);\n  };", request)
        self.assertNotIn('else if(signaling?.readyState===WebSocket.OPEN)',
                         request)

    def test_proved_gameplay_is_reported_once_with_the_page_session(self):
        latch = LIVE[LIVE.index("if(s.frame>=400 && s.draws>100){"):]
        latch = latch[:latch.index("configureReplayView();")]
        self.assertEqual(latch.count('send({t:"gameplayReady"'), 1)
        self.assertIn('send({t:"gameplayReady",frame:s.frame,'
                      'session:sessionId});', latch)
        self.assertLess(latch.index("if(!gameplayReached){"),
                        latch.index('send({t:"gameplayReady"'))


class SpectatorModeTests(unittest.TestCase):
    """?spectate=<token>&seat=N: a read-only copy of the host's stream over
    the wt-bridge. No seat claim, no input, no parking of the host."""

    def test_mode_is_keyed_on_the_token_and_pins_the_seat(self):
        self.assertIn('const spectateJoinToken=/^[0-9a-f]{32}$/.test(params.get("spectate")||"")', LIVE)
        self.assertIn("const spectatorMode=!!spectateJoinToken;", LIVE)
        seat = LIVE[LIVE.index("const resolveSeat=async()=>{"):]
        seat = seat[:seat.index("const candidates=[];")]
        self.assertIn("if(spectatorMode){", seat)
        self.assertIn("activeSeat=spectateSeat;", seat)
        self.assertIn("bridgeWsBase=`wss://${location.host}/seat${spectateSeat}`;", seat)

    def test_connect_takes_no_seat_and_opens_no_peer(self):
        connect = LIVE[LIVE.index("const connectSpectator=async()=>{"):]
        connect = connect[:connect.index("const connectWithSeat=async()=>{")]
        self.assertNotIn("/signal", connect)
        self.assertNotIn("RTCPeerConnection", connect)
        self.assertNotIn("viewerGateBusy", connect)
        self.assertIn("await dialWorkerWebTransport({spectate:spectateJoinToken});", connect)
        self.assertIn('if(textureImageVersion!==1)\n    throw new Error("spectating needs WebP texture decoding");', connect)
        self.assertIn('&&typeof ImageDecoder.isTypeSupported==="function"', LIVE)
        self.assertNotIn('typeof ImageDecoder.isTypeSupported!=="function"', LIVE)
        self.assertIn("session:sessionId,spectate,budgetMs,", LIVE)
        self.assertIn("  if(spectatorMode)return connectSpectator();", LIVE)

    def test_no_input_reaches_the_relay(self):
        self.assertIn('if(spectatorMode&&!["requestRecovery","recoveryReady","commandReady",\n'
                      '      "streamError"].includes(event.t))return false;', LIVE)
        self.assertIn('const lockEnabled=params.get("lock")!=="0"&&!spectatorMode;', LIVE)
        self.assertIn('if(replayViewConfigured||params.get("autoView")==="0"||spectatorMode)', LIVE)
        self.assertIn("if(spectatorMode){spectatorVisibility(hidden);return;}", LIVE)

    def test_game_cursor_is_suppressed_and_local_hand_is_drawn(self):
        self.assertNotIn("body.spectator #cursor { display:none!important; }", LIVE)
        self.assertIn("suppressPointerOverlay:true,", LIVE)
        self.assertNotIn("suppressPointerOverlay:!spectatorMode,", LIVE)
        self.assertIn('canvas.style.display="block";', LIVE)
        self.assertIn(
            "const position=point(event); Object.assign(cursorState,position,{inside:true});",
            LIVE)
        worker = (ROOT / "web" / "render-worker.js").read_text(encoding="utf-8")
        self.assertIn("const suppressPointer = options.suppressPointerOverlay !== false;", worker)
        self.assertIn("if (suppressPointer && !cursorFound) {", worker)

    def test_bridge_control_messages_and_host_page_protocol(self):
        handler = LIVE[LIVE.index('message.t==="streamEnded"'):]
        handler = handler[:handler.index('message.t==="engineStalled"')]
        self.assertIn('const reason=String(message.reason||"stream-ended");', handler)
        self.assertIn('if(reason==="too-slow"||reason==="stream-ended"){', handler)
        self.assertIn("recoverCommandTransport(`spectator stream ${reason}`);", handler)
        self.assertIn("finishReplay(reason);", handler)
        self.assertIn('renderWorker?.postMessage({type:"planesReset",mode:"quic"});', handler)
        self.assertIn("castOverlay.setParked(hidden);", handler)
        self.assertIn('if(spectatorMode)postToParent({type:"w3-stream-ready",watchId});', LIVE)
        self.assertIn('if(spectatorMode)postToParent({type:"w3-stream-ended",watchId,reason});', LIVE)
        self.assertIn('postToParent({type:"w3-watch-alone",', LIVE)
        self.assertIn('$("watchalone").hidden=false;', LIVE)
        self.assertIn('$("wheelbtn").hidden=true;', LIVE)

    def test_spectators_heal_at_anchors_and_replace_the_transport(self):
        # WHOS-WATCHING-NOW-SPEC.md 13.4: no host-side keyframe exists for
        # a spectator. Its transport resynchronizes at geometry anchors and
        # only a confirmed stall replaces the spectator's WT session. Old
        # bytes cannot then precede the fresh mirror snapshot.
        self.assertIn("spectator:spectatorMode,", LIVE)
        request = LIVE[LIVE.index("const requestRecovery=detail=>{"):]
        request = request[:request.index("const handleClientReady=")]
        self.assertIn('?["anchor-stall","resource-apply","dependency"]', request)
        self.assertIn("if(now-lastRecoveryRequestAt<(spectatorMode?10000:500))return;", request)
        self.assertIn('detail.reason==="anchor-stall"&&\n'
                      '        (!visibleGameFrame||spectatorResyncing)', request)
        self.assertIn('status("resyncing with the host");\n'
                      '      recoverCommandTransport(`spectator ${detail.reason}`);',
                      request)
        self.assertNotIn('postMessage({type:"rejoin"})', request)
        self.assertNotIn('send(event);\n      return;', request)
        self.assertIn("||(spectatorMode?512:96));", LIVE)
        worker = (ROOT / "web" / "render-worker.js").read_text(encoding="utf-8")
        self.assertIn("anchorRecovery: Boolean(options.spectator),", worker)
        self.assertNotIn('case "rejoin":', worker)
        self.assertIn('if (request > 0 && data.resetBeforeBind) {', worker)
        reset = worker[worker.index(
            'if (request > 0 && data.resetBeforeBind) {'):]
        reset = reset[:reset.index('} else if (request > 0) {')]
        self.assertIn('detachChannels: true,', reset)
        self.assertLess(reset.index('quiesceAndReset({'),
                        reset.index('bind();'))
        self.assertIn('dialWorkerWebTransport({\n'
                      '          spectate:spectateJoinToken});', LIVE)
        self.assertIn('resetWorkerPlanes("quic",epoch,request,12000,true);',
                      LIVE)
        self.assertIn('spectatorResyncStartFrame=Number($("frame").textContent)||0;',
                      LIVE)
        self.assertIn('Number(s.frame)>spectatorResyncStartFrame', LIVE)
        self.assertIn('recoverCommandTransport=spectatorMode\n'
                      '      ?recoverSpectatorTransport:recoverTransport;', LIVE)
        transport = (ROOT / "web" / "live-transport.js").read_text(encoding="utf-8")
        self.assertIn("const ANCHOR_FAILURE_LIMIT = 3;", transport)
        self.assertIn("Flag.DICTIONARY_ONLY", transport)
        protocol = (ROOT / "web" / "protocol.js").read_text(encoding="utf-8")
        self.assertIn("DICTIONARY_ONLY: 16,", protocol)
        native = (ROOT / "native" / "w3cs_protocol.h").read_text(encoding="utf-8")
        self.assertIn("W3CS_DICTIONARY_ONLY = 16u", native)
        bridge = (ROOT / "native" / "wt-bridge" / "spectate.go").read_text(
            encoding="utf-8")
        self.assertIn('sink.end("stream-ended")', bridge)
        self.assertNotIn('func (t *seatTap) rejoin(', bridge)

    def test_audio_rides_datagrams_through_webcodecs(self):
        self.assertIn('import { createSpectatorAudio } from "../web/spectator-audio.js";', LIVE)
        self.assertIn('case "wtAudio": spectatorAudio?.push(data.data); break;', LIVE)
        self.assertIn("spectatorAudio?.setVolume(muted?0:audioElement.volume);", LIVE)
        self.assertIn("const playAudio=()=>spectatorMode?spectatorAudio.resume():", LIVE)
        worker = (ROOT / "web" / "render-worker.js").read_text(encoding="utf-8")
        self.assertIn('post({ type: "wtAudio", data: copy.buffer },\n                    [copy.buffer]);', worker)
        planes = (ROOT / "web" / "wt-planes.js").read_text(encoding="utf-8")
        self.assertIn("if (spectate) hello.spectate = spectate;", planes)


class RelayTapTests(unittest.TestCase):
    """The relay's loopback tap is the only fan-out source; it never feeds
    the host's governors (WHOS-WATCHING-NOW-SPEC.md 5.2)."""

    def test_every_host_send_is_copied_to_the_tap_first(self):
        for name, tag in (("send_resource_packet", 2), ("send_recovery_packet", 3),
                          ("send_frame_packet", 1)):
            body = RELAY[RELAY.index(f"bool {name}(const Bytes &packet) {{"):]
            body = body[:body.index("return send_packet(")]
            self.assertIn(f"tap_send_binary({tag}, packet);", body, name)

    def test_tap_is_loopback_only_and_token_bound(self):
        self.assertIn('soup_server_add_websocket_handler(server_, "/tap", nullptr, nullptr,', RELAY)
        for handler in ("on_tap_websocket", "on_input_websocket"):
            body = RELAY[RELAY.index(f"static void {handler}("):]
            body = body[:body.index("static_cast<Relay *>(user_data)->accept_")]
            self.assertIn("if (!message_from_loopback(message)) {", body, handler)
        accept = RELAY[RELAY.index("void accept_tap_websocket("):]
        accept = accept[:accept.index("static void tap_websocket_message(")]
        self.assertIn('soup_websocket_connection_close(connection, 4005, "bad-spectate-token");', accept)
        self.assertIn("rotate_spectate_token();\n    send_spectate_token();", RELAY)

    def test_epoch_is_in_band_and_never_resets_the_codec(self):
        body = RELAY[RELAY.index("void request_epoch() {"):]
        body = body[:body.index("static gboolean epoch_timer_fired(")]
        self.assertIn("++recovery_nonce_;\n    ++snapshot_nonce_;\n    write_game_control();", body)
        self.assertNotIn("codec_.reset()", body)
        self.assertIn("kEpochMinIntervalMs = 3000;", RELAY)
        self.assertEqual(2, RELAY.count("relay_->enqueue_reliable(PacketBatch{Bytes{}});"))
        drain = RELAY[RELAY.index("if (packet.empty()) {"):]
        self.assertIn("tap_epoch_marker();", drain[:300])

    def test_bridge_taps_its_own_host_session_for_the_state_mirror(self):
        # WHOS-WATCHING-NOW-SPEC.md 13.4: the bridge dials /tap?session=<host
        # session> when the host's planes attach, so its mirror sees the
        # bootstrap; the first tap message names the spectate token. A
        # spectator the mirror cannot heal asks for a keyframe only.
        self.assertIn('query_param(message, "session"));', RELAY)
        accept = RELAY[RELAY.index("void accept_tap_websocket("):]
        accept = accept[:accept.index("void request_host_recovery() {")]
        self.assertIn("const bool by_token = !spectate_token_.empty() && token == spectate_token_;", accept)
        self.assertIn("session == viewer_session_id_;", accept)
        self.assertIn("if (!host_open || !(by_token || by_session)) {", accept)
        self.assertIn(r'tap_send_text("{\"t\":\"spectate\",\"token\":\"" + spectate_token_ +', accept)
        recovery = RELAY[RELAY.index("void request_host_recovery() {"):]
        recovery = recovery[:recovery.index("static void tap_websocket_message(")]
        self.assertIn("++recovery_nonce_;\n    write_game_control();", recovery)
        self.assertNotIn("++snapshot_nonce_;", recovery)
        self.assertIn('} else if (!std::strcmp(kind, "recovery")) {\n          request_host_recovery();', RELAY)
        bridge = (ROOT / "native" / "wt-bridge" / "spectate.go").read_text(encoding="utf-8")
        self.assertIn('"%s/tap?session=%s"', bridge)
        self.assertIn("go hostTap(ctx, seat, sessionID)",
                      (ROOT / "native" / "wt-bridge" / "main.go").read_text(encoding="utf-8"))
        mirror = (ROOT / "native" / "wt-bridge" / "mirror.go").read_text(encoding="utf-8")
        self.assertIn("flagDictionaryOnly = 0x10", mirror)
        self.assertIn('var mirrorMaxBytes = int64(envInt("W3CS_MIRROR_MAX_MB", 256)) * 1024 * 1024', mirror)

    def test_host_events_reach_spectators(self):
        self.assertIn(r'tap_send_text("{\"t\":\"reset\"}");', RELAY)
        self.assertIn(r'tap_send_text("{\"t\":\"parked\",\"hidden\":true}");', RELAY)
        self.assertIn(r'tap_send_text("{\"t\":\"sessionEnded\",\"reason\":\"replay-ended\"}");', RELAY)
        self.assertIn(r'tap_send_text("{\"t\":\"sessionEnded\",\"reason\":\"host-left\"}");', RELAY)
        self.assertIn('"tee name=atap ! queue ! rtpopuspay pt=111 "', RELAY)
        self.assertIn(r'"\"spectate\":%s,\"spectators\":%u,"', RELAY)


OVERLAY = (ROOT / "web" / "cast-overlay.js").read_text(encoding="utf-8")


class CastDescriptorTests(unittest.TestCase):
    def test_box_descriptor_is_retried_after_a_catalog_fallback(self):
        # The seat writes /cast/<sha1>.json during the first play; the page
        # must pick it up even though its first request came too early.
        self.assertIn("const boxUrl = urls[0];", OVERLAY)
        self.assertIn("if (source === boxUrl) return;", OVERLAY)
        self.assertIn("for (const delay of [4000, 10000, 25000, 60000]) {", OVERLAY)
        self.assertIn("adopt(await fetchDescriptor(boxUrl), boxUrl);", OVERLAY)
        self.assertIn("root.dataset.castOverlaySource = url;", OVERLAY)

    def test_launch_scripts_generate_the_descriptor_once(self):
        for name in ("lab-game-session.sh", "lab-switch-replay.sh"):
            script = (ROOT / "native" / name).read_text(encoding="utf-8")
            self.assertIn('ensure-cast-descriptor.sh" \\\n  "$REPLAY_SOURCE" "$REPLAY_ID" || true', script, name)
        ensure = (ROOT / "native" / "ensure-cast-descriptor.sh").read_text(encoding="utf-8")
        self.assertIn('[[ -s "$OUT" ]] && exit 0', ensure)
        self.assertIn('--replay "$REPLAY" --id "$ID" --out "$OUT"', ensure)


PROXY = (ROOT / "native" / "d3d9_proxy.c").read_text(encoding="utf-8")
PROTOCOL = (ROOT / "native" / "w3cs_protocol.h").read_text(encoding="utf-8")
COMMAND_STATE = (ROOT / "web" / "command-state.js").read_text(encoding="utf-8")
RENDERER = (ROOT / "web" / "webgpu-renderer.js").read_text(encoding="utf-8")


class AdaptiveGpuResolutionTests(unittest.TestCase):
    def test_live_page_enables_adaptation_but_parity_mode_does_not(self):
        self.assertIn('maxOutputWidth:4096', LIVE)
        self.assertIn('params.get("adaptiveResolution")!=="0"', LIVE)
        self.assertIn('adaptiveResolution: options.adaptiveResolution',
                      RENDER_WORKER)
        self.assertIn('this.adaptiveResolution = Boolean(adaptiveResolution)\n'
                      '      && !this.parityMode;', RENDERER)

    def test_gpu_probe_is_nonblocking_and_scale_changes_at_a_frame_boundary(self):
        self.assertIn('this.device.queue.onSubmittedWorkDone().then(() => {',
                      RENDERER)
        self.assertNotIn('await this.device.queue.onSubmittedWorkDone()',
                         RENDERER)
        begin = RENDERER[RENDERER.index('beginFrame(meta) {'):]
        begin = begin[:begin.index('suppressTextureDraws(textureId)')]
        self.assertLess(begin.index('this._applyPendingRenderScale();'),
                        begin.index('this.frameMeta = meta;'))

    def test_resolution_state_is_visible_to_live_monitoring(self):
        self.assertIn('document.documentElement.dataset.gpuResolution=', LIVE)
        for field in ('renderScale:', 'gpuQueueMs:', 'gpuProbePending:',
                      'maximumRenderScale:'):
            self.assertIn(field, RENDERER)

    def test_console_logs_the_initial_scale_and_each_transition_once(self):
        self.assertIn('let loggedRenderScale=null;', LIVE)
        self.assertIn('scale===loggedRenderScale)return;', LIVE)
        self.assertIn('console.log(`Web render ${scale.toFixed(2)} · '
                      '${reason}${queue}${output}`);', LIVE)
        self.assertIn('logRenderScale(s);', LIVE)


class EngineClockTests(unittest.TestCase):
    """The overlay's clock is the engine's replay position carried in
    every frame, not wall time scaled by a guessed speed: right at 8X,
    paused, after a restart, and for a spectator joining mid-game."""

    def test_recorder_stamps_every_frame_with_the_engine_clock(self):
        self.assertIn("W3CS_OP_REPLAY_CLOCK = 38", PROTOCOL)
        self.assertIn("struct w3cs_replay_clock {", PROTOCOL)
        self.assertIn("    raw_record_locked(W3CS_OP_FRAME_STATE, 0, &state_version,\n"
                      "                      sizeof(state_version));\n"
                      "    record_replay_clock_locked();", PROXY)
        # Read-only: the recorder never writes engine memory, and a freed
        # address reports an error instead of faulting the game.
        self.assertIn("ReadProcessMemory(GetCurrentProcess(), (LPCVOID)address, value,", PROXY)
        self.assertNotIn("WriteProcessMemory", PROXY)
        self.assertIn('"W3CS_REPLAY_CLOCK", create,', PROXY)

    def test_discovery_selects_a_counter_that_ticks_like_a_clock(self):
        # Measured on a live 1.12 seat: "advanced within the speed ladder"
        # alone picks buffer sizes that jump 4096 at a time and counters
        # that stop after loading. The replay position ticks every
        # simulation step, so it moves on nearly every 50 ms sample.
        self.assertIn("#define CLOCK_FINE_GAP_MS 50u", PROXY)
        self.assertIn("#define CLOCK_MIN_TICKING 70u", PROXY)
        self.assertIn("if (100u * ticks / (CLOCK_FINE_SAMPLES - 1) < CLOCK_MIN_TICKING)", PROXY)
        self.assertIn("if (mean <= 0 || worst / mean > CLOCK_MAX_UNEVENNESS)", PROXY)
        # and it must be where a clock started with this stream would be
        self.assertIn("uint32_t minimum = (uint32_t)(CLOCK_MIN_RATE * floor_age);", PROXY)
        self.assertIn("uint32_t maximum = (uint32_t)(CLOCK_MAX_RATE * (age + CLOCK_LOAD_SLACK_MS));", PROXY)
        # ties go to the rate closest to a speed the panel can be in
        self.assertIn("static const double kClockLadder[]", PROXY)
        self.assertIn("distance = clock_ladder_distance(rate);", PROXY)

    def test_discovery_prefers_a_counter_that_matches_the_stream_age(self):
        # Several counters tick at exactly 1X. The replay position is the
        # one that has been running as long as the stream has (measured:
        # 22.6 s into a 28 s old stream, against another 1X counter at
        # 5.5 s), and it can never be AHEAD of the stream.
        self.assertIn("static double clock_lag(double value, double rate, double age_ms)", PROXY)
        self.assertIn("if (value > expected + slack)\n        return 1e9;", PROXY)
        self.assertIn("score = distance * 2.0 + lag;", PROXY)
        # the group cap must not evict it: the engine keeps ONE copy of
        # the live position and many copies of other counters
        self.assertIn("} else if (score < lag[capacity - 1]) {", PROXY)

    def test_the_clock_is_zero_when_gameplay_starts_not_when_the_seat_does(self):
        # A counter started at map load reads seconds ahead of the game's
        # own Elapsed Time (measured: 1:09.5 against 1:02). The 3D scene's
        # draw count marks where the replay position is zero.
        self.assertIn("if (g_draw_ordinal >= 180u) {", PROXY)
        self.assertIn("InterlockedExchange(&g_gameplay_tick, (LONG)GetTickCount());", PROXY)
        self.assertIn("DWORD age = gameplay ? GetTickCount() - (DWORD)gameplay : 0;", PROXY)
        # and it follows the LAST gameplay start: a warm seat streams its
        # previous replay's gameplay until the switch lands, so the
        # map-load screen (frames below the floor) drops the anchor, and a
        # published counter that did not rewind with the restart goes.
        self.assertIn("++g_gameplay_gap >= CLOCK_GAMEPLAY_GAP_FRAMES", PROXY)
        self.assertIn("InterlockedIncrement(&g_gameplay_epoch);", PROXY)
        self.assertIn('clock_unpublish("did-not-rewind");', PROXY)
        # The session script runs under set -u: a variable it never sets
        # aborted every cold launch before the clock block was reached.
        session = (ROOT / "native" / "lab-game-session.sh").read_text(encoding="utf-8")
        self.assertNotIn("$LAB_ROOT", session)
        self.assertIn("${W3CS_LAB_ROOT:-/home/ubuntu/w3cs-lab}/clock", session)

    def test_a_clock_that_only_follows_real_time_is_replaced(self):
        # At 1X, real-time-since-gameplay and the replay position are the
        # same number, so discovery cannot separate them. They separate at
        # any other speed (measured at 2X: position 2.0x, impostor 1.0x),
        # so the runners-up stay under observation.
        self.assertIn("#define CLOCK_TRACKED 4u", PROXY)
        self.assertIn("static void clock_watch(void)", PROXY)
        self.assertIn("item->rate > 1.0 + CLOCK_REALTIME_MARGIN", PROXY)
        self.assertIn("if (++item->confirmations >= 2", PROXY)
        self.assertIn("clock_lag((double)value, item->rate, age) <= 0.25", PROXY)
        self.assertIn("tracked real time; ", PROXY)
        # and only a clock that IS keeping pace with the wall is replaced
        self.assertIn("if (item->rate > 1.0 - CLOCK_REALTIME_MARGIN\n                && item->rate < 1.0 + CLOCK_REALTIME_MARGIN) {", PROXY)

    def test_a_known_address_is_verified_instead_of_rescanning(self):
        # Scanning process memory is the only cost the GAME pays for the
        # clock, so it happens once per engine build: the hit is
        # remembered and later sessions verify it with one read. Claim
        # images restore the same layout every time, so this always hits.
        self.assertIn("static void clock_remember(uintptr_t address)", PROXY)
        # the remembered set carries the runners-up, or a session that
        # starts from it cannot tell the position from real time later
        self.assertIn("if (remembered && g_clock_tracked_count >= 2", PROXY)
        self.assertIn("&& rate < 1.0 + CLOCK_REALTIME_MARGIN && lag < 0.15)\n            score += 0.5;", PROXY)
        self.assertIn("static uintptr_t clock_recall(void)", PROXY)
        self.assertIn("clock_log(\"w3cs: replay clock reused a known address", PROXY)
        self.assertIn("g_clock_attempts < CLOCK_MAX_ATTEMPTS", PROXY)
        session = (ROOT / "native" / "lab-game-session.sh").read_text(encoding="utf-8")
        self.assertIn('export W3CS_REPLAY_CLOCK_DIR=', session)
        images = (ROOT / "deploy" / "w3cs-criu-build-images.sh").read_text(encoding="utf-8")
        self.assertIn("W3CS_REPLAY_CLOCK_DIR=", images)
        # every classic claim image bakes ONE env file, so the engine has
        # to come from the game's own path, not from the environment
        self.assertIn("static void clock_choose_file(void)", PROXY)
        self.assertIn("GetModuleFileNameA(NULL, module,", PROXY)

    def test_failed_criu_restore_falls_back_to_a_clean_cold_launch(self):
        session = (ROOT / "native" / "lab-game-session.sh").read_text(
            encoding="utf-8"
        )
        restore = session.index('"$CRIU_WORKER" restore "$SEAT"')
        fallback = session.index("|| CLAIM_FALLBACK=1", restore)
        cleanup = session.index("  criu_cleanup\n  trap - EXIT INT TERM", fallback)
        cold_launch = session.index('export WINEPREFIX="$PREFIX"', cleanup)
        self.assertLess(restore, fallback)
        self.assertLess(fallback, cleanup)
        self.assertLess(cleanup, cold_launch)

    def test_frame_decodes_the_clock_into_the_renderer_stats(self):
        self.assertIn("REPLAY_CLOCK: 38,", COMMAND_STATE)
        self.assertIn("if (output) output.replayClockMs = this.replayClockMs;", COMMAND_STATE)
        self.assertIn("replayClockMs: Number.isFinite(frame.replayClockMs)", RENDERER)

    def test_page_hands_each_frame_clock_to_the_overlay(self):
        self.assertIn("if(Number.isFinite(s.replayClockMs))castOverlay.setEngineClock(s.replayClockMs);", LIVE)
        self.assertNotIn("live-watch-position", LIVE)
        self.assertNotIn("syncClock", LIVE)

    def test_overlay_prefers_the_engine_clock_over_its_estimate(self):
        self.assertIn('import { ReplayClock } from "./replay-clock.js";', OVERLAY)
        self.assertIn("setEngineClock(ms) {", OVERLAY)
        self.assertIn("if (state.engine.seen) {\n      if (!state.ended) state.clockMs = state.engine.estimate(now);", OVERLAY)
        self.assertIn('document.documentElement.dataset.castClockSource = "engine";', OVERLAY)
        self.assertIn("if (!state.engine.seen) state.clockMs = 0;", OVERLAY)


class HostPageProtocolTests(unittest.TestCase):
    def test_embed_mode_talks_to_the_catalog_origin_only(self):
        self.assertIn('const embedMode=params.get("embed")==="1";', LIVE)
        self.assertIn("try{return new URL(document.referrer).origin;}catch{return \"\";}", LIVE)
        self.assertIn("if(event.origin!==parentOrigin||!event.data)return;", LIVE)
        self.assertIn('postToParent({type:"w3-standby-ready"});', LIVE)
        self.assertIn('postToParent({type:"w3-close"});return;', LIVE)
        self.assertIn('postToParent({type:"w3-replay-ended",game:`arc${replayId}`});', LIVE)
        self.assertIn('if(event.data.type==="w3-visibility"){', LIVE)

    def test_enter_opens_the_parent_chat_only_while_armed(self):
        # LIVE-CHAT-SPEC.md 8.1: the parent arms chat; an armed Enter is
        # consumed (down and up), pointer lock is released first so the
        # mouse can reach the input, and nothing else changes.
        self.assertIn("let chatArmed=false,chatEnterHeld=false;", LIVE)
        self.assertIn('if(event.data.type==="w3-chat-state"){chatArmed=!!event.data.armed;}', LIVE)
        self.assertIn('if(event.key==="Enter"&&(chatArmed||chatEnterHeld)&&!event.ctrlKey\n'
                      "    &&!event.altKey&&!event.metaKey){", LIVE)
        self.assertIn("if(event.repeat)return;\n      chatEnterHeld=true;", LIVE)
        self.assertNotIn("if(event.repeat||chatEnterHeld)return;", LIVE)
        self.assertIn('window.addEventListener("blur",()=>{releasePan();releaseHeldButton();'
                      "chatEnterHeld=false;});", LIVE)
        self.assertIn('$("game").dataset.blockedInput="chat-enter";', LIVE)
        self.assertIn("if(document.pointerLockElement)document.exitPointerLock();\n"
                      '      postToParent({type:"w3-chat-open"});', LIVE)
        self.assertIn("}else chatEnterHeld=false;\n    return;", LIVE)
        branch = LIVE.index('if(event.key==="Enter"&&(chatArmed||chatEnterHeld)')
        self.assertLess(LIVE.index("event.preventDefault();\n  if(event.key"), branch)
        self.assertLess(branch, LIVE.index("if(isProtectedReplayKey(event.key)){"))

    def test_parking_follows_the_host_page_inside_an_iframe(self):
        self.assertIn("||(embedMode?parentAttentionAway", LIVE)
        self.assertIn("if(embedMode)return;   // the host page reports attention", LIVE)
        self.assertIn("parentAttentionAway=!!event.data.hidden&&!attentionHere();", LIVE)
        self.assertIn("sendVisibility(viewerAway());", LIVE)

    def test_local_attention_beats_a_stale_host_page_report(self):
        # A visible, focused player is proof of attention: a host-page
        # "away" that arrives then is stale (its blur fired because focus
        # moved INTO the frame). Focus and the veil click clear the flag, so
        # a signaling reconnect cannot re-assert it and re-park a viewer who
        # is looking at the game.
        self.assertIn('const attentionHere=()=>document.visibilityState==="visible"\n'
                      "  &&document.hasFocus();", LIVE)
        self.assertIn("const attentionProvenHere=()=>{\n"
                      "  parentAttentionAway=false;\n"
                      "  sendVisibility(false);\n"
                      "};", LIVE)
        self.assertIn('$("pauseveil").onclick=()=>{window.focus();attentionProvenHere();};', LIVE)
        self.assertIn('window.addEventListener("focus",attentionProvenHere);', LIVE)
        self.assertNotIn("parentAttentionAway=!!event.data.hidden;", LIVE)


class AudioRecoveryTests(unittest.TestCase):
    """Audio rides its own WebRTC peer. A failed or silent peer is rebuilt
    by the page and re-offered by the relay; both sides log it."""

    def test_page_rebuilds_the_audio_peer_and_asks_for_a_fresh_offer(self):
        self.assertIn("const buildAudioPeer=()=>{", LIVE)
        self.assertIn("queuedIce.audio.length=0;", LIVE)
        body = LIVE[LIVE.index("restartAudioPeer=(reason,{notify=true}={})=>{"):]
        body = body[:body.index("return true;\n  };")]
        self.assertIn("if(signaling?.readyState!==WebSocket.OPEN||replayEnded)return false;", body)
        self.assertIn("||audioRestarts>=AUDIO_RESTART_LIMIT))return false;", body)
        self.assertIn("try{audioPeer.close();}catch{}", body)
        self.assertIn("buildAudioPeer();", body)
        self.assertIn('if(notify)signaling.send(JSON.stringify({type:"audioRestart",reason}));', body)

    def test_watchdog_triggers_on_failure_disconnect_and_silence_only_in_play(self):
        body = LIVE[LIVE.index("const updateAudioStats=async()=>{"):]
        body = body[:body.index("updateTransport();setTimeout(updateTransport,1500);")]
        self.assertIn('if(state==="failed")restartAudioPeer("audio peer failed");', body)
        self.assertIn("&&now-audioDisconnectedSince>=AUDIO_DISCONNECT_GRACE_MS)", body)
        self.assertIn("&&now-audioStalledSince>=AUDIO_SILENCE_GRACE_MS)", body)
        # A parked or hidden game sends nothing on purpose.
        self.assertIn("const gameRunning=gameplayReached&&!replayEnded\n"
                      '          &&document.visibilityState==="visible"\n'
                      '          &&!document.body.classList.contains("parkveil");', body)

    def test_claim_that_arrives_late_rebinds_the_audio_source(self):
        # A CRIU claim publishing its environment after the audio peer was
        # built left the relay reading the shared seat sink: a live track
        # of digital silence (seen live on both London boxes 2026-08-23).
        body = RELAY[RELAY.index("native claim environment ready display="):]
        body = body[:body.index("void park_engine() {")]
        self.assertIn("} else if (audio_pipeline_ && !audio_from_claim_ &&", body)
        self.assertIn('send_interactive_control("{\\"t\\":\\"audioRebind\\"}");', body)
        self.assertIn("destroy_audio_peer();\n        schedule_audio_peer();", body)
        self.assertIn('std::printf("native audio source %s claim=%d\\n"', RELAY)
        # A rebuilt pipeline re-derives its binding; a stale flag would
        # hide a second wrong binding.
        self.assertIn("void destroy_audio_peer() {\n    audio_offer_started_ = false;\n"
                      "    audio_from_claim_ = false;", RELAY)
        # The page rebuilds its peer for the new offer without asking again.
        self.assertIn('else if(message.t==="audioRebind"){', LIVE)
        self.assertIn('restartAudioPeer("host rebound audio to the claim daemon",\n'
                      "        {notify:false});", LIVE)
        self.assertIn('if(notify)signaling.send(JSON.stringify({type:"audioRestart",reason}));', LIVE)

    def test_network_report_carries_audio_health(self):
        self.assertIn("const audio=audioPeer?{state:audioPeer.connectionState,", LIVE)
        self.assertIn("awaitingRecovery:view.awaitingRecovery,clientMs,audio}));", LIVE)

    def test_network_report_identifies_live_command_transport(self):
        self.assertIn('const commandTransport=wtActive?"webtransport":', LIVE)
        self.assertIn('dataset.frameTransport==="websocket"', LIVE)
        self.assertIn('document.documentElement.dataset.peerStats=JSON.stringify({\n'
                      '      commandTransport,', LIVE)
        self.assertIn('JSON.stringify({type:"network",commandTransport,', LIVE)
        self.assertIn('const measuredRtt=Number.isFinite(', LIVE)
        self.assertIn('rtt:measuredRtt,receiveBps,renderFps,p95GapMs,', LIVE)
        self.assertIn('if(signaling?.readyState===WebSocket.OPEN){', LIVE)

    def test_relay_rebuilds_the_audio_peer_on_request_and_logs_audio_health(self):
        body = RELAY[RELAY.index('} else if (!std::strcmp(type, "audioRestart")) {'):]
        body = body[:body.index('} else if (!std::strcmp(type, "resourceCache")) {')]
        self.assertIn("destroy_audio_peer();", body)
        self.assertIn("schedule_audio_peer();", body)
        self.assertIn("std::chrono::seconds(2)", body)
        self.assertIn('"native audio peer state ice=%d connection=%d\\n"', RELAY)
        self.assertIn('"native client audio state=%.24s ice=%.24s "', RELAY)
        self.assertIn('"audioPkt=%.1f/s clientAudio=%.24s/%.24s %.0fkbps\\n"', RELAY)
        # close_peer uses the same teardown as a restart.
        body = RELAY[RELAY.index("void close_peer(bool terminate_engine = false) {"):]
        body = body[:body.index("if (frame_pipeline_)")]
        self.assertIn("destroy_audio_peer();", body)


class SpectatorStatsTests(unittest.TestCase):
    """A spectator has no WebRTC peer; its numbers come from the client
    view and the worker's WebTransport byte counter."""

    def test_stats_interval_runs_for_spectators(self):
        self.assertIn("if(!peer&&!spectatorMode&&!wtActive", LIVE)
        self.assertIn('dataset.frameTransport!=="websocket")return;', LIVE)
        self.assertNotIn("  setInterval(async()=>{\n    if(!peer)return;", LIVE)

    def test_spectator_has_no_datachannel_buffer_numbers(self):
        self.assertIn("resourceBuffered:reliable?.bufferedAmount,", LIVE)
        self.assertIn("frameBuffered:frames?.bufferedAmount,", LIVE)


class DrawnFrameLogTests(unittest.TestCase):
    """poc/e2e-spectate-canary.mjs compares drawn frame ids across players
    through a DOM attribute the page keeps fresh."""

    def test_page_publishes_recent_drawn_frame_ids(self):
        self.assertIn("const DRAWN_LOG_LIMIT=600;", LIVE)
        self.assertIn("noteDrawnFrame(s.frame);", LIVE)
        self.assertIn("document.documentElement.dataset.drawnFrames=JSON.stringify(drawnLog);", LIVE)
        self.assertIn("document.documentElement.dataset.drawnTick=String(++drawnLogTick);", LIVE)

    def test_faults_before_gameplay_hand_off_and_later_faults_do_not(self):
        self.assertIn("if(feederHandedOff||gameplayReached)return false;", LIVE)
        self.assertIn('postToParent({type:"w3-feeder-failed",reason});', LIVE)
        self.assertIn('if(handOffToFallback((spectatorMode?"spectate-failed:":"session-failed:")', LIVE)
        self.assertIn('}else if(error===NO_FREE_SEATS&&handOffToFallback("no-free-seat")){', LIVE)
        self.assertIn('}else if(error===VIEWER_BUSY&&parentOrigin&&handOffToFallback("viewer-busy")){', LIVE)
        # the handoff helper is defined before the top-level connect await
        self.assertLess(LIVE.index("const handOffToFallback=reason=>{"),
                        LIVE.index("({reliable,frames,frameAlternate=null,recovery,control,"))

    def test_hand_off_needs_a_host_page(self):
        # No ?fallback= link: a deep link lands on the catalog's replay page
        # (/watch -> /replay/<sha1>?watch=1), which hosts this page in its
        # iframe and re-routes on w3-feeder-failed.
        self.assertNotIn("fallbackUrl", LIVE)
        self.assertNotIn("CATALOG_ORIGINS", LIVE)
        body = LIVE[LIVE.index("const handOffToFallback=reason=>{"):]
        body = body[:body.index('postToParent({type:"w3-standby-ready"});')]
        self.assertIn("if(parentOrigin){", body)
        self.assertIn("  return false;\n};", body)

    def test_no_worker_hop_inside_an_iframe(self):
        self.assertIn("if(parentOrigin)return false;   // the host page routes again instead", LIVE)

    def test_back_button_outside_an_iframe_returns_to_the_catalog(self):
        self.assertIn('else location.href="https://war3replays.com/";', LIVE)
        self.assertIn("if(history.length>1)history.back();", LIVE)


if __name__ == "__main__":
    unittest.main()


class CastOverlaySourcesTests(unittest.TestCase):
    OVERLAY = (ROOT / "web" / "cast-overlay.js").read_text(encoding="utf-8")

    def test_page_offers_the_staged_descriptor_then_the_catalog_roster(self):
        self.assertIn("const castUrls=[new URL(`../cast/${encodeURIComponent(replayId)}.json`,", LIVE)
        self.assertIn("castUrls.push(`https://war3replays.com/api/cast/${replayId}`);", LIVE)
        self.assertIn("castUrl:castUrls,clickGame:(x,y)=>clickGameControl(x,y)", LIVE)

    def test_overlay_tries_each_source_in_order(self):
        self.assertIn("const urls = (Array.isArray(castUrl) ? castUrl : [castUrl]).filter(Boolean);", self.OVERLAY)
        self.assertIn("for (const url of urls) {", self.OVERLAY)
        self.assertIn("root.dataset.castOverlaySource = url;", self.OVERLAY)
        self.assertIn('throw new Error("cast descriptor has no roster");', self.OVERLAY)


WT_PLANES = (ROOT / "web" / "wt-planes.js").read_text(encoding="utf-8")
WT_DIAL = (ROOT / "web" / "wt-dial.js").read_text(encoding="utf-8")
WORKER = (ROOT / "web" / "render-worker.js").read_text(encoding="utf-8")
BRIDGE_MAIN = (ROOT / "native" / "wt-bridge" / "main.go").read_text(encoding="utf-8")
DEPLOY = (ROOT / "deploy" / "deploy-cpu-server.sh").read_text(encoding="utf-8")
PROVISION = (ROOT / "deploy" / "provision-cpu-server.sh").read_text(encoding="utf-8")


class WebTransportDialTests(unittest.TestCase):
    """The connection-rate work of 2026-08-23 (RELEASE-PLAN.md "WebTransport
    connection rate"): UDP 443 first, a fresh object per retry under one
    budget, attempt ids with cancellation, and per-try telemetry."""

    def test_udp_443_is_the_primary_endpoint_and_4443_the_secondary(self):
        self.assertIn('const WT_PORTS=Object.freeze(["443","4443"]);', LIVE)
        self.assertIn('const wtPorts=params.get("wtPort")?[params.get("wtPort")]:WT_PORTS;', LIVE)
        # The 443 URL is canonical (no port), the legacy one explicit.
        self.assertIn('`https://${location.hostname}${port==="443"?"":`:${port}`}`', LIVE)
        self.assertIn('listenAddrs = flag.String("listen", ":443,:4443",', BRIDGE_MAIN)
        self.assertIn("conns, err := listenAll(addrs)", BRIDGE_MAIN)
        self.assertIn("connected from %s via %s (origin %q)", BRIDGE_MAIN)
        self.assertIn("ufw allow 443/udp >/dev/null\n  ufw allow 4443/udp >/dev/null", DEPLOY)
        self.assertIn("443/udp 4443/udp; do", PROVISION)

    def test_every_wt_connect_goes_through_the_dial_ladder(self):
        # One helper for the host connect, the spectator join and the
        # mid-session re-dial; the worker runs the ladder itself.
        self.assertEqual(1, LIVE.count('renderWorker.postMessage({type:"wtConnect"'))
        self.assertGreaterEqual(LIVE.count("await dialWorkerWebTransport("), 3)
        self.assertNotIn("wtStateWaiter", LIVE)
        self.assertIn('import { dialWebTransport } from "./wt-dial.js";', WORKER)
        self.assertIn("const result = await dialWebTransport({", WORKER)
        self.assertIn("budgetMs: 7500,", WT_DIAL)
        self.assertIn("firstAttemptMs: 3500,", WT_DIAL)
        self.assertIn("retryDelayMs: 250,", WT_DIAL)
        # A fresh object per try: the previous try's signal is fired before
        # the next open, and `open` receives the try's own signal.
        self.assertIn("controller.abort();\n    }\n    if (!last && remaining() > retryDelayMs + minAttemptMs)\n      await sleep(retryDelayMs);", WT_DIAL)
        self.assertIn("const planes = await open(endpoint, { signal: controller.signal,\n        timeoutMs });", WT_DIAL)

    def test_one_budget_shared_by_every_startup_stage(self):
        # No per-stage allowance: one deadline timer, cleared on settle,
        # and the page's own wait is longer than the worker's budget.
        self.assertIn("const deadline = setTimeout(() => failStartup(new WebTransportPlanesError(\n    `${phase} timed out`, { phase })), timeoutMs);", WT_PLANES)
        self.assertIn("clearTimeout(deadline);", WT_PLANES)
        self.assertNotIn("timed out`)), timeoutMs))]);", WT_PLANES)
        self.assertIn("const WT_PAGE_MARGIN_MS=1500;", LIVE)
        self.assertIn("},budgetMs+WT_PAGE_MARGIN_MS);", LIVE)
        self.assertIn("const WT_DIAL=Object.freeze({budgetMs:7500,firstAttemptMs:3500,", LIVE)

    def test_attempt_ids_cancellation_and_stale_replies(self):
        self.assertIn('const attempt=`${sessionId.slice(0,8)}-${++wtAttemptSerial}`;', LIVE)
        self.assertIn("if(data.attempt!==attempt)return;", LIVE)
        self.assertIn('renderWorker.postMessage({type:"wtCancel",attempt});', LIVE)
        self.assertIn('case "wtCancel":', WORKER)
        self.assertIn("if (wtDial && wtDial.attempt === data.attempt) wtDial.controller.abort();", WORKER)
        self.assertIn('post({ type: "wtState", state: "ready", attempt,', WORKER)
        self.assertIn('post({ type: "wtState", state: "failed", attempt,', WORKER)
        # A superseded or cancelled dial never becomes the live session.
        self.assertIn("if (wtDial !== dial || controller.signal.aborted) {", WORKER)
        self.assertIn("wtDial?.controller.abort();", WORKER)
        self.assertIn("if (signal?.aborted) throw cancelled();", WT_PLANES)
        # The page's dispatcher only acts on a closed session; dial answers
        # belong to the helper.
        self.assertIn('if(data.state!=="closed")break;', LIVE)

    def test_failures_are_reported_through_signaling(self):
        self.assertIn('const line={type:"wtReport",...report};', LIVE)
        self.assertIn('if(data.type==="wtAttempt"){', LIVE)
        self.assertIn('wtReport({attempt:"",phase:"eligibility",outcome:"skipped",', LIVE)
        for reason in ("wt=0", "no render worker", "no WebTransport in this browser",
                       "insecure context", "no seat number"):
            self.assertIn(f'"{reason}"', LIVE)
        self.assertIn('outcome:wtActive?"active":"fallback",', LIVE)
        self.assertIn('wtReport({attempt:state.attempt,phase:"redial",', LIVE)
        self.assertIn('} else if (!std::strcmp(type, "wtReport")) {', RELAY)
        self.assertIn('"native wt report attempt=%s try=%lld phase=%s outcome=%s "', RELAY)
        # Every startup failure names its stage for that log line.
        for phase in ('phase = "control";', 'phase = "handshake";',
                      '{ phase: "cancelled" }'):
            self.assertIn(phase, WT_PLANES)
        self.assertIn("const name = error?.name && error.name !== \"Error\" ? `${error.name}: ` : \"\";", WT_PLANES)


class RecoveryTransportRebindTests(unittest.TestCase):
    """A lost recovery keyframe must heal in place without reusing the path
    that stalled. The browser canary covers the runtime; these assertions pin
    the transport ordering and the exact-frame watchdog contract."""

    def setUp(self):
        self.recover = LIVE[LIVE.index("const recoverTransport=async reason=>{"):]
        self.recover = self.recover[:self.recover.index(
            "recoverCommandTransport=spectatorMode")]

    def test_recovery_tries_fresh_quic_then_fresh_websockets(self):
        quic = self.recover.index("await dialWorkerWebTransport();")
        websockets = self.recover.index("await Promise.allSettled([")
        datachannels = self.recover.index(
            'if(!mode&&wasWt&&["reliable","frame","recovery"]')
        self.assertLess(quic, websockets)
        self.assertLess(websockets, datachannels)
        for path in ("resource", "frame", "recovery"):
            self.assertIn(f'openPlaneSocket("{path}"', self.recover)
        self.assertIn('mode="websocket";', self.recover)

    def test_datachannel_stall_never_reuses_the_same_association(self):
        # `wasWt` is captured before the QUIC retry. DataChannel fallback is
        # therefore possible only when QUIC, not a DataChannel, was active.
        self.assertEqual(self.recover.count('mode="datachannel";'), 1)
        self.assertIn(
            'if(!mode&&wasWt&&["reliable","frame","recovery"].every(name=>',
            self.recover)
        self.assertNotIn(
            'if(!mode&&["reliable","frame","recovery"].every(name=>',
            self.recover)

    def test_intentional_quic_close_does_not_start_another_handoff(self):
        # A WebSocket handoff closes the old worker-owned QUIC session on
        # purpose. Its asynchronous `closed` event must be ignored after the
        # page has already marked QUIC inactive, or it can start a second
        # recovery and replace the fresh sockets.
        handler = LIVE[LIVE.index('case "wtState":'):]
        handler = handler[:handler.index('default: break;')]
        self.assertIn('}else if(!ignoreWtClose&&(wtActive||transportRecovering)',
                      handler)
        self.assertIn('if(transportRecovering){transportRecoveryPending=',
                      self.recover)
        self.assertIn('ignoreWtClose=true;wtActive=false;', self.recover)

    def test_failed_redial_keeps_the_old_quic_session_closeable(self):
        connect = WORKER[WORKER.index('case "wtConnect":'):]
        connect = connect[:connect.index('case "wtCancel":')]
        self.assertIn("const previousPlanes = wtPlanes;", connect)
        self.assertNotIn("wtPlanes = null;\n          post({ type: \"wtState\", "
                         "state: \"failed\"", connect)
        self.assertIn("previousPlanes.close();", connect)
        close = WORKER[WORKER.index('case "wtClose":'):]
        close = close[:close.index('case "chan":')]
        self.assertLess(close.index("wtPlanes = null;"),
                        close.index("closing?.close();"))
        self.assertIn('post({ type: "wtClosed",', close)
        self.assertIn("await closeWorkerWebTransport();", self.recover)

    def test_recovery_plane_opens_are_bounded(self):
        helper = LIVE[LIVE.index("const opened = "):]
        helper = helper[:helper.index("const makeSessionStarter=")]
        self.assertIn("if(timeoutMs>0)timer=setTimeout", helper)
        self.assertIn("channel.removeEventListener(\"close\",onClose);", helper)
        self.assertIn("const PLANE_OPEN_TIMEOUT_MS=5000;", helper)
        self.assertIn("try{socket.close();}catch{}", helper)

    def test_websocket_reset_binds_clean_forwarded_worker_planes(self):
        reset = WORKER[WORKER.index('const resetPlanes = async data => {'):]
        reset = reset[:reset.index('self.onmessage =')]
        self.assertIn('} else if (mode === "websocket") {', reset)
        for name in ("reliable", "frame", "recovery"):
            self.assertIn(f'{name}: forwardedChannel("{name}")', reset)
        for argument in ("reliableChannel: next.reliable",
                         "frameChannel: next.frame",
                         "recoveryChannel: next.recovery"):
            self.assertIn(argument, reset)
        self.assertIn("await client.transport.quiesceAndReset()", reset)
        self.assertIn("channelEpoch = epoch;", reset)
        self.assertIn('post({ type: "planesResetResult", request, mode,', reset)
        self.assertIn('throw new Error("fresh QUIC planes closed before bind")',
                      reset)
        self.assertIn('throw new Error("fresh DataChannel planes are not all open")',
                      reset)
        dispatch = WORKER[WORKER.index('case "chan":'):]
        dispatch = dispatch[:dispatch.index('case "chanReal":')]
        self.assertIn("Number(data.epoch || 0) !== channelEpoch", dispatch)
        # The relay prepares and holds the new epoch first. The worker then
        # creates clean shims before the page forwards sockets, and the exact
        # commit releases bytes only after both actions complete.
        self.assertLess(self.recover.index(
            'signaling.send(JSON.stringify({type:"streamReset",request}));'),
            self.recover.index("await dialWorkerWebTransport();"))
        self.assertLess(self.recover.index(
            'signaling.send(JSON.stringify({type:"streamReset",request}));'),
            self.recover.index("await closeWorkerWebTransport();"))
        self.assertLess(self.recover.index(
            'await resetWorkerPlanes(mode,epoch,request);'),
            self.recover.index('forwardPlaneToWorker("reliable",reliable);'))
        self.assertLess(self.recover.index(
            'forwardPlaneToWorker("recovery",recovery);'),
            self.recover.index(
                'signaling.send(JSON.stringify({type:"streamResetCommit",request,'))
        forward = LIVE[LIVE.index("const forwardPlaneToWorker="):]
        forward = forward[:forward.index("if(renderWorker){")]
        self.assertIn("const epoch=forwardedPlaneEpoch;", forward)
        self.assertIn('{type:"chan",name,epoch,data}', forward)
        self.assertIn("const epoch=++forwardedPlaneEpoch;", self.recover)
        self.assertIn('{type:"planesReset",mode,epoch,request,\n'
                      '          resetBeforeBind}', LIVE)
        self.assertIn('message.t==="streamResetPrepared"', LIVE)
        self.assertIn('message.t==="streamResetCommitted"', LIVE)

    def test_relay_holds_and_commits_one_explicit_transport(self):
        prepare = RELAY[RELAY.index(
            '} else if (!std::strcmp(type, "streamReset")) {'):]
        prepare = prepare[:prepare.index(
            '} else if (!std::strcmp(type, "visibility")) {')]
        self.assertIn('detach_stale_planes("viewer stream reset prepared")',
                      prepare)
        self.assertIn("transport_reset_hold_ = true;", prepare)
        self.assertIn('"streamResetPrepared', prepare)
        self.assertIn('!std::strcmp(mode, "datachannel")', prepare)
        self.assertIn('!std::strcmp(mode, "webtransport")', prepare)
        self.assertIn('channel_is_open(resource_dc_)', prepare)
        self.assertIn('ws_open(resource_ws_)', prepare)
        self.assertIn('detach_stale_planes("datachannel transport committed")',
                      prepare)
        self.assertIn('"streamResetCommitted', prepare)
        self.assertLess(prepare.index("transport_reset_hold_ = false;"),
                        prepare.index("schedule_drain_now();"))
        self.assertIn("void hold_for_forced_websocket_loss", RELAY)
        for plane in ("resource", "recovery", "frame"):
            self.assertIn(
                f'self->hold_for_forced_websocket_loss("{plane}");', RELAY)
        self.assertGreaterEqual(RELAY.count(
            "if (all_command_ws_forced_) return false;"), 3)
        self.assertIn("all_command_ws_forced_ = !std::strcmp", RELAY)
        self.assertIn('allCommands:wtActive', LIVE)
        self.assertIn('allCommands:relayMode!=="datachannel"', LIVE)
        self.assertIn('allCommands:true', LIVE)
        self.assertIn('realChannels.get("reliable") !== '
                      'selectedDataChannels[0]', WORKER)
        self.assertIn('!selectedDataChannels.every(planeOpen)', WORKER)
        self.assertIn('let activeTransportCandidate="";', LIVE)
        self.assertIn('return activeTransportCandidate==="datachannel";', LIVE)
        self.assertIn('activeTransportCandidate=mode;', self.recover)

    def test_forwarded_plane_lifecycle_is_epoch_guarded(self):
        forward = LIVE[LIVE.index("const forwardPlaneToWorker="):]
        forward = forward[:forward.index("if(renderWorker){")]
        self.assertIn("if(epoch!==forwardedPlaneEpoch||replayEnded", forward)
        self.assertIn('channel.addEventListener("error",failedPlane);', forward)
        self.assertIn('channel.addEventListener("close",failedPlane);', forward)
        self.assertIn("recoverCommandTransport(message);", forward)

    def test_transferred_datachannel_close_uses_warm_recovery(self):
        dispatcher = LIVE[LIVE.index('if(data.type==="chanState"){'):]
        dispatcher = dispatcher[:dispatcher.index("if(workerLateDispatch)")]
        self.assertIn("transferredPlaneIsActive(data.name)", dispatcher)
        self.assertIn(
            "recoverCommandTransport(`${data.name} channel ${data.state}`);",
            dispatcher)
        self.assertNotIn("failed(new Error", dispatcher)

    def test_frame_plane_selection_reaches_the_worker(self):
        handler = LIVE[LIVE.index('message.t==="framePlane"'):]
        handler = handler[:handler.index('message.t==="quality"')]
        self.assertIn(
            'renderWorker?.postMessage({type:"framePlane",plane:message.mode});',
            handler)
        self.assertIn('case "framePlane":', WORKER)

    def test_websocket_and_replaced_quic_bytes_stay_monotonic(self):
        accounting = LIVE[LIVE.index("if(view.wt){"):]
        accounting = accounting[:accounting.index("const receiveBps=")]
        self.assertIn("wtReceivedBytes+=counter>=wtReceivedCounter", accounting)
        for dataset in ("resourceAcked", "frameAcked", "recoveryAcked"):
            self.assertIn(dataset, accounting)
        self.assertIn(
            "const receivedBytes=webrtcBytes+wtReceivedBytes+websocketBytes;",
            accounting)

    def test_exact_recovery_arrival_controls_the_watchdog_deadline(self):
        handler = LIVE[LIVE.index('message.t==="recoveryStalled"'):]
        handler = handler[:handler.index('message.t==="sessionEnded"')]
        self.assertIn("expectedFrame:frame", handler)
        self.assertIn(
            "if(frame&&Number(workerView?.lastFrame||0)>frame)return;",
            handler)
        self.assertIn("frame===lastRecoveryStallFrame", handler)
        self.assertIn("stats.recoveryFragmentsArrived", handler)
        self.assertIn("stats.lastRecoveryFragmentFrame", handler)
        self.assertIn("stats.lastRecoveryFragmentAt", handler)
        self.assertIn("RECOVERY_FRAGMENT_IDLE_TIMEOUT_MS", handler)

        watchdog = LIVE[LIVE.index(
            "if(!spectatorMode&&recoveryArrivalWatch){"):]
        watchdog = watchdog[:watchdog.index("// Stuck-stream watchdog.")]
        self.assertIn("view.stats.lastRecoveryFrameArrived", watchdog)
        self.assertIn("fragmentFrame===exactFrame", watchdog)
        self.assertIn("watch.arrivalDeadline=now+"
                      "RECOVERY_FRAGMENT_IDLE_TIMEOUT_MS;", watchdog)
        self.assertIn("watch.applicationDeadline=now+"
                      "RECOVERY_APPLICATION_TIMEOUT_MS;", watchdog)
        self.assertIn('send({t:"recoveryReady",frame:exactFrame});', watchdog)
        self.assertIn("lastFrame>watch.unlockAfterFrame", watchdog)
        self.assertIn("RECOVERY_UNLOCK_TIMEOUT_MS", watchdog)
        self.assertNotIn("window.__w3csQuality", watchdog)

    def test_generic_watchdog_defers_to_exact_recovery_window(self):
        watchdog = LIVE[LIVE.index("// Stuck-stream watchdog."):]
        watchdog = watchdog[:watchdog.index('$("bandwidth")')]
        self.assertIn("if(recoveryArrivalWatch){", watchdog)
        self.assertLess(watchdog.index("if(recoveryArrivalWatch){"),
                        watchdog.index("++reliableStallSeconds>=25"))


class VideoLaneContractTests(unittest.TestCase):
    """The video player speaks the relay and catalog protocols the command
    player does, over one media peer instead of command planes."""

    def test_video_page_reports_its_feeder_and_uses_one_media_peer(self):
        self.assertIn('feeder:"video"', VIDEO)
        self.assertIn('document.documentElement.dataset.lane="video"', VIDEO)
        self.assertIn('type:"sdp",peer:"audio"', VIDEO)
        self.assertIn('event.channel.label!=="control"', VIDEO)
        self.assertIn('pickVideoSeat(', VIDEO)
        for absent in ("new WebTransport(", "render-worker.js",
                       "/resource-bundle/", "requestRecovery(", "wtConnect"):
            self.assertNotIn(absent, VIDEO, absent)

    def test_video_page_keeps_the_catalog_protocol(self):
        for message in ("w3-standby-ready", "w3-stream-info", "w3-replay-ended",
                        "w3-feeder-failed", "w3-close", "w3-position",
                        "w3-visibility", "w3-chat-open", "w3-chat-state"):
            self.assertIn(message, VIDEO, message)
        self.assertIn('handOffToFallback("launch-timeout")', VIDEO)
        self.assertIn('handOffToFallback("no-free-seat")', VIDEO)
        self.assertIn('handOffToFallback("viewer-busy")', VIDEO)

    def test_video_page_ends_the_replay_from_the_relay_scene_stats(self):
        self.assertIn('message.t==="videoStats"', VIDEO)
        self.assertIn("new ReplaySceneGuard()", VIDEO)
        self.assertIn('finishReplay("terminal-ui")', VIDEO)
        self.assertIn("castOverlay.setEngineClock(clock)", VIDEO)

    def test_video_starts_muted_and_unmutes_after_playback(self):
        self.assertIn('<video id="game" autoplay playsinline muted', VIDEO)
        self.assertIn('document.documentElement.dataset.audio="click-to-enable"', VIDEO)
        self.assertIn('"w3replays-volume"', VIDEO)