undercooked/tests/integration/test_lobby.py

"""Integration: the lobby (phase 05, task 03; US-02, US-03) — seats,
ready-up, the host's authority, level picking and spectators.

The contract, driven by hand on a `FakeClock` (no test sleeps):

* **ready-up** — `Seat.toggle_ready` / the room's `set_ready`, visible in
  `lobby_payload` (the `ready` flags and `can_start`: a solo host, or ≥ 1
  ready non-host, and a selected level, and a crown holder),
* **start** — `Room.start(token)`: `LOBBY` only, host only
  (`NotInLobby` / `NotHost`), then task 02's preconditions as
  `NothingToStart` (a `StartRefused` — the reason rides along),
* **level pick** — `Room.set_level(token, level_id)`: `LOBBY` only, host
  only, the id in the campaign (`UnknownLevel`) and its `order` within
  the requesting session's `unlocked_level` (`LevelLocked`; a token
  without a session record counts as unlocked 1),
* **the host rules** — the creator's seat 0 is the host; when the crown
  holder leaves or their connection drops in `LOBBY`/`ROUND_END` the
  crown moves to the lowest occupied, *connected* seat; in
  `COUNTDOWN`/`PLAYING` the host slot is emptied and **no mid-round
  election happens** (asserted explicitly); a returning host does not
  automatically regain the crown,
* **spectators** — a `join` in `COUNTDOWN`/`PLAYING`/`ROUND_END` adds a
  spectator (no seat, `role="spectator"`, a colour from the same
  first-free palette); a spectator's input is an `err not_seated`; a
  `leave` removes the entry; `to_lobby` promotes them into the free
  seats (PLAN §6: "late joins wait in `SPECTATING` until `LOBBY`"),
* **colours** — the 4-colour palette, first free in seat order; a
  token's colour is remembered for the life of the room (a reconnect
  keeps it),
* **emotes** — the `EMOTES` whitelist (`UnknownEmote`), the 2-per-1.0 s
  per-token throttle, re-broadcast as an `ev` event,
* **names** — sanitized at the boundary (a markup name comes back ≤ 16
  chars, markup-free).
"""

from __future__ import annotations

import copy
from typing import Any

import pytest

from undercooked.config import Settings
from undercooked.server.clock import FakeClock
from undercooked.server.room import Room, RoomState, StartRefused
from undercooked.server.rooms import PlayerNotInRoom, RoomSupervisor
from undercooked.server.seats import (
    EMOTE_EVENT_KIND,
    EMOTES,
    SEAT_PALETTE,
    LevelLocked,
    NothingToStart,
    NotHost,
    NotInLobby,
    Seat,
    Spectator,
    UnknownEmote,
    UnknownLevel,
)
from undercooked.server.sessions import SessionRecord, new_session
from undercooked.sim.content import ContentBundle
from undercooked.sim.level import Level, LevelDef
from undercooked.sim.model import EventKind, ItemForm, ItemKind, StationKind
from undercooked.sim.recipe import Ingredient, Recipe, RecipeBook, Step
from undercooked.sim.step import PlayerInput

pytestmark = pytest.mark.integration

# ------------------------------------------------------------------ fixtures


def make_book() -> RecipeBook:
    """A one-recipe book: the small levels' order pool names `soup`."""
    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_id: str, order: int, *, time_limit: float = 3.0, target: int = 1000
) -> Level:
    """A 7x4 kitchen (one spawn shelf, one serve window) at campaign
    `order`; `target` above any idle score, so a round ends `LOST` on
    the buzzer — the shortest path to `ROUND_END`."""
    grid = [
        "#######",
        "#S...V#",
        "#.....#",
        "#######",
    ]
    tiles: dict[str, dict[str, str]] = {
        ".": {"kind": "floor"},
        "#": {"kind": "wall"},
        "S": {"kind": "floor", "station": "spawn_shelf", "item": "tomato"},
        "V": {"kind": "floor", "station": "serve"},
    }
    data: dict[str, object] = {
        "id": level_id,
        "name": f"Lobby kitchen {level_id}",
        "order": order,
        "time_limit": time_limit,
        "target_score": target,
        "grid": copy.deepcopy(grid),
        "tiles": copy.deepcopy(tiles),
        "spawns": [[2, 2]],
        "orders": {
            "pool": ["soup"],
            "concurrent": 1,
            "spawn_interval": [4, 6],
            "deadline": 30,
        },
    }
    return Level.build(LevelDef.model_validate(data))


