"""Integration: the snapshot stream over live sockets + the deterministic
fan-out drive (phase 06, task 04).

Live half (a real uvicorn on an ephemeral loopback port): the first
`PLAYING` frame's keysets (the shape `docs/PROTOCOL.md` documents —
the captured frame in the docs carries exactly these), the size
budgets (4 players / 4 orders / 3 hazards ≤ 1.5 KB on the budget
kitchen; a 64x64 level ≤ 4 KB), the ≥ 40% delta shrink, the spectator
half-rate pacing, the `UNDERCOOKED_SIM_LATENCY_MS` downlink delay, and
the 100-tick downlink bound (≤ 60 KB/s).

Deterministic half (a `FakeClock`-driven room + the `SnapshotFanout`
directly, no wall clock): the player/spectator pass pacing, the full
cadence, the delta decode-equivalence (the applied chain lands on the
server's own re-encode), the `MAX_EVENTS_PER_FRAME` cap's
`events_dropped` count, and the `needs_full` resync forcing a full.
"""

from __future__ import annotations

import asyncio
import contextlib
import json
import shutil
import time
from collections.abc import Callable
from pathlib import Path
from typing import Any, cast

import httpx
import pytest
import websockets

from undercooked.config import Settings
from undercooked.server.clock import FakeClock
from undercooked.server.encoding import (
    MAX_EVENTS_PER_FRAME,
    encode_snapshot,
    snapshot_size,
)
from undercooked.server.room import Room, RoomState
from undercooked.server.ws import ConnectionRegistry, SnapshotFanout
from undercooked.sim.content import ContentBundle
from undercooked.sim.level import Level, LevelDef
from undercooked.sim.model import (
    Event,
    ItemForm,
    ItemKind,
    StationKind,
)
from undercooked.sim.recipe import Ingredient, Recipe, RecipeBook, Step

pytestmark = pytest.mark.integration

#: The playable campaign (the shipped level 1, validated by the content
#: gate at server boot) — a room built from it can start a round.
PLAYABLE_LEVELS_DIR = (
    Path(__file__).resolve().parents[1] / "fixtures" / "levels_playable"
)

# The keysets the wire contract promises (docs/PROTOCOL.md "The `snap`
# frame" — the captured frame there carries exactly these; on the wire
# the frame rides under `raw`, beside the message's `type` and `p`).
FULL_TOP_KEYS: frozenset[str] = frozenset(
    {"t", "s", "tgt", "tl", "full", "p", "st", "o", "hz", "e"}
)
PLAYER_KEYS: frozenset[str] = frozenset({"i", "x", "y", "f", "h", "s", "pi"})
STATION_KEYS: frozenset[str] = frozenset({"k", "x", "y", "g", "q", "p", "s"})
ORDER_KEYS: frozenset[str] = frozenset({"id", "r", "l", "pts"})
HAZARD_KEYS: frozenset[str] = frozenset({"t", "x", "y", "l"})
STATION_KINDS: frozenset[str] = frozenset(
    {"spawn_shelf", "chop_board", "stove", "prep", "serve", "sink"}
)
PLAYER_STATES: frozenset[str] = frozenset(
    {"idle", "moving", "stunned", "working"}
)
HAZARD_KINDS: frozenset[str] = frozenset(
    {"fire", "spill", "moving_obstacle", "rush"}
)


# ------------------------------------------------------------------ helpers


def _uri(url: str, code: str) -> str:
    return url.replace("http://", "ws://", 1) + f"/ws/{code}"


async def _connect(
    url: str,
    code: str,
    token: str | None = None,
    *,
    name: str | None = None,
) -> Any:
    """A connected client with the `hello` already sent (a `token=None`
    hello mints a fresh token — a spectator: no seat for it)."""
    ws = await websockets.connect(_uri(url, code), open_timeout=10.0)
    hello: dict[str, Any] = {"type": "hello", "p": 1}
    if token is not None:
        hello["token"] = token
    if name is not None:
        hello["name"] = name
    await ws.send(json.dumps(hello))
    return ws


async def _next_json(ws: Any, wait_s: float = 10.0) -> dict[str, Any]:
    return json.loads(await asyncio.wait_for(ws.recv(), wait_s))


