pinch-points 0.9.0

//! The round after this one: who is still owed a seat, who has been
//! waiting in line for one, and the terms everybody agrees to play the
//! next board on.
//!
//! All of it is the host's business. A joiner keeps no launch plan and
//! answers nobody, since the star's spokes cannot hear each other, so
//! every decision here is made once, at the hub, and sent out.

use super::*;

/// What a host's `Start` invites this peer to: the size of the table, the
/// chair (or none, for a watcher), the terms, who else is at it, where
/// the series stands, and the host's own beach when it sent one. The
/// message's own payload, under the name the session takes it up by.
pub use crate::transport::Start as Invitation;

/// The lockstep a peer plays a round over: the `players`' seats (every
/// seat but the AI's, [`MatchTerms::player_seats`]), from the given chair,
/// or from no chair at all for a watcher, with the round's input `delay`.
pub(crate) fn lockstep_for(seat: Option<u8>, players: Vec<u8>, delay: u32) -> Lockstep {
    match seat {
        None => Lockstep::observer(players, delay),
        Some(seat) => Lockstep::new(seat, players, delay),
    }
}

/// The input delay a host deals a round with: enough for the slowest link
/// among the peers `plan` seats, each played as its round trips say
/// (`RoundTrips::telling`, [`crate::sim::delay_for_round_trip`]),
/// or the default for a table nobody has answered at (the direct pair, a
/// table of the host and the AI). A watcher is waited on by nobody, so its
/// link does not count. Nor does a player that has never answered a ping:
/// it is played at what the rest of the table needs, the least the host
/// can say of a link it knows nothing about.
pub(crate) fn input_delay(peers: &PeerBook, plan: &[Option<u8>]) -> u32 {
    plan.iter()
        .enumerate()
        .filter(|(_, seat)| seat.is_some())
        .filter_map(|(peer, _)| peers.get(peer)?.round_trips.telling())
        .reduce(f32::max)
        .map_or(crate::sim::DEFAULT_DELAY, crate::sim::delay_for_round_trip)
}

/// The AI a round between rounds is dealt behind `humans`, the terms
/// asking for `dial`: as many as there are chairs for, and enough for two
/// castles, since a host whose table has emptied is still entitled to a
/// round against the AI. The same floor `seat_count` keeps.
fn next_round_ai(humans: u8, dial: u8) -> u8 {
    dial.min(MAX_PLAYERS as u8 - humans)
        .max(2u8.saturating_sub(humans))
}

/// How many of the host's own bots (route 2) a table with `people` seated
/// has chairs for: the people sit first, and a bot gives way to one.
pub(crate) fn chairs_for_bots(people: u8) -> usize {
    MAX_PLAYERS.saturating_sub(usize::from(people))
}

/// The host's own bots (route 2) seated after the `people`, as many as
/// there are chairs for, in the order they came: at the launch and again
/// before every round, since a person leaving moves them up a chair.
pub(crate) fn seat_the_hosts_bots(bots: &mut Vec<HostedBot>, people: u8) {
    bots.truncate(chairs_for_bots(people));
    for (i, bot) in bots.iter_mut().enumerate() {
        bot.seat = people + i as u8;
    }
}

/// What every seat is before the deal says more: the AI in the seats the
/// terms give it at a table of `seats` ([`MatchTerms::ai_seats`]), a
/// person in the rest.
fn ai_where_the_terms_say(terms: MatchTerms, seats: u8) -> [crate::sim::SeatKind; MAX_PLAYERS] {
    let ai = terms.ai_seats(seats);
    std::array::from_fn(|seat| match ai.contains(&(seat as u8)) {
        true => crate::sim::SeatKind::Ai,
        false => crate::sim::SeatKind::Human,
    })
}

/// Put a dealt table in the seats its host arranged when it picked
/// partners ([`super::order`]): `plan` and `bots_here` come dealt in places
/// (the host, the players in peer order, the host's bots), and leave in
/// the seats each was arranged in, the AI in what is left, the order in
/// `terms`; and the seats are remembered for the next deal. A table never
/// arranged, or played free-for-all ([`order::picks_hold`]), is dealt in
/// seat order, as every table was before.
pub(crate) fn arrange_the_deal(
    (plan, bots_here): (&mut [Option<u8>], &mut [HostedBot]),
    peers: &mut PeerBook,
    arranged: Option<&mut order::Arranged>,
    terms: &mut MatchTerms,
    seats: u8,
) {
    use crate::app::match_setup::TermsDials;
    let held = order::picks_hold(terms.teams().played_by(seats));
    let Some(arranged) = arranged.filter(|_| held) else {
        terms.order = crate::transport::IN_SEAT_ORDER;
        return;
    };
    let humans = terms.humans(seats);
    let mut wants = vec![None; usize::from(seats)];
    for (peer, place) in plan.iter().enumerate() {
        if let Some(place) = place
            && let Some(want) = wants.get_mut(usize::from(*place))
        {
            *want = peers.get(peer).and_then(|row| row.seat_wish);
        }
    }
    for bot in bots_here.iter() {
        if let Some(want) = wants.get_mut(usize::from(bot.seat)) {
            *want = arranged
                .bots
                .iter()
                .find(|(id, _)| *id == bot.id)
                .map(|&(_, seat)| seat);
        }
    }
    terms.order = order::seat_order(&wants, &arranged.ai, humans, seats);
    info!(
        "dealt in the seats the host's partner picks put them in: {:?}",
        &terms.order[..usize::from(seats)]
    );
    for (peer, place) in plan.iter_mut().enumerate() {
        if let Some(place) = place {
            *place = terms.seat_of_place(*place);
            peers.row(peer).seat_wish = Some(*place);
        }
    }
    for bot in bots_here.iter_mut() {
        bot.seat = terms.seat_of_place(bot.seat);
    }
    *arranged = order::remember(*terms, seats, bots_here);
}

/// The table's names for a dealt `plan`: the host in seat 0 under `host`,
/// each seated peer under what `peer_name` says it is called, and the
/// host's own bots under theirs. Empty slots (AI seats, the nameless) fall
/// back to seat labels on every screen.
pub(crate) fn seat_names<'a>(
    host: String,
    plan: &[Option<u8>],
    peer_name: impl Fn(usize) -> Option<&'a str>,
    bots_here: &[HostedBot],
) -> [String; MAX_PLAYERS] {
    let mut names: [String; MAX_PLAYERS] = Default::default();
    names[0] = host;
    for (peer, slot) in plan.iter().enumerate() {
        if let Some(seat) = slot
            && let Some(name) = peer_name(peer)
        {
            names[usize::from(*seat)] = name.to_string();
        }
    }
    for bot in bots_here {
        names[usize::from(bot.seat)].clone_from(&bot.name);
    }
    names
}

/// What holds each seat of a dealt `plan`: the host a person, a peer
/// whatever its greeting said, the host's own bots bots, and the AI the
/// seats `terms` give it at a table of `seats`.
pub(crate) fn seat_kinds(
    plan: &[Option<u8>],
    peers: &PeerBook,
    bots_here: &[HostedBot],
    terms: MatchTerms,
    seats: u8,
) -> [crate::sim::SeatKind; MAX_PLAYERS] {
    let mut kinds = ai_where_the_terms_say(terms, seats);
    for bot in bots_here {
        kinds[usize::from(bot.seat)] = crate::sim::SeatKind::Bot;
    }
    for (peer, slot) in plan.iter().enumerate() {
        if let Some(seat) = slot
            && peers.get(peer).is_some_and(|row| row.bot)
        {
            kinds[usize::from(*seat)] = crate::sim::SeatKind::Bot;
        }
    }
    kinds
}

impl OnlineSession {
    /// A session formed on a host's invitation, as a joiner's is from the
    /// lobby: the socket it greeted over, and everything the `Start` said.
    pub fn invited(transport: UdpTransport, invitation: Invitation) -> OnlineSession {
        let Invitation {
            seats,
            seat,
            terms,
            delay,
            ..
        } = &invitation;
        let mut session = OnlineSession::new(
            transport,
            lockstep_for(*seat, terms.player_seats(*seats), u32::from(*delay)),
            *seats,
            *terms,
        );
        session.take_up(invitation);
        session
    }

    /// Take up an invitation: the host's beach, the table's names, and a
    /// round begun afresh on its terms (see [`Self::begin_round`] for
    /// everything that resets). Does not arm `next_round`, which is the
    /// caller's to say: a session formed on its first invitation is
    /// walking into the arena, not back into it.
    pub fn take_up(&mut self, invitation: Invitation) {
        let Invitation {
            seats,
            seat,
            terms,
            names,
            kinds,
            standing,
            beach,
            deal,
            crowned,
            delay,
        } = invitation;
        self.beach = beach;
        self.begin_round(
            seats,
            seat,
            terms,
            table_from_wire(&names),
            u32::from(delay),
        );
        self.kinds = kinds.map(crate::transport::kind_from_byte);
        self.series_standing = standing;
        self.rounds_dealt = deal;
        self.crowned = crowned;
    }

    /// Whether a `Start` is the host calling the next round: heard by a
    /// joiner, and dealt after the round this one is in, which is what
    /// says so. Everything else about the round may well be the same, and
    /// a stale copy of this one or the last is the host repeating itself
    /// to somebody still greeting.
    pub(super) fn calls_the_next_round(&self, start: &Invitation) -> bool {
        !self.is_host() && self.is_next_round(start)
    }

    /// Take up the host's call for the next round, and arm the session to
    /// walk back into the arena on it. Only the session is rearmed here;
    /// the board is rebuilt on the way in, by the same path a first round
    /// uses.
    pub(super) fn take_the_next_round(&mut self, start: Invitation) {
        self.take_up(start);
        self.next_round = true;
    }