def make_bundle(*levels: Level) -> ContentBundle:
    return ContentBundle(recipe_book=make_book(), levels=levels)


def two_level_bundle() -> ContentBundle:
    return make_bundle(make_level("05a_first", 1), make_level("05b_second", 2))


def make_room(
    bundle: ContentBundle | None = None,
    sessions: dict[str, SessionRecord] | None = None,
    **settings: Any,
) -> tuple[Room, FakeClock]:
    """A `LOBBY` room on a `FakeClock`; `sessions` (or its absence) is
    the room's token → session lookup (`None` = no lookup: a token counts
    as unlocked 1)."""
    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,
        sessions=(sessions or {}).get,
    )
    return room, clock


def make_supervisor(
    bundle: ContentBundle | None = None, **settings: Any
) -> tuple[RoomSupervisor, FakeClock]:
    clock = FakeClock()
    supervisor = RoomSupervisor(Settings(**settings), clock, bundle or two_level_bundle())
    return supervisor, clock


def prime(room: Room, now: float = 0.0) -> None:
    room.tick_once(now=now)


def drive_countdown(room: Room, now: float = 0.0) -> float:
    """The 3 s countdown in 1 s frames; returns the final `now`, with the
    room in `PLAYING` (the round built)."""
    prime(room, now)
    for offset in (1.0, 2.0, 3.0):
        room.tick_once(now=now + offset)
    assert room.state is RoomState.PLAYING
    return now + 3.0


def play_round_to_end(room: Room, now: float = 3.0) -> float:
    """1 s frames through the 3.0 s round and its 1.0 s settle window,
    until `ROUND_END`; returns the final `now`."""
    while room.state is RoomState.PLAYING:
        now += 1.0
        room.tick_once(now=now)
    assert room.state is RoomState.ROUND_END
    return now


def seat(room: Room, token: str) -> Seat:
    """Narrow `seat_for_token` for the tests (the token is seated)."""
    got = room.seat_for_token(token)
    assert got is not None
    return got


def watching(room: Room, token: str) -> Spectator:
    """Narrow `spectator_for_token` for the tests (the token is watching)."""
    got = room.spectator_for_token(token)
    assert got is not None
    return got


def seat_host_and_ready(room: Room) -> None:
    room.add_seat("tok-a", "Ada", True)
    room.add_seat("tok-b", "Bob", False)
    assert room.set_ready("tok-b", True)


# ---------------------------------------------------------------- ready / start


def test_ready_toggles_are_visible_in_the_lobby_payload() -> None:
    room, _ = make_room(two_level_bundle())
    room.add_seat("tok-a", "Ada", True)
    room.add_seat("tok-b", "Bob", False)
    cy = room.add_seat("tok-c", "Cy", False)

    payload: dict[str, Any] = room.lobby_payload()
    assert [seat["ready"] for seat in payload["seats"]] == [False, False, False]
    assert payload["can_start"] is False  # a host, a level, but nobody ready

    assert cy.toggle_ready() is True
    assert cy.toggle_ready() is False  # the toggle flips back
    assert seat(room, "tok-b").toggle_ready() is True

    lobby_view: dict[str, Any] = room.lobby_payload()
    ready = {seat["index"]: seat["ready"] for seat in lobby_view["seats"]}
    assert ready == {0: False, 1: True, 2: False}
    assert room.lobby_payload()["can_start"] is True  # one ready non-host is enough

    assert room.set_ready("tok-b", False)
    assert room.lobby_payload()["can_start"] is False