async def _wait_for(
    predicate: Callable[[], bool], what: str, wait_s: float = 15.0
) -> None:
    """Poll across threads (the server's loop does the work; this thread
    only observes it)."""
    loop = asyncio.get_running_loop()
    deadline = loop.time() + wait_s
    while not predicate():
        if loop.time() > deadline:
            pytest.fail(f"timed out waiting for {what}")
        await asyncio.sleep(0.01)


def _create_room(url: str, token: str | None = None) -> str:
    """`POST /api/rooms` over the live server; returns the code."""
    body: dict[str, Any] = {}
    if token is not None:
        body["token"] = token
    response = httpx.post(f"{url}/api/rooms", json=body, timeout=5.0)
    assert response.status_code == 201, response.text
    return response.json()["room_code"]


def _join(url: str, code: str, token: str) -> None:
    response = httpx.post(
        f"{url}/api/rooms/{code}/join", json={"token": token}, timeout=5.0
    )
    assert response.status_code == 200, response.text


async def _playing_room(
    url: str, app: Any, host_token: str
) -> tuple[str, Room, Any]:
    """Create the room, connect the host, start (solo host), and wait
    for `PLAYING` (the 3 s real-time countdown on the 20 Hz tick task)."""
    code = _create_room(url, host_token)
    host = await _connect(url, code, host_token, name="Host")
    await _next_json(host)  # welcome
    await _next_json(host)  # lobby
    await host.send(json.dumps({"type": "start"}))  # solo host
    room = app.state.supervisor.get(code)
    assert room is not None
    await _wait_for(
        lambda: room.state is RoomState.PLAYING and room.sim_state is not None,
        f"room {code} to reach PLAYING",
        wait_s=20.0,
    )
    return code, room, host


def _apply_delta(base: dict[str, Any], delta: dict[str, Any]) -> dict[str, Any]:
    """The client's merge (the decode half of the equivalence check):
    upsert by identity key (preserving the base's order, new entities
    appended — the encoder's own semantics), drop the `gone` tombstones,
    take the scalars and `e` from the delta."""
    # The scalars (`t`, `s`, `tgt`, `tl`) and `e` ride the delta; the
    # entity arrays merge onto the base.
    decoded = {
        key: value
        for key, value in delta.items()
        if key not in ("d", "full")
    }
    identity: dict[str, Any] = {
        "p": lambda e: (e["i"],),
        "st": lambda e: (e["k"], e["x"], e["y"]),
        "o": lambda e: (e["id"],),
        "hz": lambda e: (e["t"], e["x"], e["y"]),
    }
    for key in ("p", "st", "o", "hz"):
        base_entries: list[dict[str, Any]] = base.get(key, [])
        delta_entries: list[dict[str, Any]] = delta["d"][key]
        by_identity = {identity[key](entry): dict(entry) for entry in delta_entries}
        merged: list[dict[str, Any]] = []
        seen: set[tuple[Any, ...]] = set()
        for entry in base_entries:
            key_ = identity[key](entry)
            if by_identity.get(key_, {}).get("gone") is True:
                continue  # tombstoned
            merged.append(by_identity.get(key_, dict(entry)))
            seen.add(key_)
        for entry in delta_entries:
            key_ = identity[key](entry)
            if key_ not in seen and entry.get("gone") is not True:
                merged.append(dict(entry))
        decoded[key] = merged
    decoded["full"] = True  # a decoded chain is a full state
    return decoded


class _FakeConn:
    """A duck for the fan-out's connection surface (the deterministic
    drive): a role, the `needs_full` resync flag, an `enqueue` sink."""

    def __init__(self, role: str) -> None:
        self.role = role
        self.needs_full = True
        self.frames: list[dict[str, Any]] = []
        self.state_frames: list[dict[str, Any]] = []

    def enqueue(self, msg: Any) -> None:
        raw = getattr(msg, "raw", None)
        if raw is not None:
            self.frames.append(raw)
        else:
            self.state_frames.append(json.loads(msg.model_dump_json()))


class _FakeRegistry:
    """A duck for `ConnectionRegistry.for_room` (the deterministic drive)."""

    def __init__(self, conns: list[_FakeConn]) -> None:
        self._conns = conns

    def for_room(self, code: str) -> list[_FakeConn]:
        return list(self._conns)