    /// The answer for a peer the socket picked up *after* the launch, or
    /// `None` for one that was at the table when the round began.
    ///
    /// Membership is the launch plan, not the seat: a peer seated as a
    /// spectator in the lobby is `Watching` in the plan, while a stranger
    /// who greeted mid-round is `Queued`. Both answer `None` to `seat_of`,
    /// so the plan is the only thing that tells them apart, and admitting
    /// the second as a spectator leaves it staring at frame zero forever.
    pub(super) fn queue_place(&self, peer: usize) -> Option<NetMsg> {
        debug_assert!(
            self.is_host() || self.peers.planned() == 0,
            "a joiner keeps no launch plan and must never answer with one"
        );
        let seated = self.peers.planned();
        // Peers are registered in arrival order, so everyone between the
        // plan and this one is already waiting: for a chair, that is. A
        // watcher, caught up or in line to watch, takes none, and a player
        // told a spectator was ahead of them was not next when they were.
        //
        // Ahead of all of them, a player the launch left without a chair:
        // the next deal seats it first, in peer order.
        peer.checked_sub(seated)?;
        let chairless = (0..seated)
            .filter(|&before| self.peers.get(before).is_some_and(Peer::wants_a_chair))
            .count();
        let ahead = chairless
            + (seated..peer)
                .filter(|&before| {
                    self.peers
                        .get(before)
                        .is_none_or(|row| row.place == Place::Queued && !row.watch)
                })
                .count();
        Some(NetMsg::Queued {
            ahead: ahead.min(u8::MAX as usize) as u8,
        })
    }

    /// Drain the socket between rounds, when the sim is stopped and `pump`
    /// is therefore not running either.
    ///
    /// The results card is when people turn up wanting in, and when the
    /// host decides there will be another round, so the two messages that
    /// matter here are a greeting to queue and the invitation that answers
    /// it. Hashes belong to the round that just ended, and are dropped with
    /// it; a peer's batch says it is still playing that round, and is
    /// answered with the frames it says it is missing (`send_the_tail`).
    ///
    /// A joiner also keeps greeting, which is not for the host's benefit:
    /// the answer is what proves the host is still there. Nothing else is
    /// sent between rounds, so without it [`Self::host_gone`] would call
    /// the round off under a table that is simply reading the scores.
    pub fn poll_between_rounds(&mut self, delta: f32) {
        let host = self.is_host();
        self.age_the_silence(delta);
        if !host && crate::app::lobby::once_a_second(&mut self.home.greet_in, delta) {
            self.transport.send(self.own_greeting());
        }
        // The card is where the next round's delay is picked, from how
        // quickly the table answers now. And whether the host is picking
        // the next map goes with it, as it stands, picking or not: one
        // datagram is the one the network eats, and a lost word that the
        // host stopped picking left the table told it still was until the
        // next round.
        if host && crate::app::lobby::once_a_second(&mut self.ping_in, delta) {
            self.ping();
            self.transport.send(self.roster_msg());
        }
        for (msg, from) in self.transport.recv_all() {
            // Except a goodbye, which says the opposite, and whose grace
            // runs from the first copy rather than the last.
            if msg != NetMsg::Leaving {
                self.mark_heard(from);
            }
            match msg {
                NetMsg::Hello { name, bot } => {
                    if host && !self.refuse_a_bot(from, bot) {
                        self.peers.row(from).bot = bot;
                        let answer = self.answer_greeting(from, &name_from_wire(&name), false);
                        self.transport.send_to(from, answer);
                        self.send_the_tail(from);
                    }
                }
                NetMsg::Watch { name } => {
                    if host {
                        let answer =
                            self.answer_greeting(from, &name_from_wire(&name), true);
                        self.transport.send_to(from, answer);
                        self.send_the_tail(from);
                    }
                }
                // A peer still in the round, resending what it committed
                // or saying what it holds: it has not finished, and may be
                // missing the last frames.
                NetMsg::Inputs(batch) if host => self.answer_a_straggler(from, &batch),
                NetMsg::Start(start) => {
                    if self.calls_the_next_round(&start) {
                        self.take_the_next_round(start);
                    }
                }
                // The next round as it stands, for a player dealt a
                // watching place whose `Start` went astray: walked into
                // part way through, as a watcher, rather than not at all.
                NetMsg::CatchUp {
                    frame,
                    part,
                    parts,
                    bytes,
                } => {
                    self.take_catch_up_part(frame, part, parts, bytes);
                }
                // The results card is where a table talks most, and the
                // spectators' prompt offers it there: a line dropped here
                // was a key that did nothing for anyone but its sender.
                NetMsg::Chat { name, text } => self.hear_chat(host, from, name, text),
                // The host's word that it is picking the next map, which
                // the results card says while the table waits on it.
                NetMsg::Roster { picking, .. } => {
                    if !host {
                        self.picking_map = picking;
                    }
                }
                NetMsg::Inputs(_)
                | NetMsg::Hash { .. }
                | NetMsg::Pause { .. }
                | NetMsg::Resume { .. }
                | NetMsg::Queued { .. }
                | NetMsg::Abandoned { .. }
                // There is no beach here to call anything onto, which is
                // why the key is not offered between rounds either.
                | NetMsg::SpectatorVote { .. }
                | NetMsg::SpectatorTally { .. }
                // Calls are for a round being played, and this one is over.
                | NetMsg::SpectatorPick { .. }
                | NetMsg::CrowdPicks { .. }
                // A clock is kept while a round is played.
                | NetMsg::Clock { .. }
                | NetMsg::Incompatible { .. } => {}
                // The host asked us to leave while the card was up.
                NetMsg::Kicked if !host => PutOut::hear(&mut self.put_out, PutOut::Kicked),
                NetMsg::NoBots if !host => PutOut::hear(&mut self.put_out, PutOut::NoBots),
                NetMsg::Full if !host => PutOut::hear(&mut self.put_out, PutOut::Full),
                NetMsg::Kicked | NetMsg::NoBots | NetMsg::Full => {}
                // Gone from the card: forgotten by the sweep just below,
                // before the next round can deal it a chair. On a joiner
                // it is the host, and there is no next round, unless the
                // goodbye was forged (`heard_a_goodbye`).
                NetMsg::Leaving => self.heard_a_goodbye(from),
                NetMsg::Ping { stamp } => self.answer_ping(host, stamp),
                NetMsg::Pong { stamp } => self.take_pong(host, from, stamp),
            }
        }
        // After the socket is drained, so a greeting that arrived in this
        // very poll counts as a word before the silence is judged.
        if host {
            self.forget_the_silent();
        }
    }

    /// Host, on the results card: a batch from a peer still in the round
    /// just played. What it says it holds is noted, as during the round,
    /// and it is answered with what it is missing of the round's end.
    fn answer_a_straggler(&mut self, from: usize, batch: &Batch) {
        if !self.of_this_round(batch.round) {
            return;
        }
        if let Some(wants) = self.wants_of_this_round(batch.wants) {
            self.peers.row(from).wants = Some(wants);
        }
        self.send_the_tail(from);
    }

    /// What a peer says it holds (`Batch::wants`), if it names nothing past
    /// a frame this round can have reached.
    pub(super) fn wants_of_this_round(
        &self,
        wants: Option<[u32; MAX_PLAYERS]>,
    ) -> Option<[u32; MAX_PLAYERS]> {
        wants.filter(|wants| {
            wants
                .iter()
                .all(|&frame| self.session.within_horizon(frame))
        })
    }

    /// Host: the end of the round just played, to a peer that followed it
    /// and may not have finished it: every input it has not said it holds,
    /// up to the last frame (`unheard_by`).
    ///
    /// The host relays an input once, as it arrives, and its sim stops
    /// the moment the round is over, so the relay of the input that
    /// completed the last frame is never repeated: lost, it left a joiner
    /// a frame short of the end for good, "waiting for" a player who had
    /// long stopped sending. A player still in the round keeps resending
    /// its own commits, a watcher keeps saying what it holds, and either
    /// greets a host gone quiet, so each is answered from the frame it
    /// says it is missing, as far back as the record reaches: a spectator
    /// seconds behind when the tide came in is still fed the round to its
    /// end, where a fixed tail of the last second left it short of it for
    /// good. A peer that has finished hears frames it has played, and
    /// drops them.
    pub(super) fn send_the_tail(&self, peer: usize) {
        if !self.follows_the_round(peer) {
            return;
        }
        let tail = Batch {
            round: self.round_tag(),
            inputs: self.unheard_by(peer, self.session.frame()),
            ..Batch::default()
        };
        self.transport.send_batch(&tail, |to| to == peer);
    }

    /// What this peer says to its host to be let in, or to be answered:
    /// the greeting the lobby sends, wish to watch included, so a watcher
    /// does not greet as a nameless player, and a bot's seat says a bot
    /// drives it. One greeting from the results card and from a round
    /// whose host has gone quiet (`greet_a_quiet_host`).
    pub(super) fn own_greeting(&self) -> NetMsg {
        let me = self
            .session
            .seat()
            .map_or("", |seat| &self.names[usize::from(seat)]);
        crate::app::lobby::greeting(self.bot.as_deref(), self.watch_wish(), me)
    }

    /// A goodbye off the wire. On a joiner it is the host's, and a claim
    /// nothing checks: believed only once it has stood for a moment's
    /// grace with no word from the host since (`has_gone`). Between rounds
    /// the host says nothing unless spoken to, so the joiner greets at once
    /// rather than at its next once-a-second greeting: a host still there
    /// answers within a round trip, inside the grace, and that answer takes
    /// a forged goodbye back.
    pub(super) fn heard_a_goodbye(&mut self, from: usize) {
        self.peers.said_goodbye(from);
        if !self.is_host() {
            self.transport.send(self.own_greeting());
        }
    }

    /// Host: what a greeting earns its sender, with what it said kept.
    ///
    /// One answer whether the round is running (`pump`) or the table is on
    /// the results card (`poll_between_rounds`): a peer in line is told
    /// so, and one in the plan is told its place again, in case the Start
    /// went astray. The name and the wish to watch are written down either
    /// way, because they are what the next round's plan is dealt from.
    pub(super) fn answer_greeting(&mut self, from: usize, told: &str, watch: bool) -> NetMsg {
        debug_assert!(self.is_host(), "only the host answers greetings");
        if !told.is_empty() {
            self.remember_peer_name(from, told);
        }
        if watch {
            self.note_watch_wish(from);
        }
        if let Some(queued) = self.queue_place(from) {
            // And shown the table it is waiting for. The roster is the
            // lobby's own message and the round sends nobody one, so a
            // peer that turned up mid-round sat in front of an empty
            // beach panel, told it was next up for a match it could see
            // nothing of, with a terms card falling back to its own
            // machine's settings as though they were the host's.
            self.transport.send_to(from, self.roster_msg());
            return queued;
        }
        // A watcher is told it is watching, so the seat it gets is not one.
        let seat = if watch { None } else { self.seat_of(from) };
        if let Some(slot) = seat.and_then(|seat| self.names.get_mut(usize::from(seat)))
            && !told.is_empty()
        {
            *slot = told.to_string();
        }
        self.start_msg(seat)
    }