def test_solo_start_is_allowed_and_the_payload_says_can_start() -> None:
    room, _ = make_room(two_level_bundle())
    room.add_seat("tok-a", "Ada", True)
    assert room.lobby_payload()["can_start"] is True  # a solo host: no ready needed
    room.start("tok-a")
    assert room.state is RoomState.COUNTDOWN
    assert room.round_no == 1


def test_start_requires_a_ready_non_host_and_names_the_refusal() -> None:
    room, _ = make_room(two_level_bundle())
    seat_host_and_ready(room)
    seat(room, "tok-b").ready = False  # the ready-up is not there yet

    with pytest.raises(NothingToStart, match="no_ready_players") as exc_info:
        room.start("tok-a")
    assert exc_info.value.reason == "no_ready_players"
    assert isinstance(exc_info.value, StartRefused)  # the task-02 contract rides along
    assert room.state is RoomState.LOBBY  # the refusal leaves no state change

    room.set_ready("tok-b", True)
    room.start("tok-a")
    assert room.state is RoomState.COUNTDOWN


def test_start_without_a_level_refuses_no_level() -> None:
    room, _ = make_room(None)  # no content bundle: no level selected
    room.add_seat("tok-a", "Ada", True)
    with pytest.raises(NothingToStart, match="no_level"):
        room.start("tok-a")
    assert room.state is RoomState.LOBBY


def test_start_and_set_level_by_a_non_host_are_rejected() -> None:
    bundle = two_level_bundle()
    sessions = {"tok-b": SessionRecord("tok-b", "Bob", 0.0, 2)}
    room, _ = make_room(bundle, sessions)
    room.add_seat("tok-a", "Ada", True)
    room.add_seat("tok-b", "Bob", False)
    room.set_ready("tok-b", True)  # so the host's start can run past the guards

    with pytest.raises(NotHost):
        room.start("tok-b")
    with pytest.raises(NotHost):
        room.start("tok-stranger")
    with pytest.raises(NotHost):
        room.set_level("tok-b", "05a_first")

    # Outside the lobby, neither verb runs — even for the host.
    room.start("tok-a")
    assert room.state is RoomState.COUNTDOWN
    with pytest.raises(NotInLobby):
        room.start("tok-a")
    with pytest.raises(NotInLobby):
        room.set_level("tok-a", "05a_first")


# ------------------------------------------------------------- level picking


def test_set_level_requires_a_known_unlocked_level() -> None:
    sessions = {
        "tok-a": SessionRecord("tok-a", "Ada", 0.0, 1),
        "tok-b": SessionRecord("tok-b", "Bob", 0.0, 2),
    }
    room, _ = make_room(two_level_bundle(), sessions)
    room.add_seat("tok-a", "Ada", True)

    with pytest.raises(UnknownLevel):
        room.set_level("tok-a", "05_zzz")
    with pytest.raises(LevelLocked, match="order 2"):
        room.set_level("tok-a", "05b_second")  # the session is unlocked 1
    assert room.level_id == "05a_first"  # the refusal leaves the pick unchanged

    room.add_seat("tok-b", "Bob", False)
    seat(room, "tok-a").host = False  # the crown moves to Bob (unlocked 2)
    seat(room, "tok-b").host = True
    level = room.set_level("tok-b", "05b_second")
    assert level.id == "05b_second"
    assert room.level_id == "05b_second"


def test_set_level_without_a_session_lookup_counts_as_unlocked_1() -> None:
    room, _ = make_room(two_level_bundle())  # no sessions map at all
    room.add_seat("tok-a", "Ada", True)
    with pytest.raises(LevelLocked):
        room.set_level("tok-a", "05b_second")
    with pytest.raises(UnknownLevel):
        room.set_level("tok-a", "05_zzz")
    picked = room.set_level("tok-a", "05a_first")  # the floor is always pickable
    assert picked.id == "05a_first"


def test_set_level_refuses_an_id_outside_a_bundleless_campaign() -> None:
    room, _ = make_room(None)  # no content bundle: no campaign to pick from
    room.add_seat("tok-a", "Ada", True)
    with pytest.raises(UnknownLevel):
        room.set_level("tok-a", "05a_first")