def _make_book() -> RecipeBook:
    """A one-recipe book: the small levels' order pool names `soup`, and
    the room passes the book to `step`, so a due order spawn must find
    it."""
    return RecipeBook(
        recipes={
            "soup": Recipe(
                id="soup",
                name="Soup",
                base_score=40,
                ingredients=[
                    Ingredient(kind=ItemKind.TOMATO, form=ItemForm.RAW, count=1)
                ],
                steps=[Step(kind="serve", station=StationKind.SERVE, seconds=0.0)],
            )
        },
        ingredients=[ItemKind.TOMATO],
    )


def _make_level() -> Level:
    """A 7x5 kitchen with all six station kinds (the frame-size test
    needs a content-rich full for the ≥ 40% delta shrink) and one
    spawn point. `target` is above anything an idle cook scores, so a
    `PLAYING` round on it always ends `LOST` on the buzzer."""
    grid = [
        "#######",
        "#T.C.P#",
        "#...V.#",
        "#S..K.#",
        "#######",
    ]
    tiles: dict[str, dict[str, str]] = {
        ".": {"kind": "floor"},
        "#": {"kind": "wall"},
        "S": {"kind": "floor", "station": "spawn_shelf", "item": "tomato"},
        "V": {"kind": "floor", "station": "serve"},
        "T": {"kind": "floor", "station": "stove"},
        "C": {"kind": "floor", "station": "chop_board"},
        "P": {"kind": "floor", "station": "prep"},
        "K": {"kind": "floor", "station": "sink"},
    }
    data: dict[str, object] = {
        "id": "TICK1",
        "name": "Tick kitchen",
        "order": 1,
        "time_limit": 60.0,
        "target_score": 1000,
        "grid": grid,
        "tiles": tiles,
        "spawns": [[2, 2]],
        "orders": {
            "pool": ["soup"],
            "concurrent": 1,
            "spawn_interval": [4, 6],
            "deadline": 30,
        },
    }
    return Level.build(LevelDef.model_validate(data))


def _make_room(
    bundle: ContentBundle | None = None, **settings: Any
) -> tuple[Room, FakeClock]:
    clock = FakeClock()
    room = Room(
        code="TEST1",
        settings=Settings(**settings),
        clock=clock,
        content=bundle,
        level_id=bundle.levels[0].id if bundle and bundle.levels else None,
    )
    return room, clock


def _enter_playing(room: Room) -> float:
    """Two seats (the solo-host start + a ready non-host), the 3 s
    countdown in 1 s frames, `PLAYING`; returns the final `now`."""
    room.add_seat("tok-a", "Ada", True)
    seat = room.add_seat("tok-b", "Bob", False)
    assert room.set_ready("tok-b", True)
    assert seat is not None
    room.start("tok-a")
    room.tick_once(now=0.0)
    for offset in (1.0, 2.0, 3.0):
        room.tick_once(now=offset)
    assert room.state is RoomState.PLAYING
    return 3.0


# ------------------------------------------------- live: the keysets (L1)


async def test_first_playing_frame_is_full_with_the_contract_keysets(
    ws_server: Any,
) -> None:
    with ws_server(levels_dir=str(PLAYABLE_LEVELS_DIR)) as (url, app):
        _, _, host = await _playing_room(url, app, "tok-keys")
        # The countdown frames drain off first (they predate PLAYING).
        first: dict[str, Any] | None = None
        while first is None or first["type"] == "countdown":
            first = await _next_json(host)
        assert first is not None
        # The first `PLAYING` frame of a fresh connection is a full one
        # (its `needs_full` default). The wire wraps the frame in
        # `raw` (the message's own `p` is the protocol version).
        assert first["type"] == "snap"
        raw = first["raw"]
        assert raw["full"] is True
        # The keysets, exactly — the shape the docs' captured frame
        # carries (and no more: the grid, the station ids and the spawn
        # coordinates never leave the server).
        assert set(raw) == FULL_TOP_KEYS
        for player in raw["p"]:
            assert set(player) == PLAYER_KEYS
            assert player["f"] in (0, 1, 2, 3)
            assert player["s"] in PLAYER_STATES
            assert isinstance(player["pi"], int)
        for station in raw["st"]:
            assert set(station) == STATION_KEYS
            assert station["k"] in STATION_KINDS
        for order in raw["o"]:
            assert set(order) == ORDER_KEYS
        for hazard in raw["hz"]:
            assert set(hazard) == HAZARD_KEYS
            assert hazard["t"] in HAZARD_KINDS
        for entry in raw["e"]:
            assert "k" in entry
        # Level 1's kitchen, exactly: seven stations, the one solo
        # cook, no hazards (the tutorial's `hazards: []`).
        assert len(raw["st"]) == 7
        assert len(raw["p"]) == 1
        assert raw["hz"] == []
        assert raw["t"] >= 0
        assert raw["s"] == 0
        assert raw["tgt"] == 45