    /// The table as it stands, for a peer that has none: who holds which
    /// seat, and the terms they are playing. The same message the lobby
    /// sends while it is gathering, from the round's own copy of both.
    fn roster_msg(&self) -> NetMsg {
        // The beach being played, by name, when it is the host's own: the
        // bytes carry it, and they are what every board is built from, by
        // the same rule (`beach_in_play`), so a beach the table outgrew is
        // not named while the generated one is played.
        use crate::app::match_setup::{TermsDials, beach_from, beach_in_play};
        let beach = beach_in_play(self.terms.map(), beach_from(&self.beach), self.seats)
            .map_or_else(String::new, |level| level.name);
        // The places at the table, as the lobby counts them (`send_roster`):
        // the people and bots in place order, with the AI and the seat each
        // place sits in in the terms.
        let humans = self.terms.humans(self.seats);
        let by_place = |place: u8| usize::from(self.terms.seat_of_place(place));
        let names: Vec<String> = (0..humans)
            .map(|place| self.names[by_place(place)].clone())
            .collect();
        let mut kinds = [crate::sim::SeatKind::Human; MAX_PLAYERS];
        for place in 0..humans {
            kinds[usize::from(place)] = self.kinds[by_place(place)];
        }
        NetMsg::Roster {
            seats: humans,
            names: wire_table(&names),
            terms: self.terms,
            kinds: crate::transport::wire_kinds(&kinds),
            beach: crate::transport::wire_name(&beach),
            // Its place is the `Queued` it is sent with.
            chair: None,
            picking: self.picking_map,
        }
    }

    /// Host: write down a peer's name against its socket index, growing the
    /// row to reach it. Kept for peers the seat table does not cover yet.
    pub(super) fn remember_peer_name(&mut self, peer: usize, name: &str) {
        self.peers.row(peer).name = name.to_string();
    }

    /// Host: note that a peer asked to watch rather than play.
    pub(super) fn note_watch_wish(&mut self, peer: usize) {
        self.peers.row(peer).watch = true;
    }

    /// Host: a greeting that says a bot drives the peer, at a beach whose
    /// host said bots are not welcome. The peer is asked to go and is
    /// never dealt a chair; it stops greeting, and is forgotten with the
    /// silent. A peer already seated keeps its seat: the dial does not
    /// turn mid-match, so it was welcome when it sat down.
    pub(super) fn refuse_a_bot(&mut self, from: usize, bot: bool) -> bool {
        if !bot || self.home.bots_welcome || self.peers.seat_of(from).is_some() {
            return false;
        }
        self.peers.row(from).watch = true;
        self.transport.send_to(from, NetMsg::NoBots);
        true
    }

    /// The name a peer index goes by, from its greeting: its seat's name if
    /// it holds one, else what it greeted with while queued.
    pub(super) fn peer_name(&self, peer: usize) -> Option<&str> {
        if let Some(seat) = self.peers.seat_of(peer)
            && let Some(name) = self.names.get(usize::from(seat))
            && !name.is_empty()
        {
            return Some(name);
        }
        self.peers
            .get(peer)
            .map(|peer| peer.name.as_str())
            .filter(|name| !name.is_empty())
    }

    /// Host: how many seats the next round is dealt if it is called now
    /// with the AI dial at `dial`, as [`Self::call_next_round`] deals them:
    /// whoever queued admitted, whoever left gone, and the AI behind them.
    /// What the host's map page between rounds greys its cards by.
    pub fn next_round_seats(&self, dial: u8) -> u8 {
        let plan = self.next_round_plan();
        let people = 1 + plan.iter().flatten().count() as u8;
        let bots = self.bots_here.len().min(chairs_for_bots(people)) as u8;
        let humans = people + bots;
        humans + next_round_ai(humans, dial)
    }

    /// Host: who the next round seats, one entry per peer on the socket,
    /// `None` for one that watches or that the table cannot fit: whoever
    /// queued admitted and whoever left gone ([`PeerBook::seat_plan`]).
    /// While the host picks the map for it, the table it was picking for
    /// and nobody else ([`PeerBook::frozen_plan`]): one that turned up
    /// meanwhile watches the round and plays the one after, as at the
    /// lobby's start, so the map is played by the table it was picked for.
    fn next_round_plan(&self) -> Vec<Option<u8>> {
        let peers = self.transport.peer_count();
        match self.picking_map && self.is_host() {
            true => self.peers.frozen_plan(peers),
            false => self.peers.seat_plan(peers),
        }
    }

    /// Say whether the host is picking the next round's map: on the host,
    /// noted and told to every peer at once, as the roster that carries it
    /// (`NetMsg::Roster::picking`); a lost one, either way, is said again
    /// on the results card's once-a-second tick. Picking, the next round's table is
    /// frozen as it stands ([`Self::next_round_plan`]).
    pub fn say_picking(&mut self, picking: bool) {
        if picking && !self.picking_map {
            let plan = self.peers.seat_plan(self.transport.peer_count());
            self.peers.freeze(&plan);
        }
        self.picking_map = picking;
        self.transport.send(self.roster_msg());
    }

    /// Host: call the next round on `terms`, admitting the queue, and tell
    /// every peer which seat it now holds. Arms this session too, so host
    /// and joiners take the same path back into the arena.
    ///
    /// `standing` is the series as it is now, by *this* round's seats, or
    /// `None` outside a series; the return value is the same standing
    /// re-dealt to the seats the next round hands out, which the caller
    /// folds back into its `Tournament`. Seats are re-dealt every round, in
    /// peer order or in the seats the host's partner picks put everybody
    /// in ([`arrange_the_deal`]), so a peer that leaves can move others a
    /// chair and the wins have to move with them.
    ///
    /// `terms.bots` is the host's AI dial, not the AI of the round just
    /// played: the next round deals its AI from the table as it then sits
    /// ([`next_round_ai`]). Read off the round before, an AI seated only
    /// to give a host alone a second castle stayed on after a friend sat
    /// down in its place.
    pub fn call_next_round(
        &mut self,
        mut terms: MatchTerms,
        standing: Option<SeriesStanding>,
    ) -> Option<SeriesStanding> {
        // The table for the next round, admitting whoever queued while
        // this one played: one entry per peer on the socket, `None` for a
        // peer that watches or that the table could not fit.
        let mut plan = self.next_round_plan();
        let people = 1 + plan.iter().flatten().count() as u8;
        let mut bots_here = std::mem::take(&mut self.bots_here);
        let bots_were: Vec<u8> = bots_here.iter().map(|bot| bot.seat).collect();
        seat_the_hosts_bots(&mut bots_here, people);
        let humans = people + bots_here.len() as u8;
        terms.bots = next_round_ai(humans, terms.bots);
        let seats = humans + terms.bots;
        // Everybody back in the seat the host's partner picks put them in.
        arrange_the_deal(
            (&mut plan, &mut bots_here),
            &mut self.peers,
            self.arranged.as_mut(),
            &mut terms,
            seats,
        );
        // The map was stepped on for the table as it sat, and admitting the
        // queue can have grown it past what that beach holds: stepped on
        // again for the table as it now is, or two players at a table of
        // six would have no castle.
        //
        // A handmade beach is stepped off, too, once the table outgrows its
        // castles, and is not sent any more: the terms name the beach that
        // is played, and that is what every card reads.
        terms = crate::app::match_setup::fit_terms(terms, seats, &self.beach);
        if !crate::app::match_setup::is_custom(&terms) {
            self.beach = Vec::new();
        }
        let standing = standing.map(|SeriesStanding { round, wins }| {
            // The wins follow the chairs: seat 0 is always the host's, and
            // each peer's new seat inherits what its old seat had won, and
            // so does each of the host's bots.
            let mut new_wins = [0u8; MAX_PLAYERS];
            new_wins[0] = wins[0];
            for (peer, slot) in plan.iter().enumerate() {
                if let Some(new_seat) = slot
                    && let Some(old_seat) = self.peers.seat_of(peer)
                {
                    new_wins[usize::from(*new_seat)] = wins[usize::from(old_seat)];
                }
            }
            for (bot, old_seat) in bots_here.iter().zip(&bots_were) {
                new_wins[usize::from(bot.seat)] = wins[usize::from(*old_seat)];
            }
            // And the game's AI, in the seats the terms give it: the first
            // AI place of this round's table is the first of the next
            // one's. Left out, an AI that won a round was back at nothing
            // the next, and could never take an online series.
            for (old_seat, new_seat) in self
                .terms
                .ai_seats(self.seats)
                .into_iter()
                .zip(terms.ai_seats(seats))
            {
                new_wins[usize::from(new_seat)] = wins[usize::from(old_seat)];
            }
            // The number of the round about to begin: the caller passes
            // the one just played, and every peer shows the same next
            // number without each counting for itself.
            SeriesStanding {
                round: round.saturating_add(1),
                wins: new_wins,
            }
        });
        // Names travel with the invitation, as they did at the launch: a
        // player admitted from the queue is a stranger to every other
        // screen until this says otherwise.
        let names = seat_names(
            self.names[0].clone(),
            &plan,
            |peer| self.peer_name(peer),
            &bots_here,
        );
        let kinds = seat_kinds(&plan, &self.peers, &bots_here, terms, seats);
        let delay = input_delay(&self.peers, &plan);
        self.begin_round(seats, Some(0), terms, names, delay);
        self.deal(&plan, kinds, bots_here, standing);
        self.next_round = true;
        standing
    }