# ---------------------------------------------------------------- the host rules


def test_host_handoff_in_the_lobby_goes_to_the_lowest_connected_seat() -> None:
    room, _ = make_room(two_level_bundle())
    room.add_seat("tok-a", "Ada", True)
    room.add_seat("tok-b", "Bob", False)
    room.add_seat("tok-c", "Cy", False)

    # A dropped host connection is a host leave: the crown moves to the
    # lowest occupied, *connected* seat.
    assert room.mark_seat_disconnected("tok-a")
    assert not seat(room, "tok-a").host
    assert seat(room, "tok-b").host is True  # seat 1, the lowest connected
    assert room.lobby_payload()["host_seat"] == 1

    # The host's leave does the same: on to the next connected seat.
    assert room.remove_seat("tok-b")
    assert seat(room, "tok-c").host is True  # seat 2

    # When no connected seat remains, the crown is left empty.
    room.mark_seat_disconnected("tok-c")
    room.mark_seat_disconnected("tok-a")
    assert not any(seat.host for seat in room.seats)
    assert room.lobby_payload()["host_seat"] is None
    assert room.lobby_payload()["can_start"] is False  # nobody can press Start


def test_host_handoff_in_round_end_elects_the_lowest_connected_seat() -> None:
    room, _ = make_room(two_level_bundle())
    seat_host_and_ready(room)
    room.start("tok-a")
    now = drive_countdown(room)
    now = play_round_to_end(room, now)
    assert room.state is RoomState.ROUND_END

    assert room.mark_seat_disconnected("tok-a")
    assert seat(room, "tok-b").host is True  # elected in ROUND_END
    assert room.lobby_payload()["host_seat"] == 1


def test_no_mid_round_host_election_happens() -> None:
    """The documented rule, asserted explicitly (not by omission): in
    `COUNTDOWN`/`PLAYING` the host slot is emptied when the host leaves
    or drops — and the round continues, with no election, until
    `ROUND_END` (and no election happens on the state change either)."""
    room, _ = make_room(two_level_bundle())
    seat_host_and_ready(room)
    room.start("tok-a")
    now = drive_countdown(room)  # PLAYING

    assert room.mark_seat_disconnected("tok-a")
    assert all(not seat.host for seat in room.seats)  # the host slot is empty
    assert room.lobby_payload()["host_seat"] is None
    assert room.state is RoomState.PLAYING  # the round continues

    now = play_round_to_end(room, now)
    assert all(not seat.host for seat in room.seats)  # still empty at ROUND_END
    assert room.lobby_payload()["host_seat"] is None
    assert room.lobby_payload()["can_start"] is False

    # The host's *leave* mid-round: the same rule.
    room2, _ = make_room(two_level_bundle())
    seat_host_and_ready(room2)
    room2.start("tok-a")
    drive_countdown(room2)
    assert room2.remove_seat("tok-a")
    assert all(not seat.host for seat in room2.seats)
    assert room2.state is RoomState.PLAYING


def test_a_returning_host_does_not_regain_the_crown() -> None:
    room, _ = make_room(two_level_bundle())
    room.add_seat("tok-a", "Ada", True)
    room.add_seat("tok-b", "Bob", False)
    room.mark_seat_disconnected("tok-a")  # the crown moves to Bob
    assert seat(room, "tok-b").host is True

    # Ada's connection returns: the seat is live again, the crown stays.
    assert room.mark_seat_connected("tok-a")
    assert seat(room, "tok-a").connected is True
    assert not seat(room, "tok-a").host
    assert seat(room, "tok-b").host is True

    # Ada leaves the lobby and re-joins: the fresh seat is not the host.
    room.remove_seat("tok-a")
    again = room.add_seat("tok-a", "Ada", False)
    assert again.host is False
    assert seat(room, "tok-b").host is True


# --------------------------------------------------------------- spectators