# ------------------------------------------------ live: the size budgets


BUDGET_LEVEL = """\
id: 00_budget
name: Budget Kitchen
order: 1
difficulty: 1
time_limit: 60
target_score: 100
blurb: >-
  The frame-budget kitchen: four spawn points, four concurrent orders,
  three fast hazards (the task's 4 players / 4 orders / 3 hazards case).
grid:
  - "############"
  - "#T..C...S..#"
  - "#..........#"
  - "#.P......X.#"
  - "#..........#"
  - "#..K....O..#"
  - "#..........#"
  - "#U......W..#"
  - "############"
tiles:
  "#": {kind: wall}
  ".": {kind: floor}
  "T": {kind: floor, station: spawn_shelf, item: tomato}
  "U": {kind: floor, station: spawn_shelf, item: tomato}
  "W": {kind: floor, station: spawn_shelf, item: onion}
  "O": {kind: floor, station: spawn_shelf, item: onion}
  "C": {kind: floor, station: chop_board}
  "S": {kind: floor, station: stove}
  "P": {kind: floor, station: prep}
  "X": {kind: floor, station: serve}
  "K": {kind: floor, station: sink}
spawns: [[1, 2], [1, 4], [1, 6], [5, 2]]
orders:
  pool: [soup]
  concurrent: 4
  spawn_interval: [1, 2]
  deadline: 30
hazards:
  - {type: fire, x: 8, y: 1, chance_per_s: 0.5, respawn_s: 2}
  - {type: fire, x: 4, y: 1, chance_per_s: 0.5, respawn_s: 2}
  - {type: spill, x: 5, y: 2, chance_per_s: 0.5, respawn_s: 2}
escalation_notes: >-
  Test content: the budget case, not a campaign level.
"""


def _level_64_yaml() -> str:
    """The 64x64 budget level (a wall ring, a row of stations, floor)."""
    top = "#" * 64
    station_row = "#" + "T..C.O.S.." + "." * 52 + "#"
    station_row_2 = "#" + "P..X..K..." + "." * 52 + "#"
    middle = "#" + "." * 62 + "#"
    grid = [top, station_row, station_row_2] + [middle] * 60 + [top]
    assert all(len(row) == 64 for row in grid)
    assert len(grid) == 64
    return (
        "id: 00_big\n"
        "name: Big Kitchen\n"
        "order: 1\n"
        "difficulty: 1\n"
        "time_limit: 60\n"
        "target_score: 100\n"
        "blurb: >-\n"
        "  The 64x64 budget: the frame must stay under 4 KB whatever the\n"
        "  grid (the grid never leaves the server — the frame is entities\n"
        "  only).\n"
        "grid:\n"
        + "".join(f'  - "{row}"\n' for row in grid)
        + "tiles:\n"
        '  "#": {kind: wall}\n'
        '  ".": {kind: floor}\n'
        '  "T": {kind: floor, station: spawn_shelf, item: tomato}\n'
        '  "O": {kind: floor, station: spawn_shelf, item: onion}\n'
        '  "C": {kind: floor, station: chop_board}\n'
        '  "S": {kind: floor, station: stove}\n'
        '  "P": {kind: floor, station: prep}\n'
        '  "X": {kind: floor, station: serve}\n'
        '  "K": {kind: floor, station: sink}\n'
        "spawns: [[2, 3]]\n"
        "orders:\n"
        "  pool: [soup]\n"
        "  concurrent: 1\n"
        "  spawn_interval: [16, 20]\n"
        "  deadline: 30\n"
        "hazards: []\n"
        "escalation_notes: >-\n"
        "  Test content: the 64x64 size case.\n"
    )