    /// Host: seat a dealt table and tell every peer in `plan` its place.
    ///
    /// The one way a host deals, at the launch and before every round
    /// after it: the caller has set the round's seats, terms, names and
    /// beach on this session, and this hands out the chairs, writes down
    /// what holds each seat, speaks for the host's own bots, and sends each
    /// peer the `Start` it plays the round from. Every `Start` is written
    /// off the session as it now stands, so the host plays on the very
    /// table it sent.
    pub(crate) fn deal(
        &mut self,
        plan: &[Option<u8>],
        kinds: [crate::sim::SeatKind; MAX_PLAYERS],
        bots_here: Vec<HostedBot>,
        standing: Option<SeriesStanding>,
    ) {
        self.peers.deal(plan);
        self.kinds = kinds;
        let spoken: Vec<u8> = bots_here.iter().map(|bot| bot.seat).collect();
        self.session.speak_for(&spoken);
        self.bots_here = bots_here;
        // Kept for the round as well as sent: the invitation the host
        // re-answers a stray greeting with, and the round it hands a late
        // watcher, carry it too.
        self.series_standing = standing;
        // A round dealt after every one before it, which is what tells it
        // from a stale invitation still on the wire (`is_next_round`).
        self.rounds_dealt = self.rounds_dealt.wrapping_add(1);
        for (peer, slot) in plan.iter().enumerate() {
            self.transport.send_to(peer, self.start_msg(*slot));
        }
    }

    /// The launch plan alone, no `Start` sent: for a table whose every
    /// peer was handed the session the plan gives it some other way, as the
    /// soak table builds them (`examples/soak/table.rs`). The game deals
    /// with [`Self::deal`].
    pub fn seat_the_table(&mut self, plan: &[Option<u8>]) {
        self.peers.deal(plan);
    }

    /// Take up the terms of a new round: a fresh board, a lockstep back at
    /// frame zero, and whatever the table is called now.
    #[cfg_attr(debug_assertions, track_caller)]
    pub fn begin_round(
        &mut self,
        seats: u8,
        seat: Option<u8>,
        terms: MatchTerms,
        names: [String; MAX_PLAYERS],
        delay: u32,
    ) {
        debug_assert!(
            (2..=MAX_PLAYERS as u8).contains(&seats),
            "a round for {seats} seats, which no per-seat array can hold"
        );
        debug_assert!(
            seat.is_none_or(|seat| seat < seats),
            "seated at {seat:?} of {seats}"
        );
        self.session = lockstep_for(seat, terms.player_seats(seats), delay);
        self.seats = seats;
        self.terms = terms;
        self.names = names;
        // The AI where the terms seat it; anything else a person until the
        // round's kinds say otherwise, which its caller writes after this.
        self.kinds = ai_where_the_terms_say(terms, seats);
        self.hashes.reset();
        self.resume_echo = 0;
        self.round_ticks = 0;
        self.host_newest = None;
        self.host_wants = None;
        self.clock_apart = None;
        // Seats given up on were last round's seats, at last round's
        // frames. The seats are dealt afresh, so a stale entry would read
        // a re-seated player as dropped and hand its chair to the AI.
        self.abandoned.clear();
        // A new round is everyone's from its first frame.
        self.catch_up = Default::default();
        self.stands.winner.new_round();
        // A pick and a pause are both this round's. A pause frame kept from
        // the last round matched one called at the same frame in this, and
        // the peer that had called the old one, leaving, lifted somebody
        // else's (`release_a_departed_pause`).
        for peer in self.peers.iter_mut() {
            peer.pick = None;
            peer.pause = None;
            peer.wants = None;
        }
        // And nobody is late for a round that has not begun. The results
        // card is a place a table sits for a while, and carrying that
        // silence into the new round would call the host gone on its first
        // frame.
        self.stall.reset(self.session.frame());
        self.peers.hush();
        // The map for it is picked.
        self.picking_map = false;
    }

    /// Whether a `Start` off the wire is the next round rather than a
    /// stale one a host re-answered a greeting with: dealt after the round
    /// this session is in.
    ///
    /// Numbered rather than told apart by its seed: the host answers
    /// greetings on its results card with the round just played, and an
    /// answer overtaken on the wire by the next round's call differed
    /// from it in seed all the same, so a joiner took it up and re-armed
    /// onto a beach everyone else had left.
    pub fn is_next_round(&self, start: &Invitation) -> bool {
        start.deal > self.rounds_dealt
    }

    /// The `Start` the host answers a stray greeting with: the terms plus
    /// the current name table, so a joiner that missed the launch still
    /// learns what everyone is called.
    pub(super) fn start_msg(&self, seat: Option<u8>) -> NetMsg {
        // The stale `Start` a greeting is re-answered with carries the same
        // number as the round in play, so a joiner reads it for the names and
        // does not mistake it for a fresh round; the series standing rides
        // along all the same, in case this is the message that seats a
        // latecomer next round.
        NetMsg::Start(Invitation {
            seats: self.seats,
            seat,
            terms: self.terms,
            names: wire_table(&self.names),
            kinds: crate::transport::wire_kinds(&self.kinds),
            // Only ever a series' standing: the terms say whether this is
            // one, and a standing left over from another match is not
            // this one's to hand out.
            standing: self.series_standing.filter(|_| self.terms.is_series()),
            beach: self.beach.clone(),
            deal: self.rounds_dealt,
            crowned: self.crowned,
            // At most `MAX_DELAY`, which is what the decoder allows.
            delay: self.session.delay().min(crate::sim::MAX_DELAY) as u8,
        })
    }

    /// Host: ask every peer on the socket how long a round trip to it
    /// takes (`take_pong` hears the answers).
    pub(super) fn ping(&self) {
        self.transport.send(NetMsg::ping());
    }

    /// A `Ping` off the wire, answered at once on a joiner: only the host
    /// asks, so the one heard on a host is nobody's to answer.
    pub(super) fn answer_ping(&self, host: bool, stamp: u32) {
        if !host {
            self.transport.send(NetMsg::Pong { stamp });
        }
    }

    /// A `Pong` off the wire: on the host, one more round trip for the
    /// row of the peer that answered.
    pub(super) fn take_pong(&mut self, host: bool, from: usize, stamp: u32) {
        if host {
            self.peers.answered(from, stamp);
        }
    }

    /// The seat a peer index holds, or `None` for a watcher, and for a
    /// peer the plan has never heard of. The host keeps the same list the
    /// lobby handed out, so a late `Hello` is answered with the seat that
    /// peer already has.
    pub(super) fn seat_of(&self, peer: usize) -> Option<u8> {
        self.peers.seat_of(peer)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sim::DEFAULT_DELAY;
    use crate::transport::UdpTransport;

    /// A session as the host holds it: seat 0, with a launch plan naming
    /// everyone who was at the table when the round began.
    fn hosting(plan: Vec<Option<u8>>) -> OnlineSession {
        let transport = UdpTransport::host(0).expect("game socket");
        let players = (0..=plan.iter().flatten().count() as u8).collect();
        let mut session = OnlineSession::new(
            transport,
            Lockstep::new(0, players, DEFAULT_DELAY),
            2,
            MatchTerms::default(),
        );
        session.peers.deal(&plan);
        session
    }

    /// A bot's seat (route 1) watching a round says its lines under the
    /// seat name it greeted with, as it does in the lobby: the host vouches
    /// its chat under that name, and the sender's own echo said the
    /// player's name instead.
    #[test]
    fn a_bot_watching_speaks_under_its_seat_name() {
        let mut session = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", 47999)).expect("join"),
            Lockstep::observer(vec![0, 1], DEFAULT_DELAY),
            2,
            MatchTerms::default(),
        );
        let me = |session: &OnlineSession| {
            crate::app::lobby::speaking_as(session.bot.as_deref(), "Ana").to_string()
        };
        assert_eq!(me(&session), "Ana", "a person, by name");
        session.bot = Some("Greedy (Ana)".to_string());
        assert_eq!(me(&session), "Greedy (Ana)");
        let mut joined = crate::app::lobby::Joined::dialled(
            UdpTransport::join(("127.0.0.1", 47999)).expect("join"),
            true,
        );
        joined.bot = session.bot.clone();
        assert_eq!(
            joined.greeting("Ana"),
            session.own_greeting(),
            "one greeting in the lobby and in the round"
        );
    }

    /// A peer in line reads the beach being played off the roster by
    /// name: the host's handmade one, from the bytes every board is built
    /// from, and nothing for a built-in map.
    #[test]
    fn the_roster_names_the_handmade_beach_in_play() {
        use crate::app::cycle::Cycle;
        use crate::app::match_setup::MapChoice;
        let mut session = hosting(vec![Some(1)]);
        let NetMsg::Roster { beach, .. } = session.roster_msg() else {
            panic!("a roster");
        };
        assert_eq!(beach, [0; crate::transport::WIRE_NAME], "a built-in map");

        // Two castles, for the two seats at this table.
        let text = "name: Sent Beach\nposts: 2\ncrab: 0,1 R R common\nmap:\n\
                    +-+-+-+-+-+\n|1 . . . .|\n+ + + + + +\n|. . . . 0|\n\
                    + + + + + +\n|. . . . .|\n+-+-+-+-+-+\n";
        let level = crate::sim::Level::parse(text).expect("a level");
        session.beach = crate::lzw::compress(level.to_text().as_bytes(), 8);
        session.terms.map = MapChoice::Custom.index() as u8;
        let NetMsg::Roster { beach, .. } = session.roster_msg() else {
            panic!("a roster");
        };
        assert_eq!(crate::transport::beach_name_from_wire(&beach), "Sent Beach");
    }

    /// A peer in line is shown the table's places, people and bots, and
    /// not the AI's, which the terms count: the lobby's own roster says
    /// it that way, and the two have to agree for one joiner to read both.
    #[test]
    fn the_roster_counts_the_places_as_the_lobby_does() {
        let mut session = hosting(vec![Some(1)]);
        session.seats = 4;
        session.terms.bots = 2;
        let NetMsg::Roster { seats, .. } = session.roster_msg() else {
            panic!("a roster");
        };
        assert_eq!(seats, 2, "the host and Bo; the AI is in the terms");
    }