def test_join_during_play_becomes_a_spectator() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    ada = new_session("Ada", clock)
    room = supervisor.create_room(ada, "Ada")
    bob = new_session("Bob", clock)
    supervisor.join(room.code, bob.token, "Bob")
    assert room.set_ready(bob.token, True)
    room.start(ada.token)
    drive_countdown(room)  # PLAYING

    cy = new_session("Cy", clock)
    supervisor.join(room.code, cy.token, "Cy")
    assert room.seat_for_token(cy.token) is None  # no seat
    spec = room.spectator_for_token(cy.token)
    assert spec is not None
    assert spec.role == "spectator" and spec.connected
    assert spec.display_name == "Cy"
    assert spec.color == SEAT_PALETTE[2]  # first free, after the two seats

    payload = room.lobby_payload(ada.token)
    assert payload["spectators"] == [{"name": "Cy", "color": SEAT_PALETTE[2]}]
    assert supervisor.stats()["spectators"] == 1
    # The token index still finds the watcher (by_token works for them).
    assert supervisor.by_token(cy.token) is room


def test_a_full_room_still_accepts_spectators() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    eve = new_session("Eve", clock)
    room = supervisor.create_room(eve, "Eve")
    tokens = [eve.token]
    for name in ("Fay", "Gus", "Hal"):
        token = new_session(name, clock).token
        supervisor.join(room.code, token, name)
        tokens.append(token)
    assert len(room.seats) == 4  # max_players
    room.set_ready(tokens[1], True)
    room.start(eve.token)
    drive_countdown(room)

    imran = new_session("Imran", clock)
    supervisor.join(room.code, imran.token, "Imran")  # the 5th player: a watcher
    assert room.seat_for_token(imran.token) is None
    assert room.spectator_for_token(imran.token) is not None
    # Four seats take the whole palette: the late watcher has no colour.
    assert watching(room, imran.token).color == ""


def test_join_during_countdown_becomes_a_spectator() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    ida = new_session("Ida", clock)
    room = supervisor.create_room(ida, "Ida")
    room.start(ida.token)  # a solo host: the countdown runs
    prime(room)  # the 3 s is in flight

    jon = new_session("Jon", clock)
    supervisor.join(room.code, jon.token, "Jon")
    assert room.seat_for_token(jon.token) is None
    assert room.spectator_for_token(jon.token) is not None


def test_spectator_input_is_rejected_with_not_seated() -> None:
    """A spectator's `input` is ignored — as a named `err`, not a
    silent no-op (and distinct from `unknown_player`)."""
    supervisor, clock = make_supervisor(two_level_bundle())
    ada = new_session("Ada", clock)
    room = supervisor.create_room(ada, "Ada")
    room.start(ada.token)
    drive_countdown(room)

    cy = new_session("Cy", clock)
    supervisor.join(room.code, cy.token, "Cy")
    assert room.queue_input(cy.token, PlayerInput(player_id=cy.token)) is False

    events = room.take_events()
    assert len(events) == 1
    assert events[0].kind is EventKind.ERROR
    assert events[0].payload == {
        "reason": "not_seated",
        "player_id": cy.token,
        "seq": 0,
    }


def test_spectator_leave_removes_the_entry_and_disconnect_keeps_it() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    ada = new_session("Ada", clock)
    room = supervisor.create_room(ada, "Ada")
    room.start(ada.token)
    drive_countdown(room)

    cy = new_session("Cy", clock)
    supervisor.join(room.code, cy.token, "Cy")
    supervisor.mark_disconnected(room.code, cy.token)  # the entry is kept
    spec = room.spectator_for_token(cy.token)
    assert spec is not None and spec.connected is False

    supervisor.leave(room.code, cy.token)
    assert room.spectator_for_token(cy.token) is None
    assert supervisor.stats()["spectators"] == 0
    with pytest.raises(PlayerNotInRoom):
        supervisor.leave(room.code, cy.token)