def _write_levels_dir(tmp_path: Path, level_yaml: str) -> Path:
    """A throwaway content dir: the test level + the campaign's recipe
    book (a `recipes.yaml` the content gate validates with it)."""
    directory = tmp_path / "levels"
    directory.mkdir()
    (directory / "00_test.yaml").write_text(level_yaml, encoding="utf-8")
    shutil.copy(PLAYABLE_LEVELS_DIR / "recipes.yaml", directory / "recipes.yaml")
    return directory


async def test_the_4p_4o_3h_frame_fits_the_1_5_kb_budget(
    ws_server: Any, tmp_path: Path
) -> None:
    with ws_server(
        levels_dir=str(_write_levels_dir(tmp_path, BUDGET_LEVEL))
    ) as (url, app):
        code = _create_room(url, "tok-p1")
        for token in ("tok-p2", "tok-p3", "tok-p4"):
            _join(url, code, token)
        clients = [
            await _connect(url, code, "tok-p1", name="P1"),
            await _connect(url, code, "tok-p2", name="P2"),
            await _connect(url, code, "tok-p3", name="P3"),
            await _connect(url, code, "tok-p4", name="P4"),
        ]
        for client in clients:
            await _next_json(client)  # welcome
            await _next_json(client)  # lobby
            await client.send(json.dumps({"type": "ready"}))
        await clients[0].send(json.dumps({"type": "start"}))
        room = app.state.supervisor.get(code)
        assert room is not None
        await _wait_for(
            lambda: room.state is RoomState.PLAYING and room.sim_state is not None,
            "the room to reach PLAYING",
            wait_s=20.0,
        )
        # Wait for one frame carrying the budget case (4 cooks, 4
        # orders) and measure *it* (orders spawn at 1-2 s intervals;
        # the sim's hazard stage is identity until phase 10, so a live
        # frame carries no `hz` yet — the hazard encoding is covered
        # unit-side, and the 4p/4o frame already bounds the 1.5 KB
        # budget: 3 hazards would add < 120 B).
        frame: dict[str, Any] | None = None
        deadline = time.monotonic() + 30.0
        while frame is None and time.monotonic() < deadline:
            data = await _next_json(clients[0])
            if data["type"] != "snap":
                continue
            body = data["raw"]
            if (
                len(body.get("p", [])) == 4
                and len(body.get("o", [])) >= 4
            ):
                frame = body
        assert frame is not None, "no frame carried the 4p/4o case in 30 s"
        size = snapshot_size(frame)
        assert size <= 1500, f"the budget frame is {size} bytes (budget 1.5 KB)"
        # The keysets hold at the budget shape too (four distinct cooks).
        assert set(frame) == FULL_TOP_KEYS
        assert len({p["i"] for p in frame["p"]}) == 4


async def test_the_64x64_level_frame_fits_the_4_kb_budget(
    ws_server: Any, tmp_path: Path
) -> None:
    with ws_server(
        levels_dir=str(_write_levels_dir(tmp_path, _level_64_yaml()))
    ) as (url, app):
        _, _, host = await _playing_room(url, app, "tok-big")
        frame: dict[str, Any] | None = None
        while frame is None:
            data = await _next_json(host)
            if data["type"] == "snap":
                frame = data["raw"]
        assert frame is not None
        assert frame["full"] is True  # the fresh connection's first frame
        size = snapshot_size(frame)
        assert size <= 4096, f"the 64x64 frame is {size} bytes (budget 4 KB)"
        # The grid is not in the frame: a 64x64 kitchen encodes to the
        # same handful of stations as the small one.
        assert len(frame["st"]) == 7


# ------------------------------------------------- live: the delta stream