    /// The launch plan is what membership means, not the seat number: a
    /// lobby spectator and a stranger who greeted mid-round both answer
    /// `None` to `seat_of`, and only one of them can be served.
    #[test]
    fn a_latecomer_is_queued_and_a_lobby_spectator_is_not() {
        // Two peers at launch: one seated, one watching, as it asked.
        let mut session = hosting(vec![Some(1), None]);
        session.peers.row(1).watch = true;
        assert_eq!(session.queue_place(0), None, "a seated peer plays on");
        assert_eq!(
            session.queue_place(1),
            None,
            "and so does one that came to watch: it has been in step since \
             frame zero, which is the whole difference"
        );
        // Anyone the socket picked up afterwards is in line, in arrival
        // order, and told how many are in front of them.
        assert_eq!(session.queue_place(2), Some(NetMsg::Queued { ahead: 0 }));
        assert_eq!(session.queue_place(3), Some(NetMsg::Queued { ahead: 1 }));
        assert_eq!(session.queue_place(9), Some(NetMsg::Queued { ahead: 7 }));
        // A watcher caught up in between takes no chair, so it is nobody's
        // place in line.
        session.peers.row(2).place = Place::LateWatching;
        session.peers.row(2).watch = true;
        assert_eq!(session.queue_place(3), Some(NetMsg::Queued { ahead: 0 }));
        assert_eq!(session.queue_place(9), Some(NetMsg::Queued { ahead: 6 }));
        // One the launch left without a chair, though it never asked to
        // watch, is dealt the next one first: it is ahead of the line.
        session.peers.row(1).watch = false;
        assert_eq!(session.queue_place(3), Some(NetMsg::Queued { ahead: 1 }));
    }

    /// A joiner holds no plan at all, and answers nobody: the star's spokes
    /// never field a greeting.
    #[test]
    fn a_joiner_queues_nobody() {
        let session = hosting(Vec::new());
        assert!(!session.is_host() || session.peers.planned() == 0);
        // With an empty plan every peer index looks late, which is only ever
        // consulted on the host; the guard that matters is `if host`.
        assert_eq!(session.queue_place(0), Some(NetMsg::Queued { ahead: 0 }));
    }
}

#[cfg(test)]
mod next_round_tests {
    use super::*;
    use crate::sim::DEFAULT_DELAY;
    use crate::transport::UdpTransport;

    fn terms(seed: u64) -> MatchTerms {
        MatchTerms {
            seed,
            series: 1,
            ..MatchTerms::default()
        }
    }