def test_spectator_rejoin_keeps_the_entry_and_the_color() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    ada = new_session("Ada", clock)
    room = supervisor.create_room(ada, "Ada")
    room.start(ada.token)
    drive_countdown(room)

    cy = new_session("Cy", clock)
    supervisor.join(room.code, cy.token, "Cy")
    color = watching(room, cy.token).color
    assert color == SEAT_PALETTE[1]  # first free, after Ada's seat

    supervisor.mark_disconnected(room.code, cy.token)
    supervisor.join(room.code, cy.token, "Cy")  # the re-watch
    spec = watching(room, cy.token)
    assert spec.connected is True
    assert spec.color == color  # the same entry, the same colour


def test_spectators_are_promoted_to_free_seats_on_to_lobby() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    ada = new_session("Ada", clock)
    room = supervisor.create_room(ada, "Ada")
    bob = new_session("Bob", clock)
    supervisor.join(room.code, bob.token, "Bob")
    assert room.set_ready(bob.token, True)
    room.start(ada.token)
    now = drive_countdown(room)
    cy = new_session("Cy", clock)
    di = new_session("Di", clock)
    supervisor.join(room.code, cy.token, "Cy")
    supervisor.join(room.code, di.token, "Di")
    now = play_round_to_end(room, now)

    room.to_lobby()
    assert room.state is RoomState.LOBBY
    assert len(room.seats) == 4  # both watchers were promoted
    assert room.spectators == []
    assert not any(seat.ready for seat in room.seats)  # a fresh lobby readies up
    for token in (cy.token, di.token):
        seat = room.seat_for_token(token)
        assert seat is not None and seat.host is False and seat.connected


def test_a_spectator_of_a_full_room_stays_watching_on_to_lobby() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    eve = new_session("Eve", clock)
    room = supervisor.create_room(eve, "Eve")
    tokens = [eve.token]
    for name in ("Fay", "Gus", "Hal"):
        token = new_session(name, clock).token
        supervisor.join(room.code, token, name)
        tokens.append(token)
    room.set_ready(tokens[1], True)
    room.start(eve.token)
    now = drive_countdown(room)
    imran = new_session("Imran", clock)
    supervisor.join(room.code, imran.token, "Imran")
    now = play_round_to_end(room, now)

    room.to_lobby()
    assert len(room.seats) == 4  # the room is full: nobody new can sit
    assert len(room.spectators) == 1
    assert room.spectator_for_token(imran.token) is not None


# ------------------------------------------------------------------ colours


def test_colors_are_first_free_in_seat_order() -> None:
    room, _ = make_room(two_level_bundle())
    ada = room.add_seat("tok-a", "Ada", True)
    bob = room.add_seat("tok-b", "Bob", False)
    cy = room.add_seat("tok-c", "Cy", False)
    dee = room.add_seat("tok-d", "Dee", False)
    assert (ada.color, bob.color, cy.color, dee.color) == SEAT_PALETTE

    # Seat 2 leaves: the next joiner takes the first free colour (c2).
    room.remove_seat("tok-c")
    eve = room.add_seat("tok-e", "Eve", False)
    assert eve.seat_index == 2 and eve.color == SEAT_PALETTE[2]

    # Seat 0 leaves: the next joiner takes the first free colour (c0).
    room.remove_seat("tok-a")
    fred = room.add_seat("tok-f", "Fred", False)
    assert fred.seat_index == 0 and fred.color == SEAT_PALETTE[0]


def test_a_returning_token_keeps_its_color_over_the_first_free() -> None:
    room, _ = make_room(two_level_bundle())
    room.add_seat("tok-a", "Ada", True)  # palette[0]
    room.add_seat("tok-b", "Bob", False)  # palette[1]
    room.add_seat("tok-c", "Cy", False)  # palette[2]
    room.remove_seat("tok-b")
    room.add_seat("tok-d", "Dee", False)  # seat 1, first free = palette[1]
    room.remove_seat("tok-a")  # palette[0] is now the first free
    bob2 = room.add_seat("tok-b", "Bob", False)  # seat 0: first free = palette[0]
    assert bob2.seat_index == 0
    assert bob2.color == SEAT_PALETTE[1]  # Bob's remembered colour wins