async def test_the_delta_stream_shrinks_by_at_least_40_percent(
    ws_server: Any,
) -> None:
    with ws_server(levels_dir=str(PLAYABLE_LEVELS_DIR)) as (url, app):
        _, _, host = await _playing_room(url, app, "tok-delta")
        # Collect `snap` frames until the window holds a full one and
        # the frame right after it (the fresh connection's first is a
        # full; the cadence full, every 20 passes, lands in the window
        # too — either pair works).
        window: list[dict[str, Any]] = []
        deadline = time.monotonic() + 25.0
        while time.monotonic() < deadline:
            data = await _next_json(host)
            if data["type"] != "snap":
                continue
            window.append(data["raw"])
            for i, f in enumerate(window):
                if f["full"] and i + 1 < len(window):
                    break
            else:
                continue
            break
        full = next(f for f in window if f["full"])
        after = window[window.index(full) + 1]
        assert after["full"] is False, "the frame after the full must be a delta"
        full_size = snapshot_size(full)
        delta_size = snapshot_size(after)
        shrink = 1.0 - delta_size / full_size
        assert shrink >= 0.40, (
            f"the delta shrank by {shrink:.0%} (full {full_size} B, "
            f"delta {delta_size} B; budget ≥ 40%)"
        )
        # The decode half (live): applying the delta to the full must
        # land on exactly the base's entities (kept) plus the delta's
        # (upserted), minus the tombstoned — and the delta's scalars.
        decoded = _apply_delta(full, after)
        assert decoded["t"] == after["t"]
        assert decoded["s"] == after["s"]
        assert decoded["full"] is True
        identity = {
            "p": lambda e: (e["i"],),
            "st": lambda e: (e["k"], e["x"], e["y"]),
            "o": lambda e: (e["id"],),
            "hz": lambda e: (e["t"], e["x"], e["y"]),
        }
        for key in ("p", "st", "o", "hz"):
            base_ids = {identity[key](e) for e in full[key]}
            delta_ids = {
                identity[key](e)
                for e in after["d"][key]
                if e.get("gone") is not True
            }
            gone_ids = {
                identity[key](e)
                for e in after["d"][key]
                if e.get("gone") is True
            }
            got = {identity[key](e) for e in decoded[key]}
            assert got == (base_ids - gone_ids) | delta_ids, (
                f"the merge lost or invented {key} entities: {got} vs "
                f"{(base_ids - gone_ids) | delta_ids}"
            )


# ------------------------------------------------- live: the pacing


async def test_spectators_receive_half_the_rate(ws_server: Any) -> None:
    with ws_server(levels_dir=str(PLAYABLE_LEVELS_DIR)) as (url, app):
        code, _, host = await _playing_room(url, app, "tok-rate")
        # The spectator joins by a token-less hello (a minted token has
        # no seat — the documented spectator path).
        spectator = await _connect(url, code, None, name="Spectator")
        await _next_json(spectator)  # welcome
        await _next_json(spectator)  # lobby
        await asyncio.sleep(0.3)  # let the spectator's first full flush
        # A 3.0 s window: the default snapshot rate is the 20 Hz tick
        # (period 1 — a player pass every tick), the spectator pass at
        # the half rate (every 2nd tick, 10 Hz).
        player_bucket: dict[str, int] = {"n": 0}
        spec_bucket: dict[str, int] = {"n": 0}
        host_q = asyncio.create_task(_count(host, "snap", player_bucket))
        spec_q = asyncio.create_task(_count(spectator, "snap", spec_bucket))
        await asyncio.sleep(3.0)
        host_q.cancel()
        spec_q.cancel()
        for task in (host_q, spec_q):
            with contextlib.suppress(asyncio.CancelledError):
                await task
        player_count = player_bucket["n"]
        spectator_count = spec_bucket["n"]
        assert player_count >= 50, f"the player got {player_count} frames in 3 s (20 Hz)"
        assert 22 <= spectator_count <= 38, (
            f"the spectator got {spectator_count} frames in 3 s (10 Hz)"
        )
        assert player_count >= 2 * spectator_count - 10, (
            f"the player's {player_count} is not ~2x the spectator's "
            f"{spectator_count}"
        )


async def _count(ws: Any, frame_type: str, bucket: dict[str, int]) -> None:
    """Count `frame_type` frames on a live socket until cancelled (the
    pacing window's half)."""
    bucket["n"] = 0
    while True:
        data = json.loads(await ws.recv())
        if data["type"] == frame_type:
            bucket["n"] += 1


# ------------------------------------------------- live: the latency


