pinch-points 0.8.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.
///
/// Named rather than a tuple: `(seats, seat, ..)` is a count and an index
/// next to each other, kept apart only by their order. Taken up by
/// [`OnlineSession::take_up`] whether it arrives in the lobby, mid-round or
/// on the results card, so there is one way to begin a round.
pub struct Invitation {
    pub seats: u8,
    pub seat: Option<u8>,
    pub terms: MatchTerms,
    pub names: [crate::transport::WireName; MAX_PLAYERS],
    /// What holds each seat, in wire form (`transport::kind_byte`).
    pub kinds: [u8; MAX_PLAYERS],
    /// Where the series stands, as the host says. A peer that greeted
    /// mid-series is seated with the table's own tally rather than
    /// starting a fresh one. `None` for a single round.
    pub standing: Option<SeriesStanding>,
    /// The host's own beach, when it sent one. A joiner has never seen the
    /// file it came from, so it arrives with the invitation or not at all.
    pub beach: Vec<u8>,
}

/// The lockstep a peer plays a round over: the `humans` low seats, from
/// the given chair, or from no chair at all for a watcher.
pub(crate) fn lockstep_for(seat: Option<u8>, humans: u8) -> Lockstep {
    let players: Vec<u8> = (0..humans).collect();
    match seat {
        None => Lockstep::observer(players, crate::sim::DEFAULT_DELAY),
        Some(seat) => Lockstep::new(seat, players, crate::sim::DEFAULT_DELAY),
    }
}

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, ..
        } = &invitation;
        let mut session = OnlineSession::new(
            transport,
            lockstep_for(*seat, terms.humans(*seats)),
            *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,
        } = invitation;
        self.beach = beach;
        self.begin_round(seats, seat, terms, table_from_wire(&names));
        self.kinds = kinds.map(crate::transport::kind_from_byte);
        self.series_standing = standing;
    }

    /// 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. Inputs and hashes belong to the round that just ended, and are
    /// dropped with it.
    ///
    /// 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) {
            // The same greeting the lobby sends, wish to watch included,
            // so a watcher on the card does not greet as a nameless
            // player.
            let me = self
                .session
                .seat()
                .map_or("", |seat| &self.names[usize::from(seat)]);
            let hello = match self.bot {
                true => NetMsg::hello_bot(me),
                false => crate::app::lobby::greeting(self.watch_wish(), me),
            };
            self.transport.send(hello);
        }
        for (msg, from) in self.transport.recv_all() {
            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);
                    }
                }
                NetMsg::Watch { name } => {
                    if host {
                        let answer =
                            self.answer_greeting(from, &name_from_wire(&name), true);
                        self.transport.send_to(from, answer);
                    }
                }
                NetMsg::Start {
                    seats,
                    seat,
                    terms,
                    names,
                    kinds,
                    standing,
                    beach,
                } => {
                    if !host && self.is_next_round(&terms) {
                        self.take_up(Invitation {
                            seats,
                            seat,
                            terms,
                            names,
                            kinds,
                            standing,
                            beach,
                        });
                        self.next_round = true;
                    }
                }
                // 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 } => {
                    if let Some(said) = self.take_chat(host, from, name, text) {
                        self.heard.push(said);
                    }
                }
                NetMsg::Inputs(_)
                | NetMsg::Hash { .. }
                | NetMsg::Pause { .. }
                | NetMsg::Resume { .. }
                | NetMsg::Queued { .. }
                | NetMsg::Roster { .. }
                | 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 { .. }
                // The round is over; the next greeting is answered with
                // the next round's `Start`, which is how they come in.
                | NetMsg::CatchUp { .. }
                // Calls are for a round being played, and this one is over.
                | NetMsg::SpectatorPick { .. }
                | NetMsg::CrowdPicks { .. }
                | NetMsg::Incompatible { .. } => {}
                // The host asked us to leave while the card was up.
                NetMsg::Kicked => self.kicked |= !host,
            }
        }
        // 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: the table for the next round, admitting whoever queued while
    /// this one played.
    ///
    /// Seats are handed out in peer order as they were at the launch, so
    /// those still here keep theirs, and the queue fills what is left,
    /// pushing the AI back a seat at a time. Returns the new plan, one
    /// entry per peer, `None` for a peer that watches or that the table
    /// could not fit.
    fn next_plan(&self, peers: usize) -> Vec<Option<u8>> {
        let mut next = 1u8; // the host keeps seat 0
        (0..peers)
            .map(|peer| {
                // A watcher keeps no chair: one that asked, at the launch
                // or since. One that sat the last round out for want of a
                // chair is dealt one when there is one.
                let watches = self.peers.get(peer).is_some_and(Peer::watches);
                if watches || usize::from(next) >= MAX_PLAYERS {
                    return None;
                }
                let seat = next;
                next += 1;
                Some(seat)
            })
            .collect()
    }

    /// 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 {
        NetMsg::Roster {
            seats: self.seats,
            names: wire_table(&self.names),
            terms: self.terms,
            kinds: crate::transport::wire_kinds(&self.kinds),
        }
    }

    /// 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::Kicked);
        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: 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 in peer order
    /// every round (they have to stay the contiguous `0..humans` the sim
    /// fills the top of with AI), so a peer that leaves shifts everyone
    /// behind it up a chair and the wins have to move with them.
    pub fn call_next_round(
        &mut self,
        mut terms: MatchTerms,
        standing: Option<SeriesStanding>,
    ) -> Option<SeriesStanding> {
        let peers = self.transport.peer_count();
        let plan = self.next_plan(peers);
        let people = 1 + plan.iter().flatten().count() as u8;
        // The bots that came straight to this host (route 2) sit after the
        // people again, as many as there are chairs for.
        let mut bots_here = std::mem::take(&mut self.bots_here);
        bots_here.truncate(MAX_PLAYERS - usize::from(people));
        let bots_were: Vec<u8> = bots_here.iter().map(|(seat, ..)| *seat).collect();
        for (i, bot) in bots_here.iter_mut().enumerate() {
            bot.0 = people + i as u8;
        }
        let humans = people + bots_here.len() as u8;
        // A beach needs two castles, and a host whose table has emptied is
        // still entitled to another round, against the AI, since playing
        // itself is not a round. The same floor `seat_count` keeps.
        terms.bots = terms
            .bots
            .min(MAX_PLAYERS as u8 - humans)
            .max(2u8.saturating_sub(humans));
        let seats = humans + terms.bots;
        // 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.
        terms = crate::app::match_setup::fit_terms(terms, seats);
        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 ((new_seat, ..), old_seat) in bots_here.iter().zip(&bots_were) {
                new_wins[usize::from(*new_seat)] = wins[usize::from(*old_seat)];
            }
            // And the game's AI, which sits in the top chairs: the first AI
            // seat 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.humans(self.seats)..self.seats).zip(humans..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 mut names: [String; MAX_PLAYERS] = Default::default();
        names[0].clone_from(&self.names[0]);
        for (peer, slot) in plan.iter().enumerate() {
            if let Some(seat) = slot
                && let Some(name) = self.peer_name(peer)
            {
                names[usize::from(*seat)] = name.to_string();
            }
        }
        for (seat, _, name) in &bots_here {
            names[usize::from(*seat)].clone_from(name);
        }
        let wire = wire_table(&names);
        // What holds each seat: the host a person, a peer whatever it
        // greeted as, the host's bots bots, the top seats the AI.
        let mut kinds = [crate::sim::SeatKind::Human; MAX_PLAYERS];
        for (seat, ..) in &bots_here {
            kinds[usize::from(*seat)] = crate::sim::SeatKind::Bot;
        }
        for (peer, slot) in plan.iter().enumerate() {
            if let Some(seat) = slot
                && self.peers.get(peer).is_some_and(|row| row.bot)
            {
                kinds[usize::from(*seat)] = crate::sim::SeatKind::Bot;
            }
        }
        for seat in terms.humans(seats)..seats {
            kinds[usize::from(seat)] = crate::sim::SeatKind::Ai;
        }
        for (peer, slot) in plan.iter().enumerate() {
            self.transport.send_to(
                peer,
                NetMsg::Start {
                    seats,
                    seat: *slot,
                    terms,
                    names: wire,
                    kinds: crate::transport::wire_kinds(&kinds),
                    standing,
                    beach: self.beach.clone(),
                },
            );
        }
        self.peers.deal(&plan);
        self.begin_round(seats, Some(0), terms, names);
        self.kinds = kinds;
        let spoken: Vec<u8> = bots_here.iter().map(|(seat, ..)| *seat).collect();
        self.session.speak_for(&spoken);
        self.bots_here = bots_here;
        self.series_standing = standing;
        self.next_round = true;
        standing
    }

    /// Take up the terms of a new round: a fresh board, a lockstep back at
    /// frame zero, and whatever the table is called now.
    ///
    /// The seed is what marks this a *new* round rather than the stale
    /// `Start` a host re-answers stray greetings with, so a caller that
    /// does not change it will find nothing happens.
    #[cfg_attr(debug_assertions, track_caller)]
    pub fn begin_round(
        &mut self,
        seats: u8,
        seat: Option<u8>,
        terms: MatchTerms,
        names: [String; MAX_PLAYERS],
    ) {
        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.humans(seats));
        self.seats = seats;
        self.terms = terms;
        self.names = names;
        // The AI holds the top seats; anything else a person until the
        // round's kinds say otherwise, which its caller writes after this.
        self.kinds = std::array::from_fn(|seat| {
            match seat >= usize::from(terms.humans(seats)) && seat < usize::from(seats) {
                true => crate::sim::SeatKind::Ai,
                false => crate::sim::SeatKind::Human,
            }
        });
        self.hashes.reset();
        self.resume_echo = 0;
        // 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.new_round();
        for peer in self.peers.iter_mut() {
            peer.pick = 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();
    }

    /// Whether a `Start` off the wire is the next round rather than the
    /// stale one a host re-answers a greeting with.
    pub fn is_next_round(&self, terms: &MatchTerms) -> bool {
        terms.seed != self.terms.seed
    }

    /// 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
        // seed 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 {
            seats: self.seats,
            seat,
            terms: self.terms,
            names: wire_table(&self.names),
            kinds: crate::transport::wire_kinds(&self.kinds),
            standing: self.series_standing,
            beach: self.beach.clone(),
        }
    }

    /// 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
    }

    /// 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 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);
    }

    /// 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(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 { 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 { 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");
    }

    /// 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"
        );
    }

    /// 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![(2, 7, "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].0, 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::Kicked);
            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.next_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"]);
        host.peers.deal(&[Some(1), Some(2)]);

        // 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 Cy is the question: the hub is the only way Bo can
        // reach it at all, so what the hub relays 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 now: 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(vec![own, forged]));
            std::thread::sleep(std::time::Duration::from_millis(5));
            host.pump(PlayerAction::None, |_| {});
            for (msg, _) in sockets[1].recv_all() {
                if let NetMsg::Inputs(inputs) = msg {
                    relayed.extend(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.
    #[test]
    fn the_same_seed_is_not_a_new_round() {
        let session = OnlineSession::new(
            UdpTransport::host(0).expect("socket"),
            Lockstep::new(0, vec![0], DEFAULT_DELAY),
            2,
            terms(111),
        );
        assert!(!session.is_next_round(&terms(111)), "the round it is in");
        assert!(
            session.is_next_round(&terms(112)),
            "a beach it has not seen"
        );
    }

    /// 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.next_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.next_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");
    }
}