# ------------------------------------------------------------------ emotes


def test_emote_whitelist_and_throttle() -> None:
    room, _ = make_room(two_level_bundle())
    room.add_seat("tok-a", "Ada", True)

    assert room.emote("tok-a", "wave") is True  # t=0
    assert room.emote("tok-a", "cheer", now=0.1) is True
    assert room.emote("tok-a", "star", now=0.2) is False  # the 3rd in 1 s: dropped
    assert room.emote("tok-a", "star", now=1.1) is True  # the window has passed

    events = room.take_events()
    assert [(e.kind, dict(e.payload)) for e in events] == [
        (EMOTE_EVENT_KIND, {"emote": "wave", "player_id": "tok-a"}),
        (EMOTE_EVENT_KIND, {"emote": "cheer", "player_id": "tok-a"}),
        (EMOTE_EVENT_KIND, {"emote": "star", "player_id": "tok-a"}),
    ]
    # The per-token windows are independent.
    room.add_seat("tok-b", "Bob", False)
    assert room.emote("tok-b", "wave", now=0.2) is True  # unthrottled for Bob

    with pytest.raises(UnknownEmote):
        room.emote("tok-a", "dance")  # outside the whitelist: a named refusal
    assert room.emote("tok-stranger", "wave") is False  # an unknown token: dropped
    assert set(EMOTES) == {"wave", "cheer", "thumbs_up", "thumbs_down", "angry", "star"}


def test_emotes_are_dropped_outside_the_emote_states_and_for_spectators_too() -> None:
    room, _ = make_room(two_level_bundle())
    room.add_seat("tok-a", "Ada", True)
    room.add_spectator("tok-b", "Bob")
    assert room.emote("tok-b", "wave") is True  # a seated-or-watching token may

    room.start("tok-a")  # COUNTDOWN
    assert room.emote("tok-a", "wave") is False  # dropped, not an error
    events = room.take_events()
    # Only the accepted LOBBY emote is buffered; the dropped one is not.
    assert len(events) == 1
    assert events[0].kind is EMOTE_EVENT_KIND
    assert events[0].payload == {"emote": "wave", "player_id": "tok-b"}


# ------------------------------------------------------------- the payload


def test_the_lobby_payload_shape_and_viewer_unlock() -> None:
    sessions = {"tok-a": SessionRecord("tok-a", "Ada", 0.0, 2)}
    room, _ = make_room(two_level_bundle(), sessions)
    room.add_seat("tok-a", "Ada", True)
    room.add_seat("tok-b", "Bob", False)
    room.set_ready("tok-b", True)
    room.add_spectator("tok-c", "Cy")

    payload: dict[str, Any] = room.lobby_payload("tok-a")
    assert set(payload) == {
        "code",
        "state",
        "level",
        "host_seat",
        "can_start",
        "seats",
        "spectators",
        "unlocked",
        "round_no",
    }
    assert payload["code"] == room.code
    assert payload["state"] == "lobby"
    assert payload["level"] == "05a_first"
    assert payload["host_seat"] == 0
    assert payload["can_start"] is True
    assert payload["round_no"] == 0
    assert payload["unlocked"] == 2  # the viewer's session
    assert set(payload["seats"][0]) == {
        "index",
        "name",
        "color",
        "ready",
        "host",
        "connected",
    }
    assert [seat["host"] for seat in payload["seats"]] == [True, False]
    assert [seat["name"] for seat in payload["seats"]] == ["Ada", "Bob"]
    assert payload["seats"][0]["color"] == SEAT_PALETTE[0]
    assert payload["spectators"] == [{"name": "Cy", "color": SEAT_PALETTE[2]}]

    # An anonymous viewer sees no unlock, and a token never leaks out.
    assert room.lobby_payload()["unlocked"] is None
    assert room.lobby_payload("tok-stranger")["unlocked"] is None
    blob = str(room.lobby_payload("tok-a"))
    assert "tok-a" not in blob and "tok-c" not in blob


# -------------------------------------------------- names at the boundary