async def test_the_configured_sim_latency_bounds_the_downlink(
    ws_server: Any,
) -> None:
    with ws_server(levels_dir=str(PLAYABLE_LEVELS_DIR), sim_latency_ms=150) as (
        url,
        app,
    ):
        _, _, host = await _playing_room(url, app, "tok-latency")
        # The pump's per-batch delay: the next `snap` after an input is
        # written only after the pump's 150 ms sleep — the input→frame
        # round trip is lower-bounded by it.
        await host.send(
            json.dumps(
                {
                    "type": "input",
                    "seq": 1,
                    "move": {"x": 1.0, "y": 0.0},
                    "act": False,
                }
            )
        )
        t0 = time.monotonic()
        frame: dict[str, Any] | None = None
        while frame is None:
            data = await _next_json(host)
            if data["type"] == "snap":
                frame = data
        elapsed = time.monotonic() - t0
        assert elapsed >= 0.145, (
            f"the input→frame round trip took {elapsed * 1000:.0f} ms; the "
            f"configured 150 ms delay must lower-bound it"
        )


# ------------------------------------------------- live: the rate bound


async def test_the_downlink_is_bounded_over_hundred_ticks(ws_server: Any) -> None:
    with ws_server(levels_dir=str(PLAYABLE_LEVELS_DIR)) as (url, app):
        _, _, host = await _playing_room(url, app, "tok-bound")
        # 100 ticks = 5.0 s of wall clock (the 20 Hz tick task): the
        # total bytes the player receives must stay under 60 KB.
        total = 0
        frame_count = 0
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            try:
                data = await asyncio.wait_for(host.recv(), timeout=0.5)
            except TimeoutError:
                continue
            frame = json.loads(data)
            if frame["type"] == "snap":
                frame_count += 1
            total += len(data.encode("utf-8"))
        assert frame_count >= 30, (
            f"only {frame_count} snap frames in 5 s (10 Hz expected)"
        )
        assert total <= 60_000, f"the 100-tick downlink was {total} bytes"


# ======================================= deterministic: the fan-out drive


def test_the_pass_pacing_full_cadence_and_decode_equivalence() -> None:
    """The defaults (the 20 Hz tick, the 20 Hz snapshot — period 1,
    so the player pass is due every tick; the spectator pass at the
    half rate, every 2 ticks). The first pass of a fresh audience is a
    full (its `needs_full` default), the cadence fulls land at passes
    1, 21, 41, … (a full every 20 passes), and applying the delta
    chain must land on the server's own re-encode of the final state
    (the decode-equivalence the client's merge owes)."""
    room, _clock = _make_room(
        bundle=ContentBundle(recipe_book=_make_book(), levels=(_make_level(),))
    )
    player = _FakeConn("player")
    spectator = _FakeConn("spectator")
    fanout = SnapshotFanout(
        room,
        cast(ConnectionRegistry, _FakeRegistry([player, spectator])),
    )
    room.set_broadcaster(fanout.pass_)
    now = _enter_playing(room)
    # 21 PLAYING ticks at the 20 Hz cadence: 21 player passes (period
    # 1), 10 spectator passes (every 2nd tick).
    for _ in range(21):
        now += 0.05
        room.tick_once(now=now)
    assert len(player.frames) == 21, (
        f"21 player passes expected, got {len(player.frames)}"
    )
    assert len(spectator.frames) == 10, (
        f"10 spectator passes expected, got {len(spectator.frames)}"
    )
    # The fresh connection's first pass is a full; passes 2-20 are
    # deltas; pass 21 is the cadence full (a full every 20 passes —
    # one at 1, 21, 41, …).
    assert player.frames[0]["full"] is True
    for frame in player.frames[1:20]:
        assert frame["full"] is False
    assert player.frames[20]["full"] is True
    assert spectator.frames[0]["full"] is True
    for frame in spectator.frames[1:]:
        assert frame["full"] is False
    # The ≥ 40% shrink on the delta right after the fresh full (an
    # idle kitchen: only the clock and the tick move).
    full_size = snapshot_size(player.frames[0])
    delta_size = snapshot_size(player.frames[1])
    assert 1.0 - delta_size / full_size >= 0.40, (
        f"the delta shrank by {1.0 - delta_size / full_size:.0%} "
        f"(full {full_size} B, delta {delta_size} B)"
    )
    # The decode-equivalence: apply the chain (a full frame re-syncs
    # the client state; a delta merges on top) through all 20 frames
    # after the first and compare to the server's own re-encode of the
    # final state (an idle kitchen is static apart from the clock).
    state = room.sim_state
    assert state is not None
    reencoded = encode_snapshot(state)
    decoded = player.frames[0]
    for frame in player.frames[1:21]:
        # A full frame re-syncs the client state; a delta merges on top.
        decoded = frame if frame["full"] else _apply_delta(decoded, frame)
    for key in ("p", "st", "hz", "o"):
        assert decoded[key] == reencoded[key], f"the chain lost its {key}"
    # The last pass encodes *before* the tick's own step (the room's
    # pacing runs first), so the chain is exactly one tick behind the
    # re-encode — and the idle kitchen's entities are identical in
    # both.
    assert decoded["t"] == reencoded["t"] - 1
    assert decoded["s"] == reencoded["s"]
    assert decoded["tgt"] == reencoded["tgt"]