    /// A peer that missed every repeat of a resume is still announcing the
    /// pause, and is told again: it used to sit frozen for good, holding
    /// the table up while still talking, so it was never dropped either.
    #[test]
    fn a_peer_that_missed_the_resume_is_told_again() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", port)).expect("join"),
            Lockstep::new(1, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        joiner.transport.send(NetMsg::hello("Bo"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        host.peers.deal(&[Some(1)]);
        // The joiner paused; the host heard it and resumed, and every
        // repeat of that resume was lost.
        let at = joiner.session.request_pause().expect("a player may pause");
        host.session.receive_pause(at);
        host.session.resume();
        assert!(joiner.session.paused());
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            joiner.pump(PlayerAction::None, |_| {});
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.pump(PlayerAction::None, |_| {});
            if !joiner.session.paused() {
                break;
            }
        }
        assert!(!joiner.session.paused(), "told again, and playing on");
    }

    /// The invitation a host re-answers with carries the series as it
    /// stands for as long as the round lasts, and never a standing under
    /// terms that are not a series.
    #[test]
    fn the_standing_rides_every_invitation_of_a_series_only() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(1),
        );
        let standing = SeriesStanding {
            round: 2,
            wins: [1, 0, 0, 0, 0, 0],
        };
        host.series_standing = Some(standing);
        let carried = |host: &OnlineSession| {
            let NetMsg::Start(Invitation { standing, .. }) = host.start_msg(None) else {
                panic!("a Start");
            };
            standing
        };
        assert_eq!(carried(&host), Some(standing));
        host.terms.series = 0;
        assert_eq!(carried(&host), None, "a single round has no standing");
    }

    /// A two-castle beach of the host's own, and a table that grows past
    /// it between rounds. The round is played on the generated beach sized
    /// for the table, and everything says so: the terms, the bytes sent
    /// (none), the roster a peer in line reads, and the board every peer
    /// builds. The terms used to keep naming the handmade beach for good,
    /// and the roster its name, over a generated 20x13 nobody had asked
    /// for.
    #[test]
    fn a_table_that_outgrows_the_hosts_beach_moves_off_it_for_good() {
        use crate::app::cycle::Cycle;
        use crate::app::match_setup::{self, MapChoice};
        let two = crate::sim::Level::parse(
            "name: Two\nposts: 1\ncrab: 0,1 R R common\nmap:\n\
             +-+-+-+\n|0 . .|\n+ + + +\n|. . 1|\n+ + + +\n|. . .|\n+-+-+-+\n",
        )
        .expect("a level");
        let custom = MatchTerms {
            map: MapChoice::Custom.index() as u8,
            ..terms(1)
        };
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            custom,
        );
        host.beach = crate::lzw::compress(two.to_text().as_bytes(), 8);
        let roster_beach = |host: &OnlineSession| {
            let NetMsg::Roster { beach, .. } = host.roster_msg() else {
                panic!("a roster");
            };
            crate::transport::beach_name_from_wire(&beach)
        };
        let _bo = gather(&mut host, &["Bo"]);
        host.peers.deal(&[Some(1)]);
        assert_eq!(roster_beach(&host), "Two", "a pair plays the host's beach");

        // A pair again: the beach holds them, and stays.
        let next = |host: &OnlineSession, seed| {
            match_setup::next_round_terms(host.terms, host.seats, seed)
        };
        host.call_next_round(next(&host, 2), None);
        assert!(match_setup::is_custom(&host.terms));
        assert!(!host.beach.is_empty());

        // Two more turn up while it plays: four at a two-castle beach.
        let port = host.transport.local_addr().expect("addr").port();
        let _more: Vec<UdpTransport> = ["Cy", "Dee"]
            .into_iter()
            .map(|name| {
                let socket = UdpTransport::join(("127.0.0.1", port)).expect("join");
                socket.send(NetMsg::hello(name));
                std::thread::sleep(std::time::Duration::from_millis(10));
                socket
            })
            .collect();
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 3 {
                break;
            }
        }
        assert_eq!(host.transport.peer_count(), 3, "Cy and Dee are aboard");
        host.call_next_round(next(&host, 3), None);
        assert_eq!(host.seats, 4);
        let sized = match_setup::sized_for(4);
        assert_eq!(MapChoice::from_index(usize::from(host.terms.map)), sized);
        assert!(host.beach.is_empty(), "and the beach no longer travels");
        assert_eq!(roster_beach(&host), "", "nor is it named");
        let board = match_setup::board_from(&host.terms, host.seats, &host.beach);
        assert_eq!((board.width(), board.height()), sized.size());
        assert_eq!(board.castle_seats(), 4, "a castle for everybody");

        // And the series goes on from there, the way any generated map
        // does, rather than back to a beach that cannot hold the table.
        host.call_next_round(next(&host, 4), None);
        assert!(!match_setup::is_custom(&host.terms));
    }

    /// The AI's wins are carried to the next round like everyone else's.
    /// They were dropped once, so in an online series against the AI the
    /// AI was back at nothing every round and could never take it.
    #[test]
    fn the_ai_keeps_its_series_wins() {
        let ai = MatchTerms {
            bots: 2,
            ..terms(1)
        };
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0], DEFAULT_DELAY),
            3,
            ai,
        );
        let mut wins = [0; MAX_PLAYERS];
        wins[1] = 1;
        wins[2] = 2;
        let next = host
            .call_next_round(ai, Some(SeriesStanding { round: 3, wins }))
            .expect("a series goes on");
        assert_eq!(host.seats, 3);
        assert_eq!(next.wins[..3], [0, 1, 2], "{:?}", next.wins);
        assert_eq!(next.round, 4);
    }

    /// A host and a real joiner over loopback, played to the point where
    /// the host calls another round. The invitation has to reach the joiner
    /// and rearm it on the same terms, or the two rebuild different beaches
    /// and the round is a desync from frame zero.
    #[test]
    fn a_called_round_rearms_both_ends_on_the_same_terms() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", port)).expect("join"),
            Lockstep::new(1, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        // The greeting is what registers the joiner as a peer host-side.
        joiner.transport.send(NetMsg::hello("Bo"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        assert_eq!(host.transport.peer_count(), 1, "the joiner is at the table");
        host.peers.deal(&[Some(1)]);
        host.names[0] = "Anna".into();
        host.names[1] = "Bo".into();

        host.call_next_round(terms(222), None);
        assert!(host.next_round, "the host arms itself along with the table");
        assert_eq!(host.terms.seed, 222);

        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            joiner.poll_between_rounds(0.0);
            if joiner.next_round {
                break;
            }
        }
        assert!(joiner.next_round, "the invitation reached the joiner");
        assert_eq!(
            joiner.terms, host.terms,
            "and both will build the same beach from it"
        );
        assert_eq!(joiner.seats, host.seats);
        assert_eq!(joiner.names[0], "Anna", "the table travels with it");
        assert_eq!(joiner.session.seat(), Some(1), "keeping its own chair");
        assert_eq!(joiner.session.frame(), 0, "a new round starts at zero");
        assert_eq!(host.session.frame(), 0);
    }

    /// The input that completes a round's last frame is relayed once, as
    /// it arrives, and the host's sim stops right after it. That relay
    /// lost, a joiner sat a frame short of the end for good, waiting on a
    /// player who had long stopped sending. The host on its results card
    /// answers the joiner's resends with the round's last frames.
    #[test]
    fn a_lost_relay_of_the_last_frame_is_sent_again_after_the_round() {
        use crate::sim::InputMsg;
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1, 2], DEFAULT_DELAY),
            3,
            terms(9),
        );
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", port)).expect("join"),
            Lockstep::new(1, vec![0, 1, 2], DEFAULT_DELAY),
            3,
            terms(9),
        );
        joiner.transport.send(NetMsg::hello("Bo"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        host.peers.deal(&[Some(1)]);
        // Seat 2 is Cy's, whose inputs reach the host and are relayed to
        // the joiner, all but the one for the last frame.
        let last = DEFAULT_DELAY + 10;
        for _ in DEFAULT_DELAY..=last {
            let ours = host
                .session
                .commit_local(PlayerAction::None)
                .expect("commit");
            let bos = joiner
                .session
                .commit_local(PlayerAction::None)
                .expect("commit");
            let cys = InputMsg {
                player: 2,
                frame: ours.frame,
                action: PlayerAction::None,
            };
            host.session.receive(bos);
            host.session.receive(cys);
            joiner.session.receive(ours);
            if cys.frame != last {
                joiner.session.receive(cys);
            }
            // Played as they come, as a round is: eleven commits ahead of
            // a frame nobody plays are past the commit lead.
            while host.session.advance().is_some() {}
            while joiner.session.advance().is_some() {}
        }
        assert_eq!(
            host.session.frame(),
            last + 1,
            "the host played the round out"
        );
        assert_eq!(
            joiner.session.frame(),
            last,
            "and the joiner is a frame short"
        );
        assert_eq!(joiner.session.awaiting(), vec![2]);

        // The host is on its results card; the joiner is still playing.
        for _ in 0..40 {
            joiner.pump(
                PlayerAction::None,
                |net| {
                    while net.session.advance().is_some() {}
                },
            );
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            std::thread::sleep(std::time::Duration::from_millis(5));
            if joiner.session.frame() > last {
                break;
            }
        }
        assert!(
            joiner.session.frame() > last,
            "the joiner finished the round: {:?}",
            joiner.session.awaiting()
        );
    }

    /// The input delay a round is dealt with is the one the slowest player
    /// it seats needs, from the round trips the host measured; a watcher's
    /// link holds nobody up and does not count, and a table nobody has
    /// answered at plays the default. One hitch among a few answers is not
    /// the link, and an answer from long ago gives way to fresh ones. A
    /// joiner plays the delay its `Start` carries.
    #[test]
    fn a_round_is_dealt_the_delay_its_slowest_player_needs() {
        let mut peers = PeerBook::default();
        peers.reach(3);
        let plan = [Some(1), Some(2), None];
        assert_eq!(input_delay(&peers, &plan), DEFAULT_DELAY, "nobody answered");
        peers.row(0).round_trips.note(0.010);
        peers.row(1).round_trips.note(0.060);
        peers.row(1).round_trips.note(0.070);
        peers.row(2).round_trips.note(0.500);
        assert_eq!(
            input_delay(&peers, &plan),
            crate::sim::delay_for_round_trip(0.070),
            "the slowest seated answer of two, not the watcher's"
        );
        assert_eq!(crate::sim::delay_for_round_trip(0.070), 5);
        // A third answer, and the one hitch among them is let go.
        peers.row(1).round_trips.note(0.020);
        assert_eq!(
            input_delay(&peers, &plan),
            crate::sim::delay_for_round_trip(0.060)
        );
        assert_eq!(crate::sim::delay_for_round_trip(0.060), 4);
        // Minutes of a round later, the results card's own pings say the
        // link is quick now, and the old answers are let go.
        peers.age(300.0);
        assert_eq!(
            input_delay(&peers, &plan),
            crate::sim::delay_for_round_trip(0.060),
            "stale, but nothing fresher"
        );
        peers.row(1).round_trips.note(0.010);
        assert_eq!(input_delay(&peers, &plan), DEFAULT_DELAY);

        let joiner = OnlineSession::invited(
            UdpTransport::join(("127.0.0.1", 9)).expect("socket"),
            Invitation {
                seat: Some(1),
                delay: 5,
                ..Invitation::blank()
            },
        );
        assert_eq!(joiner.session.delay(), 5);
    }

    /// Register `joiners` on the host's socket in order, so their peer
    /// indices are known, and hand back the sockets they greeted from.
    fn gather(host: &mut OnlineSession, joiners: &[&str]) -> Vec<UdpTransport> {
        let port = host.transport.local_addr().expect("addr").port();
        let mut sockets = Vec::new();
        for (want, name) in joiners.iter().enumerate() {
            let socket = UdpTransport::join(("127.0.0.1", port)).expect("join");
            socket.send(NetMsg::hello(name));
            for _ in 0..40 {
                std::thread::sleep(std::time::Duration::from_millis(5));
                host.poll_between_rounds(0.0);
                if host.transport.peer_count() > want {
                    break;
                }
            }
            assert_eq!(host.transport.peer_count(), want + 1, "{name} is aboard");
            sockets.push(socket);
        }
        sockets
    }

    /// A greeting on the results card carries the same two things it
    /// carries in the lobby, a name and a wish to watch, and the host has
    /// to keep both: they are what the next round's table is dealt from.
    #[test]
    fn a_greeting_on_the_results_card_keeps_its_name_and_its_wish() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(1),
        );
        host.names[0] = "Anna".into();
        let _bo = gather(&mut host, &["Bo"]);
        let port = host.transport.local_addr().expect("addr").port();
        let dee = UdpTransport::join(("127.0.0.1", port)).expect("join");
        dee.send(crate::app::lobby::greeting(None, true, "Dee"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 2 {
                break;
            }
        }
        assert_eq!(host.transport.peer_count(), 2, "Dee is aboard");

        host.call_next_round(terms(2), None);
        assert_eq!(host.peers.seat_of(0), Some(1), "Bo is dealt a chair");
        assert_eq!(host.names[1], "Bo", "under its own name");
        assert_eq!(
            host.peers.seat_of(1),
            None,
            "Dee gets the watching place it asked for"
        );
        assert!(host.peers.get(1).is_some_and(Peer::watches));
    }

    /// What a greeting earns: a peer in the plan is told its place again,
    /// under whatever it now calls itself, and one behind the plan is told
    /// how many are ahead of it.
    #[test]
    fn a_greeting_is_answered_with_the_peers_place_and_keeps_its_name() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(1),
        );
        host.names[0] = "Anna".into();
        let _sockets = gather(&mut host, &["Bo", "Cy"]);
        host.peers.deal(&[Some(1)]);
        host.names[1] = "Bo".into();

        let answer = host.answer_greeting(0, "Bobby", false);
        assert!(
            matches!(answer, NetMsg::Start(Invitation { seat: Some(1), .. })),
            "seated, and told so: {answer:?}"
        );
        assert_eq!(host.names[1], "Bobby", "under the name it greeted with");

        let answer = host.answer_greeting(1, "Cy", false);
        assert!(
            matches!(answer, NetMsg::Queued { ahead: 0 }),
            "behind the plan, first in line: {answer:?}"
        );
        assert_eq!(
            host.peer_name(1),
            Some("Cy"),
            "and remembered for the next deal"
        );
    }

    /// A seated peer that arms W between rounds is told it is watching,
    /// and the wish is kept for the next deal.
    #[test]
    fn a_watch_from_a_seated_peer_is_told_to_watch() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(1),
        );
        let _sockets = gather(&mut host, &["Bo"]);
        host.peers.deal(&[Some(1)]);
        let answer = host.answer_greeting(0, "", true);
        assert!(
            matches!(answer, NetMsg::Start(Invitation { seat: None, .. })),
            "{answer:?}"
        );
        assert!(host.peers.get(0).is_some_and(Peer::watches));
        host.call_next_round(terms(2), None);
        assert_eq!(
            host.peers.seat_of(0),
            None,
            "dealt the watching place it asked for"
        );
    }

    /// A peer that leaves while the scores are being read leaves nothing
    /// stalled, so the host has to notice: one silent for as long as the
    /// round waits before giving a seat up is forgotten before the next
    /// table is dealt, rather than dealt a chair that freezes it.
    #[test]
    fn a_peer_silent_on_the_results_card_is_not_dealt_a_chair() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1, 2], DEFAULT_DELAY),
            3,
            terms(1),
        );
        host.names[0] = "Anna".into();
        let sockets = gather(&mut host, &["Bo", "Cy"]);
        // Cy keeps greeting, as a joiner on the card does; Bo has gone.
        sockets[1].send(NetMsg::hello("Cy"));
        std::thread::sleep(std::time::Duration::from_millis(20));
        host.poll_between_rounds(presence::ABANDON_AFTER);
        assert_eq!(host.transport.peer_count(), 1, "Bo is forgotten");
        assert_eq!(host.peers.len(), 1, "row and all");
        assert_eq!(
            host.peer_name(0),
            Some("Cy"),
            "and Cy moved down into its index"
        );

        host.call_next_round(terms(2), None);
        assert_eq!(host.seats, 2, "Anna and Cy, nobody else");
        assert_eq!(host.names[1], "Cy");
    }

    /// A peer that leaves the results card says so, and is forgotten on
    /// the spot: Enter mid-series, the next round called a moment later,
    /// and no ghost dealt a chair to freeze the table on.
    #[test]
    fn a_peer_that_says_goodbye_on_the_card_is_forgotten_at_once() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1, 2], DEFAULT_DELAY),
            3,
            terms(1),
        );
        host.names[0] = "Anna".into();
        let mut sockets = gather(&mut host, &["Bo", "Cy"]);
        // Bo walks off the card; its socket closes and says so.
        drop(sockets.remove(0));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        assert_eq!(host.transport.peer_count(), 1, "Bo is forgotten");
        assert_eq!(host.peer_name(0), Some("Cy"), "and Cy moved down");
        host.call_next_round(terms(2), None);
        assert_eq!(host.seats, 2, "Anna and Cy, nobody else");
    }

    /// The other way round: a host that quits says so, and a joiner on
    /// the card (or in the round) knows it a moment later (the goodbye's
    /// grace) instead of after twenty seconds of silence.
    #[test]
    fn a_host_that_quits_is_gone_at_once() {
        let host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(5),
        );
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", port)).expect("join"),
            Lockstep::new(1, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(5),
        );
        joiner.transport.send(NetMsg::hello("Bo"));
        let mut host = host;
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        assert!(!joiner.host_gone());
        drop(host);
        let step = 0.05;
        let mut waited = 0.0;
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            joiner.poll_between_rounds(step);
            waited += step;
            if joiner.host_gone() {
                break;
            }
        }
        assert!(joiner.host_gone(), "told, not left to wait it out");
        assert!(waited < 1.0, "and within a second: {waited}");
    }

    /// The host's call for the next round can land on a joiner that is
    /// still ticking out the last frames of the round the host has
    /// finished. The lockstep it takes up is at frame zero and the board
    /// is still the old one: the pump that rearmed must not tick, or the
    /// new round's first two frames are played out on the wrong beach and
    /// the joiner walks into the arena two frames ahead of every table.
    #[test]
    fn a_call_that_lands_mid_pump_does_not_tick_the_old_board() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", port)).expect("join"),
            Lockstep::new(1, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        joiner.transport.send(NetMsg::hello("Bo"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        host.peers.deal(&[Some(1)]);
        host.call_next_round(terms(222), None);

        let mut ticks = 0;
        let mut rearmed_in = None;
        for pump in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            let before = ticks;
            joiner.pump(PlayerAction::None, |_| ticks += 1);
            if joiner.next_round {
                assert_eq!(ticks, before, "the pump that rearmed did not tick");
                rearmed_in = Some(pump);
                break;
            }
        }
        assert!(rearmed_in.is_some(), "the call reached the joiner");
        assert_eq!(
            joiner.session.frame(),
            0,
            "and the new lockstep is untouched"
        );
        // Nor by any pump after it, until the shell has walked the joiner
        // out of the arena: the board is still the old one, and the frames
        // the new lockstep has in hand are the new round's.
        let commit = joiner.session.next_commit();
        for _ in 0..10 {
            joiner.pump(PlayerAction::None, |_| ticks += 1);
        }
        assert_eq!(
            rearmed_in.map(|_| ticks),
            Some(0),
            "the old board was ticked"
        );
        assert_eq!(joiner.session.next_commit(), commit, "or committed for");
    }

    /// A series tally is kept by seat, and seats move. A peer that leaves
    /// between rounds frees its chair and everyone behind it shuffles up
    /// one, so the host re-deals the wins along with the chairs.
    ///
    /// Without that, a tally kept by seat number hands the departed
    /// player's rounds to whoever moved into their seat, and the table
    /// crowns the wrong champion.
    #[test]
    fn the_wins_follow_the_chairs_when_a_peer_leaves() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1, 2], DEFAULT_DELAY),
            3,
            terms(1),
        );
        let _sockets = gather(&mut host, &["Bo", "Cy"]);
        host.peers.deal(&[Some(1), Some(2)]);
        host.names[0] = "Anna".into();
        host.names[1] = "Bo".into();
        host.names[2] = "Cy".into();

        // Three rounds played: Anna one, Bo two, Cy three.
        let standing = SeriesStanding {
            round: 3,
            wins: [1, 2, 3, 0, 0, 0],
        };
        // And Bo walks off while the results card is up, which frees seat
        // one and moves Cy up into it.
        host.transport.forget(0);
        host.peers.forget(0);

        let next = host
            .call_next_round(terms(2), Some(standing))
            .expect("a series has a standing");
        assert_eq!(next.round, 4, "the round about to begin");
        assert_eq!(
            next.wins[0], 1,
            "the host keeps its own, seat zero being its own"
        );
        assert_eq!(
            next.wins[1], 3,
            "Cy took its three rounds up the table with it"
        );
        assert_eq!(next.wins[2], 0, "and nothing of Bo's was left for anyone");
        assert_eq!(host.names[1], "Cy", "the name moved with the chair too");
        assert_eq!(host.series_standing, Some(next), "and the host holds it");
    }

    /// A bot that came straight to the host (route 2) keeps its wins
    /// across rounds like anyone else. Its chair is re-dealt behind the
    /// people every round, and the tally has to follow it there.
    #[test]
    fn a_hosts_bot_keeps_its_wins_into_the_next_round() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1, 2], DEFAULT_DELAY),
            3,
            terms(1),
        );
        let _sockets = gather(&mut host, &["Bo"]);
        host.peers.deal(&[Some(1)]);
        host.names[0] = "Anna".into();
        host.names[1] = "Bo".into();
        host.bots_here = vec![HostedBot {
            seat: 2,
            id: 7,
            name: "Greedy".into(),
        }];
        let standing = SeriesStanding {
            round: 2,
            wins: [0, 0, 2, 0, 0, 0],
        };
        // Bo leaves, so the bot moves up from seat two to seat one.
        host.transport.forget(0);
        host.peers.forget(0);
        let next = host
            .call_next_round(terms(2), Some(standing))
            .expect("a series has a standing");
        assert_eq!(host.bots_here[0].seat, 1, "the bot moved up a chair");
        assert_eq!(next.wins[1], 2, "and took its two wins with it");
        assert_eq!(next.wins[2], 0);
    }

    /// A bot greeting a host whose beach said bots are not welcome is
    /// asked to go, and dealt no chair, on the results card as in the
    /// lobby. Welcome, the same greeting is taken in.
    #[test]
    fn a_bot_is_turned_away_where_bots_are_not_welcome() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(1),
        );
        host.home.bots_welcome = false;
        let port = host.transport.local_addr().expect("addr").port();
        let mut bot = UdpTransport::join(("127.0.0.1", port)).expect("join");
        bot.send(NetMsg::hello_bot("Greedy"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() > 0 {
                break;
            }
        }
        let mut told = false;
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            told |= bot
                .recv_all()
                .into_iter()
                .any(|(msg, _)| msg == NetMsg::NoBots);
            if told {
                break;
            }
        }
        assert!(told, "the bot was asked to go");
        assert!(
            host.peers.get(0).is_some_and(|row| row.watch && !row.bot),
            "and is dealt no chair"
        );
        assert_eq!(host.peers.seat_plan(1), vec![None]);

        // Welcome: the same greeting is a player at the table.
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(1),
        );
        host.home.bots_welcome = true;
        let port = host.transport.local_addr().expect("addr").port();
        let bot = UdpTransport::join(("127.0.0.1", port)).expect("join");
        bot.send(NetMsg::hello_bot("Greedy"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.peers.get(0).is_some_and(|row| row.bot) {
                break;
            }
        }
        assert!(host.peers.get(0).is_some_and(|row| row.bot && !row.watch));
    }

    /// A peer in line for the next round is not sent this one.
    ///
    /// It is not simulating anything: it is sitting in the lobby looking at
    /// the terms card, and the frames it would be sent are frames it cannot
    /// use. Unheld, four people in line took a six-seat table from 930
    /// datagrams a second to 1657.
    #[test]
    fn a_peer_in_line_is_not_sent_the_round() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(11),
        );
        let mut sockets = gather(&mut host, &["Bo"]);
        host.peers.deal(&[Some(1)]);

        // And then somebody turns up mid-round, who is in line rather than
        // at the table: the default place, which is what the plan leaves
        // every peer it does not reach.
        let port = host.transport.local_addr().expect("addr").port();
        let mut latecomer = UdpTransport::join(("127.0.0.1", port)).expect("join");
        latecomer.send(NetMsg::hello("Dee"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 2 {
                break;
            }
        }
        assert_eq!(host.transport.peer_count(), 2, "Dee is on the socket");
        assert_eq!(host.peers.planned(), 1, "and in line, not at the table");

        for _ in 0..5 {
            host.pump(PlayerAction::None, |_| {});
            std::thread::sleep(std::time::Duration::from_millis(5));
        }
        let inputs = |transport: &mut UdpTransport| {
            transport
                .recv_all()
                .iter()
                .filter(|(msg, _)| matches!(msg, NetMsg::Inputs(_)))
                .count()
        };
        assert!(inputs(&mut sockets[0]) > 0, "the table hears the round");
        assert_eq!(inputs(&mut latecomer), 0, "the queue does not");
    }

    /// The host is the only peer that can tell who sent what, so it is the
    /// only one that can catch a peer speaking for somebody else's seat.
    ///
    /// It matters because of the relay: the spokes of the star hear each
    /// other only through the hub, so an input the hub believes is an input
    /// the whole table believes, and one unchecked datagram is a desync
    /// lockstep has no way back from.
    #[test]
    fn a_peer_speaks_only_for_the_seat_it_holds() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1, 2], DEFAULT_DELAY),
            3,
            terms(7),
        );
        let mut sockets = gather(&mut host, &["Bo", "Cy", "Dee"]);
        host.peers.deal(&[Some(1), Some(2), None]);

        // Bo holds seat one. It sends for its own chair, and for Cy's.
        let own = crate::sim::InputMsg {
            player: 1,
            frame: DEFAULT_DELAY,
            action: PlayerAction::Remove { x: 1, y: 1 },
        };
        let forged = crate::sim::InputMsg {
            player: 2,
            frame: DEFAULT_DELAY,
            action: PlayerAction::Remove { x: 2, y: 2 },
        };
        // What reaches Dee, watching, is the question: she is sent every
        // seat, and the hub is the only way Bo can reach her at all, so
        // what the hub passes on is what the table plays.
        let mut relayed: Vec<u8> = Vec::new();
        for _ in 0..40 {
            // Both in one batch, which is the shape the weeding has to
            // survive: the honest input and the forged one arrive in a
            // single datagram, and only one of them may be passed on.
            sockets[0].send(NetMsg::Inputs(Batch {
                inputs: vec![own, forged],
                ..Batch::default()
            }));
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.pump(PlayerAction::None, |_| {});
            for (msg, _) in sockets[2].recv_all() {
                if let NetMsg::Inputs(batch) = msg {
                    relayed.extend(batch.inputs.iter().map(|input| input.player));
                }
            }
            if relayed.contains(&1) {
                break;
            }
        }
        assert!(relayed.contains(&1), "the chair it holds reaches the table");
        assert!(
            !relayed.contains(&2),
            "the chair it does not hold never does: {relayed:?}"
        );
    }

    /// The stale `Start` a host re-answers a stray greeting with must not
    /// read as a new round, or every late hello would restart the match.
    /// Nor may one from the round before: the host answers greetings on
    /// its results card with the round just played, and an answer that
    /// the next round's call overtakes on the wire had a seed of its own,
    /// so the joiner took it up and re-armed onto the old beach while the
    /// table played the new one.
    #[test]
    fn an_invitation_already_dealt_is_not_a_new_round() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        let _bo = gather(&mut host, &["Bo"]);
        host.peers.deal(&[Some(1)]);
        let start = |host: &OnlineSession| {
            let NetMsg::Start(start) = host.start_msg(Some(1)) else {
                panic!("a Start");
            };
            start
        };
        host.call_next_round(terms(222), None);
        let first = start(&host);
        let mut joiner = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(1, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        assert!(joiner.is_next_round(&first), "the first round dealt");
        joiner.take_up(first.clone());
        assert!(!joiner.is_next_round(&first), "the round it is in");

        // The results card: a greeting answered with round one, then the
        // call for round two, which the answer arrives after.
        host.call_next_round(terms(333), None);
        let second = start(&host);
        assert!(joiner.is_next_round(&second), "the next round");
        joiner.take_up(second);
        assert!(
            !joiner.is_next_round(&first),
            "a beach it has played, under a seed it is not on"
        );
        assert_eq!(joiner.terms.seed, 333);
    }

    /// The host's word that it stopped picking the next map is said again
    /// on its results card, like the word that it started: lost, the table
    /// went on being told the host was picking until the next round.
    #[test]
    fn a_lost_word_that_the_host_stopped_picking_is_said_again() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", port)).expect("join"),
            Lockstep::new(1, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        joiner.transport.send(NetMsg::hello("Bo"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        host.peers.deal(&[Some(1)]);
        host.say_picking(true);
        std::thread::sleep(std::time::Duration::from_millis(5));
        joiner.poll_between_rounds(0.0);
        assert!(joiner.picking_map, "told the host is picking");
        // The host puts its page away, and that word is lost.
        host.say_picking(false);
        std::thread::sleep(std::time::Duration::from_millis(5));
        drop(joiner.transport.recv_all());
        for _ in 0..3 {
            host.poll_between_rounds(1.0);
            std::thread::sleep(std::time::Duration::from_millis(5));
            joiner.poll_between_rounds(0.0);
        }
        assert!(!joiner.picking_map, "told again that it stopped");
    }

    /// A player who sat a round out because the table was full asked for
    /// a chair, not a seat in the stands: the next round deals it one.
    #[test]
    fn a_player_left_without_a_chair_is_seated_when_one_frees() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(1),
        );
        host.peers.deal(&[Some(1), None]);
        assert_eq!(host.peers.seat_plan(2), vec![Some(1), Some(2)]);
    }

    /// Whoever queued while the round played gets a chair in the next one,
    /// and the AI gives way to them rather than the other way about.
    #[test]
    fn the_queue_is_seated_next_round() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(1),
        );
        // One peer played, one watched, two turned up while they played.
        host.peers.deal(&[Some(1), None]);
        host.peers.row(1).watch = true;
        let plan = host.peers.seat_plan(4);
        assert_eq!(
            plan,
            vec![Some(1), None, Some(2), Some(3)],
            "the player keeps its seat, the watcher keeps watching, and the \
             two who waited are seated behind them"
        );
        // The AI takes what the humans leave, however much the terms asked
        // for. No peer ever connected to this socket, so the table the call
        // actually plans is the host alone, five chairs for the AI.
        let mut greedy = terms(2);
        greedy.bots = 5;
        host.call_next_round(greedy, None);
        assert_eq!(host.seats, MAX_PLAYERS as u8);
        assert_eq!(host.terms.bots, MAX_PLAYERS as u8 - 1);
        // And the plan is drawn from who is actually connected, not from
        // last round's list: a peer that left does not hold a chair.
        assert!(host.peers.is_empty(), "nobody is connected any more");
    }

    /// One fixed tick, as long as the shell's.
    const TICK: f32 = 1.0 / crate::sim::TICKS_PER_SECOND as f32;

    /// The ticks a peer stands through from the call to its round's first
    /// frame: the interlude (between rounds), the line-up and the count.
    fn ticks_before(interlude: bool) -> u32 {
        let interlude = match interlude {
            true => crate::app::tournament::INTERLUDE_SECS,
            false => 0.0,
        };
        let secs =
            interlude + crate::app::lineup::LINEUP_SECS + crate::app::countdown::COUNTDOWN_SECS;
        (secs / TICK).ceil() as u32
    }

    /// One tick of a round being played, as the shell runs it: the
    /// silences aged and the stalled given up on, then the pump.
    fn play_a_tick(net: &mut OnlineSession) {
        net.keep_in_touch(TICK);
        net.abandon_stalled(TICK, false);
        net.pump(PlayerAction::None, |net| {
            net.session.advance();
        });
    }

    /// A pause long enough for a datagram over loopback.
    fn beat() {
        std::thread::sleep(std::time::Duration::from_millis(1));
    }

    /// How long a joiner says nothing while it builds the arena: the one
    /// long frame of the way in, which nothing networks through.
    const LOADING: u32 = 45;

    /// Where a joiner is on its way into a round.
    enum Way {
        /// On the results card (or in the lobby), its `Start` not heard.
        Waiting,
        /// Building the arena, this many ticks from done.
        Loading(u32),
        /// Its round dealt, this many ticks from its first frame.
        Before(u32),
        Playing,
    }

    /// Walk a host and its joiner into the round the host dealt `before`
    /// ticks ago, the joiner on `way`, both as the shell does: standing
    /// through the interlude, line-up and count before playing. Returns
    /// once both have been playing for a while.
    fn walk_in(
        host: &mut OnlineSession,
        joiner: &mut OnlineSession,
        interlude: bool,
        mut host_left: u32,
    ) {
        let mut way = Way::Waiting;
        for _ in 0..3 * ticks_before(interlude) + 300 {
            beat();
            match host_left {
                0 => play_a_tick(host),
                _ => {
                    host_left -= 1;
                    host.before_the_round(TICK);
                }
            }
            beat();
            way = match way {
                Way::Waiting => {
                    joiner.poll_between_rounds(TICK);
                    match std::mem::take(&mut joiner.next_round) {
                        true => Way::Loading(LOADING),
                        false => Way::Waiting,
                    }
                }
                Way::Loading(0) => Way::Before(ticks_before(interlude)),
                Way::Loading(left) => Way::Loading(left - 1),
                Way::Before(0) | Way::Playing => {
                    play_a_tick(joiner);
                    Way::Playing
                }
                Way::Before(left) => {
                    joiner.before_the_round(TICK);
                    Way::Before(left - 1)
                }
            };
        }
    }

    /// A joiner whose call to the next round was lost hears it again while
    /// the host still stands in its interlude, and is seated and playing
    /// once the line-up and count are through, not given to the AI: it
    /// used to hear the call only once the host's round ran, then stood
    /// through its own interlude, line-up and count without a word, past
    /// the host's patience.
    #[test]
    fn a_start_lost_between_rounds_still_seats_the_joiner() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            Lockstep::new(0, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        let port = host.transport.local_addr().expect("addr").port();
        let mut joiner = OnlineSession::new(
            UdpTransport::join(("127.0.0.1", port)).expect("join"),
            Lockstep::new(1, vec![0, 1], DEFAULT_DELAY),
            2,
            terms(111),
        );
        joiner.transport.send(NetMsg::hello("Bo"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        host.peers.deal(&[Some(1)]);
        host.call_next_round(terms(222), None);
        host.next_round = false;
        // The call never reaches the joiner.
        std::thread::sleep(std::time::Duration::from_millis(5));
        drop(joiner.transport.recv_all());
        walk_in(&mut host, &mut joiner, true, ticks_before(true));
        assert!(host.abandoned.is_empty(), "nobody was given up on");
        assert!(!joiner.dropped());
        assert_eq!(joiner.terms, host.terms, "on the round the host called");
        assert_eq!(joiner.session.seat(), Some(1));
        assert!(
            joiner.session.frame() > 60,
            "and playing it: frame {}",
            joiner.session.frame()
        );
    }

    /// The same at the launch: a joiner in the lobby whose `Start` was
    /// lost is sent it again while the host stands on its line-up, and is
    /// seated when its own count is out.
    #[test]
    fn a_start_lost_at_the_launch_still_seats_the_joiner() {
        let mut host = OnlineSession::new(
            UdpTransport::host(0).expect("host socket"),
            lockstep_for(Some(0), (0..2).collect(), DEFAULT_DELAY),
            2,
            terms(111),
        );
        let port = host.transport.local_addr().expect("addr").port();
        let mut lobby = UdpTransport::join(("127.0.0.1", port)).expect("join");
        lobby.send(NetMsg::hello("Bo"));
        for _ in 0..40 {
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.poll_between_rounds(0.0);
            if host.transport.peer_count() == 1 {
                break;
            }
        }
        host.deal(
            &[Some(1)],
            [crate::sim::SeatKind::Human; MAX_PLAYERS],
            Vec::new(),
            None,
        );
        // The launch's `Start` never reaches the lobby, which greets on,
        // once a second, as the lobby does.
        std::thread::sleep(std::time::Duration::from_millis(5));
        drop(lobby.recv_all());
        let mut host_left = ticks_before(false);
        let mut greet_in = 0.0;
        let mut invited = None;
        for _ in 0..3 * ticks_before(false) {
            beat();
            match host_left {
                0 => play_a_tick(&mut host),
                _ => {
                    host_left -= 1;
                    host.before_the_round(TICK);
                }
            }
            beat();
            if crate::app::lobby::once_a_second(&mut greet_in, TICK) {
                lobby.send(NetMsg::hello("Bo"));
            }
            invited = lobby.recv_all().into_iter().find_map(|(msg, _)| {
                let NetMsg::Start(start) = msg else {
                    return None;
                };
                Some(start)
            });
            if invited.is_some() {
                break;
            }
        }
        let start = invited.expect("the Start is sent again");
        let mut joiner = OnlineSession::invited(lobby, start);
        // In the arena on it, straight into its line-up and count.
        joiner.next_round = true;
        walk_in(&mut host, &mut joiner, false, host_left);
        assert!(host.abandoned.is_empty(), "nobody was given up on");
        assert!(!joiner.dropped());
        assert_eq!(joiner.session.seat(), Some(1));
        assert!(
            joiner.session.frame() > 60,
            "playing: frame {}",
            joiner.session.frame()
        );
    }
}