def test_a_markup_display_name_comes_back_sanitized() -> None:
    """US-02 criterion 6, at the boundary: the markup name is stored
    sanitized (≤ 16 chars, markup-free) for a create and a join."""
    supervisor, clock = make_supervisor(two_level_bundle())
    room = supervisor.create_room(new_session(None, clock), "<script>alert(1)</script>")
    name = room.seats[0].display_name
    assert name == "scriptalert(1)/s"
    assert len(name) <= 16
    assert not {"<", ">", '"'} & set(name)

    cy = new_session(None, clock)
    supervisor.join(room.code, cy.token, "<script>alert(1)</script>")
    assert seat(room, cy.token).display_name == "scriptalert(1)/s"

    # A whitespace-only name falls back to the seat number.
    dee = new_session(None, clock)
    supervisor.join(room.code, dee.token, "   ")
    assert seat(room, dee.token).display_name == "Cook #3"


# -------------------------------------------------- seats' own methods


def test_the_seat_and_spectator_verbs() -> None:
    """The `Seat`/`Spectator` verbs and the room's bookkeeping edges:
    `is_host`, `next_seat_index` on a full room, the not-seated / not-
    watching refusals, the `add_spectator` reclaim, the session lookup's
    `None` branch."""
    room, _ = make_room(two_level_bundle())
    host = room.add_seat("tok-a", "Ada", True)
    room.add_seat("tok-b", "Bob", False)
    assert host.is_host is True
    assert seat(room, "tok-b").is_host is False

    assert room.mark_seat_connected("tok-stranger") is False  # not seated
    assert room.mark_spectator_connected("tok-stranger") is False  # not watching

    # The reclaim path: a second `add_spectator` for the same token keeps
    # the entry, re-connects it, and updates the name when a new one is
    # typed (an empty name keeps the old one).
    first = room.add_spectator("tok-b", "Bob")
    assert first.color == SEAT_PALETTE[2]  # first free, after the two seats
    again = room.add_spectator("tok-b", "Bobby")
    assert again is first and again.connected is True
    assert again.display_name == "Bobby"
    room.add_spectator("tok-b", "")
    assert again.display_name == "Bobby"  # an empty name keeps the old one

    # A full room: `next_seat_index` names the refusal, `add_seat` refuses.
    room.add_seat("tok-c", "Cy", False)
    room.add_seat("tok-d", "Dee", False)
    assert room.is_full
    with pytest.raises(IndexError, match="no free seat"):
        room.next_seat_index()
    with pytest.raises(ValueError, match="is full"):
        room.add_seat("tok-e", "Eve", False)

    # A test-built room (no `sessions` argument): the lookup is `None`, and
    # `_session_for` says so (a token counts as unlocked 1).
    bare2 = Room(code="T2", settings=Settings(), clock=FakeClock())
    assert bare2.sessions is None
    assert bare2._session_for("tok-anyone") is None


# -------------------------------------------------- sessions at the boundary


def test_a_join_with_an_unknown_token_mints_a_session() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    supervisor.create_room(new_session("Ada", clock), "Ada")
    room = supervisor.all_rooms()[0]

    assert supervisor.session_for("tok-unknown") is None
    supervisor.join(room.code, "tok-unknown", "Zed")
    record = supervisor.session_for("tok-unknown")
    assert record is not None
    assert record.display_name == "Zed"
    assert record.unlocked_level == 1  # a fresh session: the campaign's floor


def test_a_known_session_record_is_left_alone_on_rejoin() -> None:
    supervisor, clock = make_supervisor(two_level_bundle())
    supervisor.create_room(new_session("Ada", clock), "Ada")
    room = supervisor.all_rooms()[0]
    unlocked = SessionRecord("tok-u2", "U2", 0.0, 3)
    supervisor.create_room(unlocked, None)  # U2 owns a room of their own

    supervisor.join(room.code, "tok-u2", None)  # evicted from their room
    assert supervisor.session_for("tok-u2") is unlocked  # the record is intact
    assert unlocked.unlocked_level == 3