def test_the_event_cap_counts_the_drop() -> None:
    """`MAX_EVENTS_PER_FRAME` (24) per frame: the 30th event is not
    spread over the next frame — it counts in
    `Room.stats()["events_dropped"]`, on the `ev` flush (a non-`PLAYING`
    boundary) and on the `PLAYING` snapshot pass alike. The room's
    buffer is not drained by the wire (the fan-out's cursor reads it
    non-destructively)."""
    room, _clock = _make_room(
        bundle=ContentBundle(recipe_book=_make_book(), levels=(_make_level(),))
    )
    sink = _FakeConn("player")
    fanout = SnapshotFanout(
        room, cast(ConnectionRegistry, _FakeRegistry([sink]))
    )
    room.set_broadcaster(fanout.pass_)
    # 30 emote events into the room's buffer (a non-`PLAYING` room:
    # the `ev` flush path).
    for index in range(30):
        room._buffer_event(
            _emote("tok-a", f"n{index}")
        )
    assert len(room.events_since_snapshot) == 30
    fanout.pass_({"ev"}, ())
    ev = sink.state_frames[-1]
    assert ev["type"] == "ev"
    assert len(ev["events"]) == MAX_EVENTS_PER_FRAME
    assert room.stats()["events_dropped"] == 30 - MAX_EVENTS_PER_FRAME
    # The buffer is intact (the cursor is non-destructive).
    assert len(room.events_since_snapshot) == 30
    # And the `PLAYING` path caps the same way (30 more events, two
    # ticks — the player pass is due on the 2nd).
    now = _enter_playing(room)
    for index in range(30):
        room._buffer_event(_emote("tok-b", f"m{index}"))
    now += 0.05
    room.tick_once(now=now)  # the first player pass (period 1: due now)
    snap = sink.frames[-1]
    assert len(snap["e"]) == MAX_EVENTS_PER_FRAME
    assert room.stats()["events_dropped"] == 2 * (30 - MAX_EVENTS_PER_FRAME)
    assert len(room.events_since_snapshot) == 60


def _emote(player_id: str, emote_id: str) -> Event:
    """A lobby emote event (the `ev` flush's payload; the kind is the
    server layer's plain string — `seats.EMOTE_EVENT_KIND`)."""
    return Event(kind="emote", payload={"player_id": player_id, "emote": emote_id})


def test_a_forced_needs_full_restarts_the_chain_full() -> None:
    """The send queue's drop policy marks `needs_full` (a client that
    has missed a frame): the audience's next pass is a full again, and
    the stream's `prev` resets (the next delta diffs against it)."""
    room, _clock = _make_room(
        bundle=ContentBundle(recipe_book=_make_book(), levels=(_make_level(),))
    )
    player = _FakeConn("player")
    fanout = SnapshotFanout(
        room, cast(ConnectionRegistry, _FakeRegistry([player]))
    )
    room.set_broadcaster(fanout.pass_)
    now = _enter_playing(room)
    # Three ticks: a pass on each (period 1): full, delta, delta.
    for _ in range(3):
        now += 0.05
        room.tick_once(now=now)
    assert [f["full"] for f in player.frames] == [True, False, False]
    # A dropped snapshot forces the next pass full.
    player.needs_full = True
    now += 0.05
    room.tick_once(now=now)
    assert player.frames[-1]["full"] is True
    # The chain resets: the following pass is a delta against the new
    # full (not against the stale `prev`).
    now += 0.05
    room.tick_once(now=now)
    assert player.frames[-1]["full"] is False