self_hosted_node/

host.rs

#![allow(clippy::too_many_arguments)]
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{mpsc as std_mpsc, Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};

use crate::config::{Config, DeckSelection, SelfPlayConfig};
use crate::engine_backend::{java_backend, rust_backend, EngineBackendKind, HostedGameOver};
use crate::shell_bridge::{ShellBridge, ShellCommand};
use crate::updater::{run_stale_monitor, StaleConfig};
use futures_util::stream::{SplitSink, SplitStream};
use futures_util::{SinkExt, StreamExt};
use manabot::{run_bot, BotConfig, BotResponder};
use manabrew_agent_interface::agent_impl::Responder;
use manabrew_agent_interface::ids_codec::{parse_player_slot, player_slot};
use manabrew_agent_interface::prompt::{AgentMessage, ClientToServerMessage, PromptOutput};
use manabrew_agent_interface::protocol::{
    identity_token, ClientMessage, ClientPlatform, EngineGate, EngineKind, GameFormat,
    GameOutcomeReport, IdentityProof, PlayerDeckInfo, ResumeRoomRequest, RoomInfo, RoomStatus,
    ServerMessage, StateEnvelope, PROTOCOL_VERSION,
};
use manabrew_protocol::deck_dto::Deck;
use manabrew_protocol::game::{GameViewDto, PlayerStatus};
use manabrew_protocol::transport::{DirectiveInput, StateUpdate};
use serde::Deserialize;
use serde_json::{json, Value};
use tokio::net::TcpStream;
use tokio::sync::mpsc as tokio_mpsc;
use tokio::task::JoinHandle;
use tokio::time;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::{connect_async_with_config, MaybeTlsStream, WebSocketStream};
use tracing::{debug, error, info, warn};
use uuid::Uuid;

type WsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
type WsWrite = SplitSink<WsStream, Message>;
type WsRead = SplitStream<WsStream>;

const SELF_HOSTED_NODE_PROTOCOL: &str = "self-hosted-node";
const PRIVATE_MESSAGE_MISSING_PLAYER: &str =
    "cannot forward private engine message without a mapped player";

struct RelayClient {
    username: String,
    outbound: tokio_mpsc::UnboundedSender<Message>,
    read: WsRead,
    writer: JoinHandle<()>,
}

/// Highest prompt id each local bot seat answered; the relay's later copy
/// of that answer is dropped against it.
type LocalBotAnswers = Arc<Mutex<HashMap<usize, u32>>>;

enum EngineSession {
    Manabrew {
        game_id: String,
        remote_response_txs: HashMap<usize, std_mpsc::Sender<ClientToServerMessage>>,
        engine_clock: EngineClock,
        local_bot_answers: LocalBotAnswers,
    },
    Forge {
        game_id: String,
        remote_response_txs: HashMap<usize, std_mpsc::Sender<ClientToServerMessage>>,
        cancel: Arc<AtomicBool>,
        engine_clock: EngineClock,
        local_bot_answers: LocalBotAnswers,
    },
}

impl EngineSession {
    fn game_id(&self) -> &str {
        match self {
            EngineSession::Manabrew { game_id, .. } | EngineSession::Forge { game_id, .. } => {
                game_id
            }
        }
    }

    fn engine_clock(&self) -> &EngineClock {
        match self {
            EngineSession::Manabrew { engine_clock, .. }
            | EngineSession::Forge { engine_clock, .. } => engine_clock,
        }
    }

    fn response_tx(&self, player_index: usize) -> Option<&std_mpsc::Sender<ClientToServerMessage>> {
        match self {
            EngineSession::Manabrew {
                remote_response_txs,
                ..
            }
            | EngineSession::Forge {
                remote_response_txs,
                ..
            } => remote_response_txs.get(&player_index),
        }
    }

    fn local_bot_answers(&self) -> &LocalBotAnswers {
        match self {
            EngineSession::Manabrew {
                local_bot_answers, ..
            }
            | EngineSession::Forge {
                local_bot_answers, ..
            } => local_bot_answers,
        }
    }

    /// The relay's copy of an answer already given in-process.
    fn is_local_bot_duplicate(&self, player_index: usize, prompt_id: u32) -> bool {
        prompt_id != 0
            && self
                .local_bot_answers()
                .lock()
                .ok()
                .and_then(|answers| answers.get(&player_index).copied())
                .is_some_and(|answered| answered >= prompt_id)
    }
}

struct BotState {
    handle: JoinHandle<()>,
    shutdown: Arc<tokio::sync::Notify>,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SpawnBotPayload {
    #[serde(default)]
    deck: Option<SpawnBotDeckPayload>,
    #[serde(default)]
    decks: Option<Vec<SpawnBotDeckPayload>>,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct SpawnBotDeckPayload {
    deck_name: String,
    deck: Deck,
    #[serde(default)]
    commander_name: Option<String>,
}

type SharedEngineSession = Arc<Mutex<Option<EngineSession>>>;
type SessionRegistry = Arc<Mutex<Vec<SharedEngineSession>>>;

fn registry_idle(registry: &SessionRegistry) -> bool {
    registry
        .lock()
        .map(|sessions| {
            sessions
                .iter()
                .all(|session| session.lock().map(|guard| guard.is_none()).unwrap_or(false))
        })
        .unwrap_or(false)
}

static DRAINING: AtomicBool = AtomicBool::new(false);

fn draining() -> bool {
    DRAINING.load(Ordering::Relaxed)
}

type SharedBotState = Arc<Mutex<Vec<BotState>>>;

#[derive(Clone)]
struct GameStart {
    game_id: String,
    player_order: Vec<String>,
    player_decks: Vec<PlayerDeckInfo>,
    starting_life: i32,
}

#[derive(Default)]
struct HostSnapshot {
    room_info: Option<RoomInfo>,
    resume_token: Option<String>,
    game: Option<GameStart>,
    last_state: Option<Value>,
    last_state_by_slot: HashMap<String, Value>,
    pending_prompts: HashMap<String, Value>,
    pending_end_game: Option<String>,
}

type SharedHostSnapshot = Arc<Mutex<HostSnapshot>>;

const RECONNECT_BACKOFF_SECS: [u64; 6] = [1, 2, 4, 8, 15, 30];

/// Pseudo-seat used by per-recipient backends for the public (spectator) view.
pub(crate) const OBSERVER_SEAT: usize = usize::MAX;
const CLOSE_DRAIN_TIMEOUT: Duration = Duration::from_secs(3);
const BOT_STOP_TIMEOUT: Duration = Duration::from_secs(5);

enum LoopExit {
    Cancelled,
    Disconnected,
}

pub async fn cli_entry() {
    {
        use tracing_subscriber::layer::SubscriberExt;
        use tracing_subscriber::util::SubscriberInitExt;
        tracing_subscriber::registry()
            .with(
                tracing_subscriber::EnvFilter::try_from_default_env()
                    .unwrap_or_else(|_| "self_hosted_node=info".into()),
            )
            .with(tracing_subscriber::fmt::layer())
            .with(crate::logs::layer_from_env())
            .init();
    }
    crate::metrics::init_from_env();

    if std::env::var("SELF_HOSTED_NODE_JAVA_SMOKE").is_ok() {
        let max_prompts = std::env::var("SELF_HOSTED_NODE_JAVA_SMOKE_PROMPTS")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(4);
        if let Err(error) = java_backend::run_smoke_game(max_prompts) {
            error!(%error, "java-forge smoke failed");
            std::process::exit(1);
        }
        info!(max_prompts, "java-forge smoke completed");
        return;
    }
    if std::env::var("SELF_HOSTED_NODE_GRAAL_SMOKE").is_ok() {
        if let Err(error) = java_backend::run_graal_smoke() {
            error!(%error, "graal-forge smoke failed");
            std::process::exit(1);
        }
        info!("graal-forge smoke completed");
        return;
    }
    if std::env::var("SELF_HOSTED_NODE_JAVA_CONCEDE_SMOKE").is_ok() {
        if let Err(error) = java_backend::run_concede_smoke() {
            error!(%error, "java-forge concede smoke failed");
            std::process::exit(1);
        }
        info!("java-forge concede smoke completed");
        return;
    }
    if let Ok(scenario_name) = std::env::var("SELF_HOSTED_NODE_JAVA_SCENARIO") {
        let max_prompts = std::env::var("SELF_HOSTED_NODE_JAVA_SCENARIO_PROMPTS")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(20);
        if let Err(error) = java_backend::run_scenario(&scenario_name, max_prompts) {
            error!(scenario_name, %error, "java-forge scenario failed");
            std::process::exit(1);
        }
        info!(scenario_name, max_prompts, "java-forge scenario completed");
        return;
    }
    if std::env::var("SELF_HOSTED_NODE_JAVA_SELF_PLAY").is_ok() {
        let cfg = SelfPlayConfig::from_env();
        let max_prompts = std::env::var("SELF_HOSTED_NODE_JAVA_SELF_PLAY_PROMPTS")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(2_000);
        let games = std::env::var("SELF_HOSTED_NODE_JAVA_SELF_PLAY_GAMES")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(1);
        if let Err(error) =
            java_backend::run_self_play(&cfg.seats, cfg.starting_life, cfg.seed, max_prompts, games)
        {
            error!(%error, "java-forge self-play failed");
            std::process::exit(1);
        }
        info!(
            players = cfg.seats.len(),
            max_prompts, games, "java-forge self-play completed"
        );
        return;
    }
    if let Ok(value) = std::env::var("SELF_HOSTED_NODE_JAVA_CONCURRENT_GAMES") {
        let cfg = SelfPlayConfig::from_env();
        let max_prompts = std::env::var("SELF_HOSTED_NODE_JAVA_SELF_PLAY_PROMPTS")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(2_000);
        let concurrency = value.parse().unwrap_or(2);
        if let Err(error) = java_backend::run_concurrent_self_play(
            &cfg.seats,
            cfg.starting_life,
            cfg.seed,
            max_prompts,
            concurrency,
        ) {
            error!(%error, "java-forge concurrent self-play failed");
            std::process::exit(1);
        }
        info!(
            players = cfg.seats.len(),
            concurrency, "java-forge concurrent self-play completed"
        );
        return;
    }
    if std::env::var("SELF_HOSTED_NODE_RUST_SELF_PLAY").is_ok() {
        let cfg = SelfPlayConfig::from_env();
        let max_turns = std::env::var("SELF_HOSTED_NODE_RUST_SELF_PLAY_TURNS")
            .ok()
            .and_then(|value| value.parse().ok())
            .unwrap_or(200);
        if let Err(error) =
            rust_backend::run_self_play(&cfg.seats, cfg.starting_life, cfg.seed, max_turns)
        {
            error!(%error, "rust self-play failed");
            std::process::exit(1);
        }
        info!(
            players = cfg.seats.len(),
            max_turns, "rust self-play completed"
        );
        return;
    }

    let config = Config::from_env();
    if let Err(error) = run(config).await {
        error!(%error, "room node exited");
        std::process::exit(1);
    }
}

/// Signal a hosted room to shut down: call `notify_one()` to stop it.
pub type RoomCancel = Arc<tokio::sync::Notify>;

fn ensure_engine_ready(config: &Config) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    if config.engine_enabled
        && config.backend.is_supported()
        && matches!(config.backend, EngineBackendKind::Forge)
    {
        info!("initializing forge engine backend");
        java_backend::init_engine()?;
    }
    Ok(())
}

/// Host a single relay room until it ends or `cancel` is signalled. The crate's
/// public entry point — env-free; the caller supplies a fully-built `Config`.
pub async fn host_room(
    config: Config,
    cancel: RoomCancel,
    ready: tokio::sync::oneshot::Sender<String>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    ensure_engine_ready(&config)?;
    host_one_room(config, None, cancel, Some(ready), None, None).await
}

/// The webview's half of the data plane, handed in by a desktop shell.
pub struct ShellBridgeHandle {
    pub bridge: Arc<ShellBridge>,
    pub commands: tokio_mpsc::UnboundedReceiver<ShellCommand>,
}

pub async fn host_room_bridged(
    config: Config,
    cancel: RoomCancel,
    ready: tokio::sync::oneshot::Sender<String>,
    bridge: ShellBridgeHandle,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    ensure_engine_ready(&config)?;
    host_one_room(config, None, cancel, Some(ready), None, Some(bridge)).await
}

async fn run(mut config: Config) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    ensure_engine_ready(&config)?;

    let registry: SessionRegistry = Arc::new(Mutex::new(Vec::new()));

    let slots = config.max_games.max(1);
    let single = config.room_id.is_some() || slots <= 1;
    let room_count = if single { 1 } else { slots };
    let cancels: Vec<RoomCancel> = (0..room_count)
        .map(|_| Arc::new(tokio::sync::Notify::new()))
        .collect();

    let monitor_registry = registry.clone();
    let stale_cancels = cancels.clone();
    tokio::spawn(run_stale_monitor(
        StaleConfig::from_env_and_args(),
        move || registry_idle(&monitor_registry),
        move || notify_all(&stale_cancels),
    ));

    let signal_cancels = cancels.clone();
    tokio::spawn(async move {
        wait_for_shutdown_signal().await;
        info!("shutdown signal received; closing rooms");
        notify_all(&signal_cancels);
    });

    #[cfg(unix)]
    tokio::spawn(async move {
        let mut usr1 =
            match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined1()) {
                Ok(usr1) => usr1,
                Err(error) => {
                    warn!(%error, "failed to install SIGUSR1 handler");
                    return;
                }
            };
        while usr1.recv().await.is_some() {
            DRAINING.store(true, Ordering::Relaxed);
            info!("drain signal received; refusing new games and closing each room once idle");
        }
    });

    #[cfg(all(unix, feature = "graal-forge"))]
    tokio::spawn(async move {
        let mut usr2 =
            match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::user_defined2()) {
                Ok(usr2) => usr2,
                Err(error) => {
                    warn!(%error, "failed to install SIGUSR2 handler");
                    return;
                }
            };
        while usr2.recv().await.is_some() {
            let dir = std::env::var("SELF_HOSTED_NODE_HEAP_DUMP_DIR")
                .unwrap_or_else(|_| "/tmp".to_string());
            let stamp = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|since| since.as_secs())
                .unwrap_or(0);
            let path = format!("{dir}/self-hosted-node-{stamp}.hprof");
            info!(path, "heap dump requested");
            match tokio::task::spawn_blocking({
                let path = path.clone();
                move || java_backend::dump_shared_heap(&path)
            })
            .await
            {
                Ok(Ok(())) => info!(path, "heap dump written"),
                Ok(Err(error)) => warn!(%error, "heap dump failed"),
                Err(error) => warn!(%error, "heap dump task failed"),
            }
        }
    });

    if single {
        return host_one_room(config, None, cancels[0].clone(), None, Some(registry), None).await;
    }

    config.format = GameFormat::Any;
    let hosts: Vec<(Config, String)> = (0..slots)
        .map(|slot| (config.clone(), (slot + 1).to_string()))
        .collect();

    info!(rooms = hosts.len(), "hosting multiple rooms on one node");
    let mut handles = Vec::with_capacity(hosts.len());
    for ((cfg, label), cancel) in hosts.into_iter().zip(cancels) {
        let registry = registry.clone();
        handles.push(tokio::spawn(async move {
            if let Err(error) =
                host_one_room(cfg, Some(label.clone()), cancel, None, Some(registry), None).await
            {
                error!(%error, label, "room host exited");
            }
        }));
    }
    for handle in handles {
        let _ = handle.await;
    }
    Ok(())
}

fn notify_all(cancels: &[RoomCancel]) {
    for cancel in cancels {
        cancel.notify_one();
    }
}

async fn wait_for_shutdown_signal() {
    #[cfg(unix)]
    {
        let mut term =
            match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
                Ok(term) => term,
                Err(error) => {
                    warn!(%error, "failed to install SIGTERM handler");
                    let _ = tokio::signal::ctrl_c().await;
                    return;
                }
            };
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {}
            _ = term.recv() => {}
        }
    }
    #[cfg(not(unix))]
    {
        let _ = tokio::signal::ctrl_c().await;
    }
}

async fn host_one_room(
    mut config: Config,
    label: Option<String>,
    cancel: RoomCancel,
    ready: Option<tokio::sync::oneshot::Sender<String>>,
    sessions: Option<SessionRegistry>,
    shell: Option<ShellBridgeHandle>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    if let Some(label) = &label {
        config.username = format!("{}-{label}", config.username);
        config.bot_username = format!("{}-{label}", config.bot_username);
        config.room_name = format!("{} ({label})", config.room_name);
    }

    let _rooms_hosted = crate::metrics::RoomHostedGuard::new(if label.is_some() {
        crate::metrics::PoolKind::Pod
    } else {
        crate::metrics::PoolKind::Solo
    });

    info!(
        relay_url = %config.relay_url,
        username = %config.username,
        room_name = %config.room_name,
        auto_start = config.auto_start,
        engine_enabled = config.engine_enabled,
        host_plays = config.host_plays,
        bot_enabled = config.bot_enabled,
        "starting room host"
    );

    let snapshot: SharedHostSnapshot = Arc::new(Mutex::new(HostSnapshot::default()));
    let engine_session: SharedEngineSession = Arc::new(Mutex::new(None));
    if let Some(sessions) = &sessions {
        if let Ok(mut registered) = sessions.lock() {
            registered.push(engine_session.clone());
        }
    }
    let bot_state: SharedBotState = Arc::new(Mutex::new(Vec::new()));
    let (outbound_tx, mut outbound_rx) = tokio_mpsc::unbounded_channel::<ClientMessage>();

    let mut host =
        RelayClient::connect(&config.relay_url, &config.username, &config.password).await?;
    let mut room_id = establish_room(&mut host, &config, &snapshot).await?;

    // Keep the sender alive without a shell, or `recv()` spins the select.
    let (bridge, mut bridge_rx, _idle_bridge_tx) = match shell {
        Some(handle) => (Some(handle.bridge), handle.commands, None),
        None => {
            let (tx, rx) = tokio_mpsc::unbounded_channel::<ShellCommand>();
            (None, rx, Some(tx))
        }
    };
    if let Some(bridge) = &bridge {
        let reprime_snapshot = snapshot.clone();
        let reprime_tx = outbound_tx.clone();
        bridge.set_on_fallback(move |username| {
            reprime_relay_cache(&reprime_snapshot, &reprime_tx, username);
        });
    }

    if bridge.is_some() {
        announce_webrtc_only(&config, &outbound_tx);
    }

    if let Some(ready) = ready {
        let _ = ready.send(room_id.clone());
    }

    if config.bot_enabled {
        spawn_bots(
            &config,
            std::slice::from_ref(&config.bot_deck),
            &room_id,
            &bot_state,
        );
    }

    loop {
        let exit = run_client_loop(
            &mut host,
            &config,
            &room_id,
            &engine_session,
            &snapshot,
            &bot_state,
            &outbound_tx,
            &mut outbound_rx,
            &cancel,
            bridge.as_ref(),
            &mut bridge_rx,
        )
        .await;
        if matches!(exit, LoopExit::Cancelled) {
            stop_bots(&bot_state);
            host.close().await;
            return Ok(());
        }

        crate::metrics::record_relay_reconnect();
        let mut attempt: usize = 0;
        host = loop {
            let delay = RECONNECT_BACKOFF_SECS[attempt.min(RECONNECT_BACKOFF_SECS.len() - 1)];
            tokio::select! {
                _ = cancel.notified() => {
                    info!(username = %config.username, "room host cancelled while reconnecting");
                    cancel_engine(&engine_session);
                    stop_bots(&bot_state);
                    return Ok(());
                }
                _ = time::sleep(Duration::from_secs(delay)) => {}
            }
            attempt += 1;
            let mut client =
                match RelayClient::connect(&config.relay_url, &config.username, &config.password)
                    .await
                {
                    Ok(client) => client,
                    Err(error) => {
                        warn!(%error, attempt, "relay reconnect failed");
                        continue;
                    }
                };
            match reestablish_room(&mut client, &config, &engine_session, &snapshot).await {
                Ok(new_room_id) => {
                    if new_room_id != room_id {
                        stop_bots(&bot_state);
                        if config.bot_enabled {
                            spawn_bots(
                                &config,
                                std::slice::from_ref(&config.bot_deck),
                                &new_room_id,
                                &bot_state,
                            );
                        }
                        room_id = new_room_id;
                    }
                    break client;
                }
                Err(error) => {
                    warn!(%error, attempt, "failed to re-establish room; retrying");
                }
            }
        };
        if let Some(bridge) = &bridge {
            bridge.clear_game();
            announce_webrtc_only(&config, &outbound_tx);
        }
        info!(username = %config.username, room_id, "relay connection re-established");
    }
}

/// Re-primes the relay's replay cache for a seat leaving the plane.
fn reprime_relay_cache(
    snapshot: &SharedHostSnapshot,
    outbound_tx: &tokio_mpsc::UnboundedSender<ClientMessage>,
    username: &str,
) {
    let Some(index) = seat_index_of(snapshot, username) else {
        return;
    };
    let slot = player_slot(index);
    let Ok(snap) = snapshot.lock() else {
        return;
    };
    let state = snap.last_state_by_slot.get(&slot).cloned();
    let prompt = snap.pending_prompts.get(&slot).cloned();
    drop(snap);

    // Stored before `patch_against_last` runs, so these are full states.
    for state in [state, prompt].into_iter().flatten() {
        let _ = outbound_tx.send(ClientMessage::BroadcastState {
            state,
            target_player: Some(username.to_string()),
        });
    }
    info!(
        username,
        "re-primed the relay cache for a seat that fell back"
    );
}

fn announce_webrtc_only(config: &Config, outbound_tx: &tokio_mpsc::UnboundedSender<ClientMessage>) {
    let endpoint = manabrew_relay_protocol::TransportEndpoint {
        endpoint_id: format!("webrtc:{}", config.username),
        relay_url: None,
        direct_addrs: Vec::new(),
        kinds: vec![manabrew_relay_protocol::TRANSPORT_KIND_WEBRTC.to_string()],
    };
    info!(endpoint_id = %endpoint.endpoint_id, "announcing a webrtc-only endpoint");
    let _ = outbound_tx.send(ClientMessage::AnnounceTransport {
        endpoint: Some(endpoint),
    });
}

async fn establish_room(
    client: &mut RelayClient,
    config: &Config,
    snapshot: &SharedHostSnapshot,
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
    let room_id = if let Some(room_id) = &config.room_id {
        client
            .send(&ClientMessage::JoinRoom {
                room_id: room_id.clone(),
                observe: !config.host_plays,
                as_bot: false,
                password: config.room_password.clone(),
            })
            .await?;
        info!(room_id, "joining configured room");
        room_id.clone()
    } else {
        client
            .send(&ClientMessage::CreateRoom {
                room_name: config.room_name.clone(),
                max_players: config.max_players,
                format: config.format.clone(),
                protocol_version: PROTOCOL_VERSION,
                hosted: !config.host_plays,
                engine: engine_kind(config),
                draft_config: None,
                sealed_config: None,
                official_key: config.official_key.clone(),
                password: config.room_password.clone(),
                reconnect_timeout_s: config.reconnect_timeout_s,
                table_style: config.table_style.clone(),
            })
            .await?;
        info!(room_name = %config.room_name, "creating room");
        wait_for_host_room(client, config, snapshot).await?
    };

    if config.host_plays {
        seat_client(client, &config.host_deck).await?;
    } else {
        info!(
            username = %config.username,
            "hosting room without occupying a player seat"
        );
    }
    Ok(room_id)
}

async fn reestablish_room(
    client: &mut RelayClient,
    config: &Config,
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
    let resume = {
        let snap = snapshot.lock().map_err(|error| error.to_string())?;
        match (&snap.resume_token, &snap.game, &snap.room_info) {
            (Some(token), Some(game), Some(room_info)) => {
                Some(resume_room_request(config, token, game, room_info))
            }
            _ => None,
        }
    };
    let game_active = engine_session
        .lock()
        .map(|guard| guard.is_some())
        .unwrap_or(false);

    if let (Some(request), true) = (resume, game_active) {
        let room_id = request.room_id.clone();
        client.send(&ClientMessage::ResumeRoom(request)).await?;
        match wait_for_room_resumed(client, engine_session, snapshot).await? {
            Some(room) => {
                let (last_state, seat_states, prompts, player_order, pending_end_game) = {
                    let mut snap = snapshot.lock().map_err(|error| error.to_string())?;
                    snap.room_info = Some(room);
                    (
                        snap.last_state.clone(),
                        snap.last_state_by_slot.clone(),
                        snap.pending_prompts.clone(),
                        snap.game
                            .as_ref()
                            .map(|game| game.player_order.clone())
                            .unwrap_or_default(),
                        snap.pending_end_game.clone(),
                    )
                };
                if let Some(state) = last_state {
                    client
                        .send(&ClientMessage::BroadcastState {
                            state,
                            target_player: None,
                        })
                        .await?;
                }
                for (slot, state) in seat_states {
                    let Some(target_player) =
                        parse_player_slot(&slot).and_then(|index| player_order.get(index).cloned())
                    else {
                        warn!(slot, "cannot replay private state without a mapped player");
                        continue;
                    };
                    client
                        .send(&ClientMessage::BroadcastState {
                            state,
                            target_player: Some(target_player),
                        })
                        .await?;
                }
                for (slot, prompt) in prompts {
                    let Some(target_player) =
                        parse_player_slot(&slot).and_then(|index| player_order.get(index).cloned())
                    else {
                        warn!(slot, "cannot replay prompt without a mapped player");
                        continue;
                    };
                    client
                        .send(&ClientMessage::BroadcastState {
                            state: prompt,
                            target_player: Some(target_player),
                        })
                        .await?;
                }
                if let Some(game_id) = pending_end_game {
                    client.send(&ClientMessage::EndGame { game_id }).await?;
                }
                info!(room_id, "room resumed; snapshot re-broadcast");
                return Ok(room_id);
            }
            None => {
                warn!(room_id, "room resume rejected; abandoning hosted game");
                abort_engine_session(engine_session);
            }
        }
    }

    establish_room(client, config, snapshot).await
}

fn engine_kind(config: &Config) -> EngineKind {
    if matches!(config.backend, EngineBackendKind::Forge) {
        EngineKind::Forge
    } else {
        EngineKind::Manabrew
    }
}

fn resume_room_request(
    config: &Config,
    token: &str,
    game: &GameStart,
    room_info: &RoomInfo,
) -> ResumeRoomRequest {
    ResumeRoomRequest {
        room_id: room_info.room_id.clone(),
        resume_token: token.to_string(),
        room_name: room_info.room_name.clone(),
        max_players: room_info.max_players,
        format: room_info.format.clone(),
        hosted: !config.host_plays,
        engine: engine_kind(config),
        official_key: config.official_key.clone(),
        password: config.room_password.clone(),
        reconnect_timeout_s: Some(room_info.reconnect_timeout_s),
        table_style: room_info.table_style.clone(),
        draft_config: room_info.draft_config.clone(),
        sealed_config: room_info.sealed_config.clone(),
        player_order: game.player_order.clone(),
        player_decks: game.player_decks.clone(),
        starting_life: game.starting_life,
        bot_players: room_info
            .players
            .iter()
            .filter(|player| player.is_bot)
            .map(|player| player.username.clone())
            .collect(),
        game_id: game.game_id.clone(),
    }
}

async fn wait_for_room_resumed(
    client: &mut RelayClient,
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
) -> Result<Option<RoomInfo>, Box<dyn std::error::Error + Send + Sync>> {
    loop {
        match client.recv().await? {
            Some(ServerMessage::RoomResumed { room }) => {
                if room.status == RoomStatus::Lobby {
                    abort_stale_engine_session(
                        engine_session,
                        snapshot,
                        "resumed room is no longer in game",
                    );
                } else {
                    concede_missing_active_seats(engine_session, snapshot, &room);
                }
                return Ok(Some(room));
            }
            Some(ServerMessage::RoomUpdate { room }) => {
                if room.status == RoomStatus::Lobby {
                    abort_stale_engine_session(
                        engine_session,
                        snapshot,
                        "room reset while reconnecting",
                    );
                } else {
                    concede_missing_active_seats(engine_session, snapshot, &room);
                }
                if let Ok(mut snap) = snapshot.lock() {
                    snap.room_info = Some(room);
                }
            }
            Some(ServerMessage::StateUpdate { from_player, state }) => {
                let Ok(envelope) = serde_json::from_value::<StateEnvelope>(state.clone()) else {
                    continue;
                };
                match envelope {
                    StateEnvelope::Response { .. } => {
                        route_remote_response(engine_session, snapshot, &from_player, &state);
                    }
                    StateEnvelope::Directive {
                        from_player: claimed_slot,
                        directive,
                    } => route_remote_directive(
                        engine_session,
                        snapshot,
                        &from_player,
                        &claimed_slot,
                        &directive,
                    ),
                    _ => {}
                }
            }
            Some(ServerMessage::Error { code, message }) => {
                warn!(code, message, "resume rejected by relay");
                return Ok(None);
            }
            Some(other) => debug!(?other, "ignored message while waiting for room resume"),
            None => return Err("relay closed while waiting for room resume".into()),
        }
    }
}

fn cancel_engine(engine_session: &SharedEngineSession) {
    if let Ok(guard) = engine_session.lock() {
        if let Some(EngineSession::Forge { cancel, .. }) = guard.as_ref() {
            cancel.store(true, Ordering::Relaxed);
        }
    }
}

fn abort_engine_session(engine_session: &SharedEngineSession) {
    cancel_engine(engine_session);
    // Dropping the response channels makes the Manabrew backend's transports
    // observe a disconnect and concede, which ends the engine thread.
    clear_engine_session(engine_session);
}

fn abort_stale_engine_session(
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    reason: &str,
) {
    let stale_game_id = engine_session
        .lock()
        .ok()
        .and_then(|guard| guard.as_ref().map(|s| s.game_id().to_string()));
    let Some(stale_game_id) = stale_game_id else {
        return;
    };
    warn!(game_id = %stale_game_id, reason, "aborting stale engine session");
    abort_engine_session(engine_session);
    clear_game_snapshot(snapshot, &stale_game_id);
}

fn clear_game_snapshot(snapshot: &SharedHostSnapshot, game_id: &str) {
    if let Ok(mut snap) = snapshot.lock() {
        if snap
            .game
            .as_ref()
            .is_some_and(|game| game.game_id == game_id)
        {
            snap.game = None;
            snap.last_state = None;
            snap.last_state_by_slot.clear();
            snap.pending_prompts.clear();
            snap.pending_end_game = None;
        }
    }
}

/// A game must have at least one human seat. Seats survive disconnects and
/// concessions (reconnect grace / spectating), but an explicit leave removes
/// the seat for good — once every human seat is gone the game has no
/// stakeholders and ends now, instead of bots playing it out until the
/// relay's humanless sweep reclaims the room minutes later.
fn end_game_without_humans(
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    outbound_tx: &tokio_mpsc::UnboundedSender<ClientMessage>,
    room: &RoomInfo,
) {
    if room.players.iter().any(|seat| !seat.is_bot) {
        return;
    }
    let game_id = engine_session
        .lock()
        .ok()
        .and_then(|guard| guard.as_ref().map(|s| s.game_id().to_string()));
    let Some(game_id) = game_id else {
        return;
    };
    info!(game_id, "no human seats remain; ending hosted game");
    abort_engine_session(engine_session);
    clear_game_snapshot(snapshot, &game_id);
    let _ = outbound_tx.send(ClientMessage::EndGame { game_id });
}

fn spawn_bots(config: &Config, decks: &[DeckSelection], room_id: &str, bot_state: &SharedBotState) {
    stop_bots(bot_state);
    let mut guard = match bot_state.lock() {
        Ok(guard) => guard,
        Err(error) => {
            warn!(%error, "bot state lock poisoned");
            return;
        }
    };
    let open_seats = config.max_players.saturating_sub(1) as usize;
    for (index, deck) in decks.iter().take(open_seats).enumerate() {
        let username = if index == 0 {
            config.bot_username.clone()
        } else {
            format!("{} {}", config.bot_username, index + 1)
        };
        let relay_url = config.relay_url.clone();
        let bot_config = BotConfig {
            username,
            password: config.password.clone(),
            room_id: room_id.to_string(),
            room_password: config.room_password.clone(),
            deck_name: deck.name.clone(),
            deck: deck.deck.clone(),
            commander_name: deck.commander_name.clone(),
            agent: config.bot_agent,
            answer_delay_ms: None,
        };
        let shutdown = Arc::new(tokio::sync::Notify::new());
        let bot_shutdown = shutdown.clone();
        let handle = tokio::spawn(async move {
            if let Err(error) = run_bot(relay_url, bot_config, bot_shutdown).await {
                error!(%error, "bot task exited");
            }
        });
        guard.push(BotState { handle, shutdown });
    }
}

fn stop_bots(bot_state: &SharedBotState) {
    match bot_state.lock() {
        Ok(mut state) => {
            for bot in state.drain(..) {
                bot.shutdown.notify_one();
                let mut handle = bot.handle;
                tokio::spawn(async move {
                    if time::timeout(BOT_STOP_TIMEOUT, &mut handle).await.is_err() {
                        handle.abort();
                    }
                });
            }
        }
        Err(error) => warn!(%error, "bot state lock poisoned"),
    }
}

async fn wait_for_host_room(
    client: &mut RelayClient,
    config: &Config,
    snapshot: &SharedHostSnapshot,
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
    loop {
        match client.recv().await? {
            Some(ServerMessage::RoomCreated {
                room_id,
                room_name,
                room,
                resume_token,
            }) => {
                info!(room_id, room_name, "room created");
                if let Ok(mut snap) = snapshot.lock() {
                    snap.resume_token = resume_token;
                    snap.room_info = Some(room);
                }
                return Ok(room_id);
            }
            Some(ServerMessage::RoomUpdate { room }) if room.host == config.username => {
                info!(room_id = %room.room_id, room_name = %room.room_name, "host room update");
                return Ok(room.room_id);
            }
            Some(ServerMessage::AuthResult { success, error, .. }) => {
                if !success {
                    return Err(format!("authentication failed: {error:?}").into());
                }
            }
            Some(ServerMessage::Error { code, message }) => {
                return Err(format!("server error while creating room: {code}: {message}").into());
            }
            Some(other) => debug!(?other, "ignored message while waiting for room creation"),
            None => return Err("relay closed while waiting for room creation".into()),
        }
    }
}

async fn seat_client(
    client: &mut RelayClient,
    deck: &DeckSelection,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    client
        .send(&ClientMessage::SetDeckSelection {
            deck_name: deck.name.clone(),
            deck: deck.deck.clone(),
            published_deck_id: None,
            commander_name: deck.commander_name.clone(),
            avatar_url: None,
        })
        .await?;
    client
        .send(&ClientMessage::SetReady { ready: true })
        .await?;
    info!(
        username = %client.username,
        deck = %deck.name,
        cards = deck.deck.cards.len(),
        "selected deck and marked ready"
    );
    Ok(())
}

async fn run_client_loop(
    client: &mut RelayClient,
    config: &Config,
    room_id: &str,
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    bot_state: &SharedBotState,
    outbound_tx: &tokio_mpsc::UnboundedSender<ClientMessage>,
    outbound_rx: &mut tokio_mpsc::UnboundedReceiver<ClientMessage>,
    cancel: &RoomCancel,
    bridge: Option<&Arc<ShellBridge>>,
    bridge_rx: &mut tokio_mpsc::UnboundedReceiver<ShellCommand>,
) -> LoopExit {
    let mut heartbeat = time::interval(Duration::from_secs(30));
    heartbeat.set_missed_tick_behavior(time::MissedTickBehavior::Delay);
    let mut drain_tick = time::interval(Duration::from_secs(5));
    drain_tick.set_missed_tick_behavior(time::MissedTickBehavior::Delay);
    let mut bot_usernames: HashSet<String> = HashSet::new();
    let mut forge_ai = config.forge_ai;

    loop {
        tokio::select! {
            _ = cancel.notified() => {
                info!(username = %client.username, "room host cancelled; shutting down");
                cancel_engine(engine_session);
                return LoopExit::Cancelled;
            }
            _ = drain_tick.tick() => {
                let idle = engine_session
                    .lock()
                    .map(|guard| guard.is_none())
                    .unwrap_or(false);
                if draining() && idle {
                    info!(username = %client.username, "draining and no active game; closing room");
                    return LoopExit::Cancelled;
                }
            }
            _ = heartbeat.tick() => {
                if let Err(error) = client.broadcast_room_message(SELF_HOSTED_NODE_PROTOCOL, json!({
                    "type": "heartbeat",
                    "node": client.username,
                    "capacity": config.max_games,
                })).await {
                    warn!(%error, username = %client.username, "relay send failed");
                    return LoopExit::Disconnected;
                }
            }
            outbound = outbound_rx.recv() => {
                let Some(outbound) = outbound else {
                    warn!(username = %client.username, "outbound channel closed");
                    return LoopExit::Cancelled;
                };
                if let ClientMessage::BroadcastState { state, target_player: Some(target) } = &outbound {
                    if bridge.is_some_and(|shell| shell.try_send(target, state)) {
                        continue;
                    }
                }
                if let Err(error) = client.send(&outbound).await {
                    warn!(%error, username = %client.username, "relay send failed");
                    return LoopExit::Disconnected;
                }
            }
            command = bridge_rx.recv() => {
                let Some(command) = command else {
                    warn!(username = %client.username, "shell bridge closed");
                    return LoopExit::Cancelled;
                };
                let Some(shell) = bridge else { continue };
                match command {
                    ShellCommand::Serving { seats } => shell.set_serving(seats),
                    ShellCommand::Signal { to, payload } => {
                        if let Err(error) = client
                            .send(&ClientMessage::SignalPeer { to, payload })
                            .await
                        {
                            warn!(%error, username = %client.username, "relay send failed");
                            return LoopExit::Disconnected;
                        }
                    }
                    ShellCommand::SeatEnvelope { from, envelope } => {
                        // `from` came from a relay-attested roster.
                        route_seat_envelope(engine_session, snapshot, &from, &envelope);
                    }
                }
            }
            message = client.recv() => {
                let message = match message {
                    Ok(Some(message)) => message,
                    Ok(None) => {
                        warn!(username = %client.username, "relay websocket closed");
                        return LoopExit::Disconnected;
                    }
                    Err(error) => {
                        warn!(%error, username = %client.username, "relay receive failed");
                        return LoopExit::Disconnected;
                    }
                };
                if let Err(error) = handle_server_message(
                    client,
                    config,
                    room_id,
                    engine_session,
                    snapshot,
                    bot_state,
                    outbound_tx,
                    &mut bot_usernames,
                    &mut forge_ai,
                    bridge,
                    message,
                ).await {
                    warn!(%error, username = %client.username, "relay send failed");
                    return LoopExit::Disconnected;
                }
            }
        }
    }
}

async fn handle_server_message(
    client: &mut RelayClient,
    config: &Config,
    room_id: &str,
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    bot_state: &SharedBotState,
    outbound_tx: &tokio_mpsc::UnboundedSender<ClientMessage>,
    bot_usernames: &mut HashSet<String>,
    forge_ai: &mut bool,
    bridge: Option<&Arc<ShellBridge>>,
    message: ServerMessage,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    match message {
        ServerMessage::RoomUpdate { room } => {
            if let Ok(mut snap) = snapshot.lock() {
                snap.room_info = Some(room.clone());
            }
            *bot_usernames = room
                .players
                .iter()
                .filter(|p| p.is_bot)
                .map(|p| p.username.clone())
                .collect();
            log_room_update(&client.username, &room);
            if room.status == RoomStatus::Lobby {
                abort_stale_engine_session(engine_session, snapshot, "room reset to lobby");
            } else {
                end_game_without_humans(engine_session, snapshot, outbound_tx, &room);
            }
            maybe_auto_start_room(client, config, &room).await?;
        }
        ServerMessage::StateUpdate { from_player, state } => {
            if let Some(shell) = bridge {
                shell.note_relay_message(&from_player);
            }
            handle_state_update(
                client,
                config,
                room_id,
                engine_session,
                snapshot,
                bot_state,
                forge_ai,
                from_player,
                state,
            )
            .await?;
        }
        ServerMessage::PeerSignal { from, payload } => match bridge {
            Some(shell) => shell.forward_signal(&from, payload),
            None => debug!(from, "signalling arrived with no shell to hand it to"),
        },
        ServerMessage::ReadyStateChanged { username, ready } => {
            info!(username, ready, observer = %client.username, "ready changed");
        }
        ServerMessage::PlayerJoined { username, room_id } => {
            info!(username, room_id, observer = %client.username, "player joined");
        }
        ServerMessage::PlayerLeft { username, room_id } => {
            info!(username, room_id, observer = %client.username, "player left");
            let leaver_still_needed =
                active_player_usernames(snapshot).is_some_and(|active| active.contains(&username));
            if leaver_still_needed {
                if let Some(index) = seat_index_of(snapshot, &username) {
                    info!(
                        username,
                        index, "player abandoned mid-game; conceding their seat"
                    );
                    concede_seat(engine_session, index);
                }
            }
        }
        ServerMessage::GameStarted {
            room_id,
            game_id,
            player_order,
            player_decks,
            starting_life,
        } => {
            info!(room_id, game_id, ?player_order, observer = %client.username, "game started");
            abort_stale_engine_session(engine_session, snapshot, "relay started a new game");
            if let Ok(mut snap) = snapshot.lock() {
                snap.game = Some(GameStart {
                    game_id: game_id.clone(),
                    player_order: player_order.clone(),
                    player_decks: player_decks.clone(),
                    starting_life,
                });
                snap.last_state = None;
                snap.last_state_by_slot.clear();
                snap.pending_prompts.clear();
                snap.pending_end_game = None;
            }
            let mut left_the_relay = Vec::new();
            if let Some(shell) = bridge {
                let seats = shell.freeze_for_game(&player_order, &[]);
                if !seats.is_empty() {
                    info!(
                        game_id,
                        ?seats,
                        "seats playing this game through the webview"
                    );
                    left_the_relay.extend(shell.transport_report(&seats));
                }
            }
            if !left_the_relay.is_empty() {
                let _ = outbound_tx.send(ClientMessage::ReportTransport {
                    game_id: game_id.clone(),
                    seats: left_the_relay,
                });
            }
            maybe_start_hosted_engine(
                config,
                engine_session,
                snapshot,
                outbound_tx,
                game_id,
                player_order,
                player_decks,
                starting_life,
                bot_usernames,
                *forge_ai,
            );
        }
        ServerMessage::RoomTransport { members, .. } => {
            if let Some(shell) = bridge {
                // Empty until every seat has opted in.
                shell.set_roster(members.iter().map(|member| member.username.clone()));
            }
        }
        ServerMessage::ServerShuttingDown { reconnect_in_s } => {
            info!(reconnect_in_s, observer = %client.username, "relay is restarting");
        }
        ServerMessage::Error { code, message } => {
            warn!(code, message, observer = %client.username, "server error");
        }
        other => {
            debug!(?other, observer = %client.username, "server message");
        }
    }

    Ok(())
}

async fn handle_state_update(
    client: &mut RelayClient,
    config: &Config,
    room_id: &str,
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    bot_state: &SharedBotState,
    forge_ai: &mut bool,
    from_player: String,
    state: Value,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let Ok(envelope) = serde_json::from_value::<StateEnvelope>(state.clone()) else {
        debug!(from_player, state = %state, observer = %client.username, "state update");
        return Ok(());
    };

    match envelope {
        StateEnvelope::Response { .. } | StateEnvelope::Directive { .. } => {
            route_seat_envelope(engine_session, snapshot, &from_player, &state);
            Ok(())
        }
        StateEnvelope::RoomRelay {
            protocol,
            from_player: _,
            room_id: requested_room_id,
            payload,
            ..
        } => {
            info!(
                from_player,
                protocol,
                payload = %payload,
                observer = %client.username,
                "room relay message"
            );
            if protocol != SELF_HOSTED_NODE_PROTOCOL {
                return Ok(());
            }
            let payload_type = payload.get("type").and_then(Value::as_str);
            match payload_type {
                Some("removeBot") => {
                    info!(observer = %client.username, "received removeBot request");
                    stop_bots(bot_state);
                }
                Some("spawnBot") => {
                    let effective_room_id = requested_room_id.as_deref().unwrap_or(room_id);
                    if effective_room_id == room_id {
                        let requested_forge_ai = payload
                            .get("forgeAi")
                            .and_then(Value::as_bool)
                            .unwrap_or(false);
                        *forge_ai = config.forge_ai || requested_forge_ai;
                        info!(observer = %client.username, room_id, forge_ai = *forge_ai, "received spawnBot request");
                        let bot_decks = bot_decks_from_payload(config, &payload);
                        spawn_bots(config, &bot_decks, room_id, bot_state);
                    } else {
                        debug!(
                            observer = %client.username,
                            current_room_id = room_id,
                            requested_room_id = effective_room_id,
                            "ignoring spawnBot request for another room"
                        );
                    }
                }
                _ => {}
            }
            Ok(())
        }
        _ => {
            debug!(from_player, state = %state, observer = %client.username, "state update");
            Ok(())
        }
    }
}

fn bot_decks_from_payload(config: &Config, payload: &Value) -> Vec<DeckSelection> {
    let parsed = serde_json::from_value::<SpawnBotPayload>(payload.clone()).ok();
    let requested = match parsed {
        Some(SpawnBotPayload {
            decks: Some(decks), ..
        }) if !decks.is_empty() => decks,
        Some(SpawnBotPayload {
            deck: Some(deck), ..
        }) => vec![deck],
        _ => return vec![config.bot_deck.clone()],
    };
    requested
        .into_iter()
        .map(deck_selection_from_payload)
        .collect()
}

fn deck_selection_from_payload(deck: SpawnBotDeckPayload) -> DeckSelection {
    info!(
        deck = %deck.deck_name,
        cards = deck.deck.cards.len(),
        commander = ?deck.commander_name,
        "using requested bot deck"
    );
    let commander_name = deck.commander_name.or_else(|| {
        deck.deck
            .commanders
            .as_ref()
            .and_then(|commanders| commanders.first())
            .map(|commander| commander.identity.name.clone())
    });
    DeckSelection {
        name: deck.deck_name,
        deck: deck.deck,
        commander_name,
    }
}

async fn maybe_auto_start_room(
    client: &mut RelayClient,
    config: &Config,
    room: &RoomInfo,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    if !config.auto_start || draining() {
        return Ok(());
    }
    if config.format == GameFormat::Any {
        return Ok(());
    }
    if room.host != config.username
        || room.status != RoomStatus::Lobby
        || room.players.len() < config.max_players as usize
    {
        return Ok(());
    }
    let all_ready = room
        .players
        .iter()
        .all(|player| player.connected && player.ready && player.selected_deck_name.is_some());
    if all_ready {
        info!(
            room_id = %room.room_id,
            players = room.players.len(),
            "all players ready; auto-starting hosted game"
        );
        client
            .send(&ClientMessage::StartGame { format: None })
            .await?;
    }
    Ok(())
}

const ENGINE_STACK_BYTES: usize = 64 * 1024 * 1024;

fn spawn_engine_thread<F: FnOnce() + Send + 'static>(body: F) {
    if let Err(error) = thread::Builder::new()
        .name("hosted-engine".to_string())
        .stack_size(ENGINE_STACK_BYTES)
        .spawn(body)
    {
        error!(%error, "failed to spawn hosted engine thread");
    }
}

// Any stays commander-capable: hosted "Any" rooms resolve their real format
// at StartGame, and dropping commanders there would break commander games.
fn is_commander_variant(format: GameFormat) -> bool {
    match format {
        GameFormat::Any | GameFormat::Commander | GameFormat::Brawl | GameFormat::Oathbreaker => {
            true
        }
        GameFormat::Standard
        | GameFormat::Pioneer
        | GameFormat::Modern
        | GameFormat::Legacy
        | GameFormat::Vintage
        | GameFormat::Pauper
        | GameFormat::Premodern
        | GameFormat::Draft
        | GameFormat::Sealed => false,
    }
}

// Forge GameType name for the java backend; empty = let the adapter infer
// commander-ness the pre-variant way (Any rooms carry no concrete format here).
fn java_game_variant(format: GameFormat) -> &'static str {
    match format {
        GameFormat::Any => "",
        GameFormat::Commander => "Commander",
        GameFormat::Brawl => "Brawl",
        GameFormat::Oathbreaker => "Oathbreaker",
        GameFormat::Standard
        | GameFormat::Pioneer
        | GameFormat::Modern
        | GameFormat::Legacy
        | GameFormat::Vintage
        | GameFormat::Pauper
        | GameFormat::Premodern
        | GameFormat::Draft
        | GameFormat::Sealed => "Constructed",
    }
}

fn maybe_start_hosted_engine(
    config: &Config,
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    outbound_tx: &tokio_mpsc::UnboundedSender<ClientMessage>,
    game_id: String,
    player_order: Vec<String>,
    player_decks: Vec<PlayerDeckInfo>,
    starting_life: i32,
    bot_usernames: &HashSet<String>,
    forge_ai: bool,
) {
    if !config.engine_enabled {
        debug!("hosted engine disabled for this node");
        return;
    }
    let backend = config.backend;
    if !backend.is_supported() {
        if let Err(error) = java_backend::JavaRuntimeConfig::from_env().validate() {
            warn!(
                backend = backend.label(),
                %error,
                "hosted engine backend runtime is not ready"
            );
            return;
        }
        warn!(
            backend = backend.label(),
            message = java_backend::unsupported_message(),
            "hosted engine backend is not implemented yet"
        );
        return;
    }

    let local_player_index = if config.host_plays {
        match player_order
            .iter()
            .position(|name| name == &config.username)
        {
            Some(index) => Some(index),
            None => {
                warn!(
                    username = %config.username,
                    ?player_order,
                    "room node is configured as a player but is not in player order; not starting engine"
                );
                return;
            }
        }
    } else {
        None
    };

    let mut guard = match engine_session.lock() {
        Ok(guard) => guard,
        Err(error) => {
            warn!(%error, "engine session lock poisoned");
            return;
        }
    };
    if let Some(session) = guard.as_ref() {
        warn!(
            game_id,
            stale_game_id = session.game_id(),
            "engine session still present at game start; not starting engine"
        );
        return;
    }

    let session_handle = engine_session.clone();
    let num_players = player_order.len();
    if num_players < 2 {
        warn!(num_players, "not enough players to start hosted engine");
        return;
    }

    let mut deck_map: HashMap<String, PlayerDeckInfo> = player_decks
        .into_iter()
        .map(|deck| (deck.username.clone(), deck))
        .collect();
    let mut ordered_decks = Vec::with_capacity(num_players);
    let mut commander_names = Vec::with_capacity(num_players);
    let mut ai_player_indices = Vec::new();
    for (index, username) in player_order.iter().enumerate() {
        let Some(deck) = deck_map.remove(username) else {
            warn!(username, "missing deck for player; not starting engine");
            return;
        };
        if forge_ai && bot_usernames.contains(username) {
            ai_player_indices.push(index);
        }
        ordered_decks.push(deck.deck);
        commander_names.push(deck.commander_name);
    }

    let player_names = player_order;
    let room_format = snapshot
        .lock()
        .ok()
        .and_then(|snap| snap.room_info.as_ref().map(|room| room.format.clone()))
        .unwrap_or(GameFormat::Any);
    let commander_variant = is_commander_variant(room_format.clone());
    let game_variant = java_game_variant(room_format).to_string();

    match backend {
        EngineBackendKind::Manabrew => {
            let (remote_prompt_tx, remote_prompt_rx) = std_mpsc::channel::<(usize, AgentMessage)>();
            let mut remote_response_txs = HashMap::new();
            let mut remote_response_rxs = Vec::new();
            for i in 0..num_players {
                if Some(i) == local_player_index {
                    continue;
                }
                let (response_tx, response_rx) = std_mpsc::channel::<ClientToServerMessage>();
                remote_response_txs.insert(i, response_tx);
                remote_response_rxs.push((i, response_rx));
            }
            let engine_clock = EngineClock::default();
            *guard = Some(EngineSession::Manabrew {
                game_id: game_id.clone(),
                remote_response_txs,
                engine_clock: engine_clock.clone(),
                local_bot_answers: LocalBotAnswers::default(),
            });
            drop(guard);

            spawn_remote_prompt_forwarder(
                outbound_tx.clone(),
                snapshot.clone(),
                engine_session.clone(),
                game_id.clone(),
                remote_prompt_rx,
                Some(player_names.clone()),
                config.state_delta,
                engine_clock,
                local_bot_seats(config, &player_names, bot_usernames),
            );
            let (game_over_tx, game_over_rx) = std_mpsc::channel::<HostedGameOver>();
            spawn_game_over_forwarder(
                outbound_tx.clone(),
                game_over_rx,
                engine_session.clone(),
                snapshot.clone(),
                game_id.clone(),
                Some(player_names.clone()),
            );
            let outbound_tx = outbound_tx.clone();
            let snapshot = snapshot.clone();
            spawn_engine_thread(move || {
                info!(
                    game_id,
                    backend = backend.label(),
                    players = num_players,
                    local_player_index,
                    "starting hosted engine thread"
                );
                crate::metrics::record_engine_session_started();
                let started = Instant::now();
                let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    rust_backend::run_hosted_engine_game(
                        game_id.clone(),
                        player_names,
                        ordered_decks,
                        commander_names,
                        local_player_index,
                        starting_life,
                        remote_prompt_tx,
                        remote_response_rxs,
                        game_over_tx,
                    )
                }));
                finish_hosted_engine(
                    result,
                    &game_id,
                    num_players,
                    started,
                    &outbound_tx,
                    &session_handle,
                    &snapshot,
                );
            });
        }
        EngineBackendKind::Forge => {
            let (remote_prompt_tx, remote_prompt_rx) = std_mpsc::channel::<(usize, AgentMessage)>();
            let mut remote_response_txs = HashMap::new();
            let mut remote_response_rxs = Vec::new();
            for i in 0..num_players {
                if Some(i) == local_player_index {
                    continue;
                }
                let (response_tx, response_rx) = std_mpsc::channel::<ClientToServerMessage>();
                remote_response_txs.insert(i, response_tx);
                remote_response_rxs.push((i, response_rx));
            }
            let cancel = Arc::new(AtomicBool::new(false));
            let engine_clock = EngineClock::default();
            *guard = Some(EngineSession::Forge {
                game_id: game_id.clone(),
                remote_response_txs,
                cancel: cancel.clone(),
                engine_clock: engine_clock.clone(),
                local_bot_answers: LocalBotAnswers::default(),
            });
            drop(guard);

            spawn_remote_prompt_forwarder(
                outbound_tx.clone(),
                snapshot.clone(),
                engine_session.clone(),
                game_id.clone(),
                remote_prompt_rx,
                Some(player_names.clone()),
                config.state_delta,
                engine_clock,
                local_bot_seats(config, &player_names, bot_usernames),
            );
            let (game_over_tx, game_over_rx) = std_mpsc::channel::<HostedGameOver>();
            spawn_game_over_forwarder(
                outbound_tx.clone(),
                game_over_rx,
                engine_session.clone(),
                snapshot.clone(),
                game_id.clone(),
                Some(player_names.clone()),
            );
            let outbound_tx = outbound_tx.clone();
            let snapshot = snapshot.clone();
            spawn_engine_thread(move || {
                info!(
                    game_id,
                    backend = backend.label(),
                    players = num_players,
                    local_player_index,
                    "starting hosted engine thread"
                );
                crate::metrics::record_engine_session_started();
                let started = Instant::now();
                let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    java_backend::run_hosted_engine_game(
                        game_id.clone(),
                        player_names,
                        ordered_decks,
                        commander_names,
                        commander_variant,
                        game_variant,
                        local_player_index,
                        ai_player_indices,
                        starting_life,
                        remote_prompt_tx,
                        remote_response_rxs,
                        game_over_tx,
                        cancel,
                    )
                }));
                finish_hosted_engine(
                    result,
                    &game_id,
                    num_players,
                    started,
                    &outbound_tx,
                    &session_handle,
                    &snapshot,
                );
            });
        }
    }
}

fn clear_engine_session(engine_session: &SharedEngineSession) {
    match engine_session.lock() {
        Ok(mut guard) => *guard = None,
        Err(error) => warn!(%error, "engine session lock poisoned on reset"),
    }
}

/// Usernames of engine players the game still needs — seated in the current
/// game and not yet eliminated per the last state broadcast. Everyone else
/// (eliminated players, non-playing members) is effectively a spectator whose
/// absence never matters. `None` = no hosted game.
fn active_player_usernames(snapshot: &SharedHostSnapshot) -> Option<HashSet<String>> {
    let snap = snapshot.lock().ok()?;
    let game = snap.game.as_ref()?;
    let mut active: HashSet<String> = game.player_order.iter().cloned().collect();
    let players = snap
        .last_state
        .as_ref()
        .and_then(|state| state.pointer("/state/gameView/players"))
        .and_then(Value::as_array);
    if let Some(players) = players {
        for player in players {
            let playing = player
                .get("status")
                .and_then(Value::as_str)
                .is_none_or(|status| status == "playing");
            if playing {
                continue;
            }
            let index = player
                .get("id")
                .and_then(Value::as_str)
                .and_then(parse_player_slot);
            if let Some(username) = index.and_then(|i| game.player_order.get(i)) {
                active.remove(username);
            }
        }
    }
    Some(active)
}

fn seat_index_of(snapshot: &SharedHostSnapshot, username: &str) -> Option<usize> {
    let snap = snapshot.lock().ok()?;
    let game = snap.game.as_ref()?;
    game.player_order.iter().position(|name| name == username)
}

fn concede_missing_active_seats(
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    room: &RoomInfo,
) {
    let Some(active) = active_player_usernames(snapshot) else {
        return;
    };
    let present: HashSet<&str> = room
        .players
        .iter()
        .map(|player| player.username.as_str())
        .collect();
    for username in active {
        if !present.contains(username.as_str()) {
            if let Some(index) = seat_index_of(snapshot, &username) {
                concede_seat(engine_session, index);
            }
        }
    }
}

fn concede_seat(engine_session: &SharedEngineSession, player_index: usize) {
    send_seat_message(
        engine_session,
        player_index,
        ClientToServerMessage::Directive {
            directive: manabrew_protocol::transport::DirectiveInput::Concede,
        },
    );
}

fn send_seat_message(
    engine_session: &SharedEngineSession,
    player_index: usize,
    message: ClientToServerMessage,
) {
    let guard = match engine_session.lock() {
        Ok(guard) => guard,
        Err(error) => {
            warn!(%error, "engine session lock poisoned");
            return;
        }
    };
    let Some(session) = guard.as_ref() else {
        debug!(player_index, "no engine session for seat action");
        return;
    };
    let txs = match session {
        EngineSession::Manabrew {
            remote_response_txs,
            ..
        }
        | EngineSession::Forge {
            remote_response_txs,
            ..
        } => remote_response_txs,
    };
    let Some(tx) = txs.get(&player_index) else {
        debug!(player_index, "no response channel for player");
        return;
    };
    if let Err(error) = tx.send(message) {
        warn!(player_index, %error, "failed to route seat message");
    }
}

fn finish_hosted_engine(
    result: std::thread::Result<Result<(), String>>,
    game_id: &str,
    players: usize,
    started: Instant,
    outbound_tx: &tokio_mpsc::UnboundedSender<ClientMessage>,
    session_handle: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
) {
    let fatal = match result {
        Ok(Ok(())) => {
            info!(game_id, "hosted engine thread finished");
            None
        }
        Ok(Err(message)) => {
            error!(game_id, message, "hosted engine exited with a fatal error");
            Some(message)
        }
        Err(panic) => {
            let message = if let Some(message) = panic.downcast_ref::<String>() {
                message.clone()
            } else if let Some(message) = panic.downcast_ref::<&str>() {
                message.to_string()
            } else {
                "the host engine panicked".to_string()
            };
            error!(game_id, message, "hosted engine thread panicked");
            Some(message)
        }
    };
    crate::metrics::record_engine_session_finished(players, started, fatal.as_deref());
    let still_owner = session_handle
        .lock()
        .map(|guard| guard.as_ref().is_some_and(|s| s.game_id() == game_id))
        .unwrap_or(false);
    if let Some(message) = fatal {
        if still_owner {
            if let Ok(mut snap) = snapshot.lock() {
                if snap
                    .game
                    .as_ref()
                    .is_some_and(|game| game.game_id == game_id)
                {
                    snap.pending_end_game = Some(game_id.to_string());
                }
            }
            let _ = outbound_tx.send(ClientMessage::ReportGameOutcome {
                game_id: game_id.to_string(),
                outcome: GameOutcomeReport {
                    fatal_message: Some(message.clone()),
                    ..GameOutcomeReport::default()
                },
            });
            if let Ok(state) = serde_json::to_value(StateEnvelope::Fatal { message }) {
                let _ = outbound_tx.send(ClientMessage::BroadcastState {
                    state,
                    target_player: None,
                });
            }
            let _ = outbound_tx.send(ClientMessage::EndGame {
                game_id: game_id.to_string(),
            });
        } else {
            warn!(game_id, message, "stale engine session finished with error");
        }
    }
}

/// `from_player` is relay-attested; never client supplied.
fn route_seat_envelope(
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    from_player: &str,
    state: &Value,
) {
    match serde_json::from_value::<StateEnvelope>(state.clone()) {
        Ok(StateEnvelope::Response { .. }) => {
            route_remote_response(engine_session, snapshot, from_player, state)
        }
        Ok(StateEnvelope::Directive {
            from_player: claimed_slot,
            directive,
        }) => route_remote_directive(
            engine_session,
            snapshot,
            from_player,
            &claimed_slot,
            &directive,
        ),
        _ => debug!(from_player, state = %state, "unroutable seat envelope"),
    }
}

fn route_remote_response(
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    authenticated_username: &str,
    state: &Value,
) {
    let envelope: StateEnvelope = match serde_json::from_value(state.clone()) {
        Ok(envelope) => envelope,
        Err(error) => {
            warn!(%error, state = %state, "relay response invalid envelope");
            return;
        }
    };
    let StateEnvelope::Response {
        from_player,
        prompt_id,
        action: action_value,
    } = envelope
    else {
        warn!(state = %state, "expected response envelope");
        return;
    };
    let Some(player_index) =
        authenticated_player_index(snapshot, authenticated_username, &from_player, "response")
    else {
        return;
    };
    let action: PromptOutput = match serde_json::from_value(action_value) {
        Ok(action) => action,
        Err(error) => {
            warn!(from_player, %error, "relay response has invalid action");
            return;
        }
    };

    let guard = match engine_session.lock() {
        Ok(guard) => guard,
        Err(error) => {
            warn!(%error, "engine session lock poisoned");
            return;
        }
    };
    let Some(session) = guard.as_ref() else {
        debug!(from_player, "no engine session for relay response");
        return;
    };
    if session.is_local_bot_duplicate(player_index, prompt_id) {
        debug!(
            from_player,
            prompt_id, "relay response already answered locally"
        );
        return;
    }
    let tx = match session {
        EngineSession::Manabrew {
            remote_response_txs,
            ..
        } => remote_response_txs.get(&player_index),
        EngineSession::Forge {
            remote_response_txs,
            ..
        } => {
            debug!(
                from_player,
                player_index, "routing relay response to java engine"
            );
            remote_response_txs.get(&player_index)
        }
    };
    let Some(tx) = tx else {
        debug!(from_player, player_index, "no response channel for player");
        return;
    };
    session.engine_clock().mark_response();
    if let Err(error) = tx.send(ClientToServerMessage::Response { prompt_id, action }) {
        warn!(from_player, %error, "failed to route relay response");
        return;
    }
    drop(guard);
    if let Ok(mut snap) = snapshot.lock() {
        snap.pending_prompts.remove(&from_player);
    }
}

fn route_remote_directive(
    engine_session: &SharedEngineSession,
    snapshot: &SharedHostSnapshot,
    authenticated_username: &str,
    claimed_slot: &str,
    directive: &Value,
) {
    let directive: DirectiveInput = match serde_json::from_value(directive.clone()) {
        Ok(directive) => directive,
        Err(error) => {
            warn!(claimed_slot, %error, "relay directive is invalid");
            return;
        }
    };
    let Some(player_index) =
        authenticated_player_index(snapshot, authenticated_username, claimed_slot, "directive")
    else {
        return;
    };
    info!(claimed_slot, player_index, ?directive, "routing directive");
    match directive {
        DirectiveInput::Concede => concede_seat(engine_session, player_index),
    }
}

fn authenticated_player_index(
    snapshot: &SharedHostSnapshot,
    authenticated_username: &str,
    claimed_slot: &str,
    message_kind: &str,
) -> Option<usize> {
    let Some(player_index) = seat_index_of(snapshot, authenticated_username) else {
        warn!(
            authenticated_username,
            message_kind, "relay sender has no engine seat"
        );
        return None;
    };
    let authenticated_slot = player_slot(player_index);
    if claimed_slot != authenticated_slot {
        warn!(
            authenticated_username,
            claimed_slot,
            authenticated_slot,
            message_kind,
            "relay sender claimed another engine seat"
        );
        return None;
    }
    Some(player_index)
}

/// Add the fingerprint of the carried state to a full state envelope, so a
/// receiver can tell whether a later patch applies to what it holds.
fn stamp_fingerprint(mut envelope: Value) -> Value {
    let Some(state) = envelope.get("state") else {
        return envelope;
    };
    let fingerprint = manabrew_relay_protocol::state_delta::fingerprint(state);
    if let Some(object) = envelope.as_object_mut() {
        object.insert("fingerprint".to_string(), Value::String(fingerprint));
    }
    envelope
}

/// Time spent building this envelope after `engine_ms` was sampled: serialising
/// the state, fingerprinting it, diffing it. Must be stamped last, or it reaches
/// the dedup comparisons and the diff base, neither of which may see it.
fn stamp_emit_ms(mut envelope: Value, elapsed: Duration) -> Value {
    if let Some(object) = envelope.as_object_mut() {
        object.insert(
            "emitMs".to_string(),
            Value::from(elapsed.as_millis().min(u128::from(u32::MAX)) as u32),
        );
    }
    envelope
}

/// Patch form of a fingerprinted state envelope, against the last one sent to
/// this seat. `None` the first time a seat is served, or if anything is missing,
/// and the caller falls back to the full state, which is always correct.
/// When the engine host last received a player's response, as milliseconds on
/// its own monotonic clock. The response path writes it and the emit path
/// reads it, so an outgoing envelope can say how long the engine took without
/// anyone comparing clocks across hosts.
#[derive(Clone, Default)]
pub struct EngineClock(std::sync::Arc<std::sync::atomic::AtomicU64>);

impl EngineClock {
    fn now() -> u64 {
        use std::sync::OnceLock;
        static START: OnceLock<Instant> = OnceLock::new();
        START.get_or_init(Instant::now).elapsed().as_millis() as u64
    }

    pub fn mark_response(&self) {
        self.0
            .store(Self::now(), std::sync::atomic::Ordering::Relaxed);
    }

    /// Cleared once the decision it was timing completes, so an envelope emitted
    /// later, while a player is thinking, cannot report that wait as engine time.
    pub fn mark_decision_complete(&self) {
        self.0.store(0, std::sync::atomic::Ordering::Relaxed);
    }

    /// Milliseconds since the last response, or `None` if none has arrived yet
    /// or the decision it was timing has already completed.
    ///
    /// There is deliberately no upper bound. Idle time used to be excluded by
    /// discarding anything past 120s, which also discarded the slow decisions
    /// the metric exists to expose: a real 175s decision reported no `engineMs`
    /// at all, so the field was absent exactly when it mattered and every
    /// quantile built from it was understated. `mark_decision_complete` draws
    /// that line by state instead of by duration.
    pub fn elapsed_ms(&self) -> Option<u32> {
        let at = self.0.load(std::sync::atomic::Ordering::Relaxed);
        if at == 0 {
            return None;
        }
        let delta = Self::now().saturating_sub(at);
        Some(delta.min(u32::MAX as u64) as u32)
    }
}

fn patch_against_last(
    bases: &mut HashMap<usize, (Value, String)>,
    player_index: usize,
    per_seat: bool,
    slot: &str,
    envelope: &Value,
    engine_ms: Option<u32>,
) -> Option<Value> {
    let next = envelope.get("state")?.clone();
    let fingerprint = envelope.get("fingerprint")?.as_str()?.to_string();
    let previous = bases.insert(player_index, (next.clone(), fingerprint.clone()));
    let (base_state, base) = previous?;
    let patch = manabrew_relay_protocol::state_delta::diff(&base_state, &next)?;
    serde_json::to_value(StateEnvelope::StateDelta {
        for_player: per_seat.then(|| slot.to_string()),
        base,
        fingerprint,
        patch,
        engine_ms,
        emit_ms: None,
    })
    .ok()
}

/// This node's own bot seats, answered in-process with the same agent their
/// relay connection runs. Saves two relay hops per bot decision.
///
/// `config.bot_agent` is what makes that first sentence true rather than a
/// coincidence: `spawn_bots` gives the seat's relay connection the same value.
/// Building a fixed agent here instead would silently outrank whatever the
/// seat is actually configured to play, because the relay's answer arrives
/// second and is dropped by `is_local_bot_duplicate`.
fn local_bot_seats(
    config: &Config,
    player_names: &[String],
    bot_usernames: &HashSet<String>,
) -> HashMap<usize, BotResponder> {
    if !config.bot_enabled || !config.bot_local_answers || config.forge_ai {
        return HashMap::new();
    }
    player_names
        .iter()
        .enumerate()
        .filter(|(_, name)| bot_usernames.contains(*name))
        .map(|(index, name)| {
            info!(
                player_index = index,
                seat = name.as_str(),
                agent = ?config.bot_agent,
                "answering this bot seat in-process; its relay answers are dropped"
            );
            (index, BotResponder::new(config.bot_agent.build()))
        })
        .collect()
}

/// Answer a bot seat's prompt in-process. State and prompt still go to the
/// relay; its copy of the answer is dropped on arrival.
fn answer_bot_locally(
    session: &EngineSession,
    snapshot: &SharedHostSnapshot,
    player_index: usize,
    bot: &mut BotResponder,
    message: &AgentMessage,
) {
    match message {
        AgentMessage::State(update) => bot.observe(update.game_view.clone()),
        AgentMessage::Prompt(prompt) => {
            let response = bot.respond(prompt.clone());
            let ClientToServerMessage::Response { prompt_id, .. } = &response else {
                return;
            };
            if let Ok(mut answers) = session.local_bot_answers().lock() {
                answers.insert(player_index, *prompt_id);
            }
            let Some(tx) = session.response_tx(player_index) else {
                debug!(player_index, "no response channel for local bot");
                return;
            };
            session.engine_clock().mark_response();
            if let Err(error) = tx.send(response) {
                warn!(player_index, %error, "failed to route local bot response");
                return;
            }
            if let Ok(mut snap) = snapshot.lock() {
                snap.pending_prompts.remove(&player_slot(player_index));
            }
        }
        _ => {}
    }
}

#[allow(clippy::too_many_arguments)]
fn spawn_remote_prompt_forwarder(
    outbound_tx: tokio_mpsc::UnboundedSender<ClientMessage>,
    snapshot: SharedHostSnapshot,
    engine_session: SharedEngineSession,
    game_id: String,
    remote_prompt_rx: std_mpsc::Receiver<(usize, AgentMessage)>,
    seat_usernames: Option<Vec<String>>,
    state_delta: bool,
    engine_clock: EngineClock,
    mut local_bots: HashMap<usize, BotResponder>,
) {
    thread::spawn(move || {
        let mut last_state_by_seat: HashMap<usize, Value> = HashMap::new();
        let mut delta_bases: HashMap<usize, (Value, String)> = HashMap::new();
        let mut last_state: Option<Value> = None;
        let mut last_display: Option<Value> = None;
        while let Ok((player_index, message)) = remote_prompt_rx.recv() {
            let Ok(session) = engine_session.lock() else {
                break;
            };
            let Some(current) = session
                .as_ref()
                .filter(|session| session.game_id() == game_id)
            else {
                break;
            };
            let per_seat = seat_usernames.is_some() && player_index != OBSERVER_SEAT;
            let slot = player_slot(player_index);
            let engine_ms = engine_clock.elapsed_ms();
            if matches!(message, AgentMessage::Prompt(_)) {
                engine_clock.mark_decision_complete();
            }
            let emit_started = Instant::now();
            let envelope = match &message {
                AgentMessage::State(state_update) if !per_seat => StateEnvelope::State {
                    for_player: None,
                    state: serde_json::to_value(state_update).unwrap_or(Value::Null),
                    fingerprint: None,
                    engine_ms,
                    emit_ms: None,
                },
                _ => StateEnvelope::for_agent_message_timed(slot.clone(), &message, engine_ms),
            };
            let Ok(state) = serde_json::to_value(envelope) else {
                continue;
            };
            let state = match &message {
                AgentMessage::State(_) if state_delta => stamp_fingerprint(state),
                _ => state,
            };
            match &message {
                AgentMessage::State(_) if per_seat => {
                    if last_state_by_seat.get(&player_index) == Some(&state) {
                        continue;
                    }
                    last_state_by_seat.insert(player_index, state.clone());
                }
                AgentMessage::State(_) if last_state.as_ref() == Some(&state) => continue,
                AgentMessage::State(_) => last_state = Some(state.clone()),
                AgentMessage::Display(_) if last_display.as_ref() == Some(&state) => continue,
                AgentMessage::Display(_) => last_display = Some(state.clone()),
                AgentMessage::Prompt(_) | AgentMessage::Error(_) => {}
            }
            if let Ok(mut snap) = snapshot.lock() {
                match &message {
                    AgentMessage::State(_) if per_seat => {
                        snap.last_state_by_slot.insert(slot.clone(), state.clone());
                    }
                    AgentMessage::State(_) => snap.last_state = Some(state.clone()),
                    AgentMessage::Prompt(_) => {
                        snap.pending_prompts.insert(slot.clone(), state.clone());
                    }
                    AgentMessage::Display(_) | AgentMessage::Error(_) => {}
                }
            }
            if let Some(bot) = local_bots.get_mut(&player_index) {
                answer_bot_locally(current, &snapshot, player_index, bot, &message);
            }
            let state = match &message {
                AgentMessage::State(_) if state_delta => patch_against_last(
                    &mut delta_bases,
                    player_index,
                    per_seat,
                    &slot,
                    &state,
                    engine_ms,
                )
                .unwrap_or(state),
                _ => state,
            };
            let target_player = if per_seat
                && matches!(
                    message,
                    AgentMessage::State(_) | AgentMessage::Prompt(_) | AgentMessage::Error(_)
                ) {
                let Some(target_player) = seat_usernames
                    .as_ref()
                    .and_then(|names| names.get(player_index).cloned())
                else {
                    warn!(player_index, PRIVATE_MESSAGE_MISSING_PLAYER);
                    continue;
                };
                Some(target_player)
            } else {
                None
            };
            if outbound_tx
                .send(ClientMessage::BroadcastState {
                    state: stamp_emit_ms(state, emit_started.elapsed()),
                    target_player,
                })
                .is_err()
            {
                break;
            }
        }
    });
}

fn outcome_report(view: &GameViewDto) -> GameOutcomeReport {
    GameOutcomeReport {
        game_over: view.game_over,
        winner_slot: view.winner_id.clone(),
        conceded_slots: view
            .players
            .iter()
            .filter(|player| player.status == PlayerStatus::Conceded)
            .map(|player| player.id.clone())
            .collect(),
        fatal_message: None,
        turns: Some(view.turn),
    }
}

fn cached_outcome_report(envelope: &Value) -> Option<GameOutcomeReport> {
    let update: StateUpdate = serde_json::from_value(envelope.get("state")?.clone()).ok()?;
    Some(outcome_report(&update.game_view))
}

fn spawn_game_over_forwarder(
    outbound_tx: tokio_mpsc::UnboundedSender<ClientMessage>,
    game_over_rx: std_mpsc::Receiver<HostedGameOver>,
    session_handle: SharedEngineSession,
    snapshot: SharedHostSnapshot,
    game_id: String,
    seat_usernames: Option<Vec<String>>,
) {
    thread::spawn(move || {
        while let Ok(game_over) = game_over_rx.recv() {
            let Ok(session) = session_handle.lock() else {
                return;
            };
            if session
                .as_ref()
                .is_none_or(|session| session.game_id() != game_id)
            {
                warn!(
                    game_id,
                    "stale engine session reached game over; not ending the relay game"
                );
                continue;
            }
            if let Ok(mut snap) = snapshot.lock() {
                if snap
                    .game
                    .as_ref()
                    .is_some_and(|game| game.game_id == game_id)
                {
                    snap.pending_end_game = Some(game_id.clone());
                }
            }
            let outcome = game_over
                .messages
                .iter()
                .rev()
                .find_map(|(_, message)| match message {
                    AgentMessage::State(update) => Some(outcome_report(&update.game_view)),
                    _ => None,
                })
                .or_else(|| {
                    let snap = snapshot.lock().ok()?;
                    snap.last_state
                        .as_ref()
                        .or_else(|| snap.last_state_by_slot.values().next())
                        .and_then(cached_outcome_report)
                        .filter(|outcome| outcome.game_over)
                });
            let mut last_state_by_seat: HashMap<usize, Value> = HashMap::new();
            let mut last_state: Option<Value> = None;
            for (player_index, message) in game_over.messages {
                let per_seat = seat_usernames.is_some() && player_index != OBSERVER_SEAT;
                let envelope = match &message {
                    AgentMessage::State(state_update) if !per_seat => StateEnvelope::State {
                        for_player: None,
                        state: serde_json::to_value(state_update).unwrap_or(Value::Null),
                        fingerprint: None,
                        engine_ms: None,
                        emit_ms: None,
                    },
                    _ => StateEnvelope::for_agent_message(player_slot(player_index), &message),
                };
                let Ok(state) = serde_json::to_value(envelope) else {
                    continue;
                };
                match &message {
                    AgentMessage::State(_) if per_seat => {
                        if last_state_by_seat.get(&player_index) == Some(&state) {
                            continue;
                        }
                        last_state_by_seat.insert(player_index, state.clone());
                    }
                    AgentMessage::State(_) if last_state.as_ref() == Some(&state) => continue,
                    AgentMessage::State(_) => last_state = Some(state.clone()),
                    AgentMessage::Display(_) | AgentMessage::Prompt(_) | AgentMessage::Error(_) => {
                    }
                }
                let target_player = if per_seat
                    && matches!(
                        message,
                        AgentMessage::State(_) | AgentMessage::Prompt(_) | AgentMessage::Error(_)
                    ) {
                    let Some(target_player) = seat_usernames
                        .as_ref()
                        .and_then(|names| names.get(player_index).cloned())
                    else {
                        warn!(player_index, PRIVATE_MESSAGE_MISSING_PLAYER);
                        continue;
                    };
                    Some(target_player)
                } else {
                    None
                };
                if outbound_tx
                    .send(ClientMessage::BroadcastState {
                        state,
                        target_player,
                    })
                    .is_err()
                {
                    return;
                }
            }
            let Ok(state) = serde_json::to_value(StateEnvelope::RoomRelay {
                protocol: SELF_HOSTED_NODE_PROTOCOL.to_string(),
                version: 1,
                message_id: Uuid::new_v4().to_string(),
                from_player: None,
                target_player: None,
                room_id: None,
                payload: json!({ "type": "gameOver", "gameId": game_over.game_id }),
            }) else {
                continue;
            };
            if outbound_tx
                .send(ClientMessage::BroadcastState {
                    state,
                    target_player: None,
                })
                .is_err()
            {
                return;
            }
            if let Some(outcome) = outcome {
                let _ = outbound_tx.send(ClientMessage::ReportGameOutcome {
                    game_id: game_id.clone(),
                    outcome,
                });
            }
            if outbound_tx
                .send(ClientMessage::EndGame {
                    game_id: game_id.clone(),
                })
                .is_err()
            {
                return;
            }
        }
    });
}

async fn relay_writer(mut write: WsWrite, mut outbound: tokio_mpsc::UnboundedReceiver<Message>) {
    while let Some(message) = outbound.recv().await {
        let started = Instant::now();
        let closing = matches!(message, Message::Close(_));
        if let Err(error) = write.send(message).await {
            warn!(%error, "relay write failed");
            break;
        }
        crate::metrics::record_relay_send(started.elapsed());
        if closing {
            break;
        }
    }
    let _ = write.close().await;
}

fn self_minted_identity_token(kind: &str, handle: &str) -> String {
    let iat = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|elapsed| elapsed.as_secs() as i64)
        .unwrap_or_default();
    identity_token::mint_unsigned(&format!("{kind}:{handle}"), handle, iat, 24 * 60 * 60)
}

impl RelayClient {
    async fn connect(
        relay_url: &str,
        username: &str,
        password: &str,
    ) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
        info!(relay_url, username, "connecting relay client");
        let (socket, _) = connect_async_with_config(relay_url, None, true).await?;
        let (write, read) = socket.split();
        let (outbound, outbound_rx) = tokio_mpsc::unbounded_channel();
        let writer = tokio::spawn(relay_writer(write, outbound_rx));
        let mut client = Self {
            username: username.to_string(),
            outbound,
            read,
            writer,
        };
        client
            .send(&ClientMessage::Authenticate {
                username: username.to_string(),
                password: password.to_string(),
                service: true,
                identity: Some(IdentityProof {
                    token: Some(self_minted_identity_token("node", username)),
                    device: None,
                }),
                client_platform: ClientPlatform::Unknown,
                client_version: None,
                engine_gate: EngineGate::Unknown,
            })
            .await?;
        client.wait_for_auth().await?;
        Ok(client)
    }

    async fn wait_for_auth(&mut self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        loop {
            match self.recv().await? {
                Some(ServerMessage::AuthResult { success, error, .. }) => {
                    if success {
                        info!(username = %self.username, "authenticated");
                        return Ok(());
                    }
                    return Err(format!("authentication failed: {error:?}").into());
                }
                Some(other) => {
                    debug!(?other, username = %self.username, "ignored pre-auth message")
                }
                None => return Err("relay closed before authentication".into()),
            }
        }
    }

    async fn send(
        &mut self,
        message: &ClientMessage,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        self.enqueue(Message::Text(serde_json::to_string(message)?))
    }

    fn enqueue(&self, message: Message) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        self.outbound
            .send(message)
            .map_err(|_| "relay writer stopped".into())
    }

    async fn recv(
        &mut self,
    ) -> Result<Option<ServerMessage>, Box<dyn std::error::Error + Send + Sync>> {
        loop {
            let Some(message) = self.read.next().await else {
                return Ok(None);
            };
            let message = message?;
            match message {
                Message::Text(text) => {
                    return Ok(Some(serde_json::from_str(&text)?));
                }
                Message::Ping(payload) => {
                    self.enqueue(Message::Pong(payload))?;
                }
                Message::Close(_) => return Ok(None),
                _ => {}
            }
        }
    }

    async fn close(&mut self) {
        if self.enqueue(Message::Close(None)).is_err() {
            return;
        }
        let _ = time::timeout(CLOSE_DRAIN_TIMEOUT, async {
            while let Some(Ok(_)) = self.read.next().await {}
        })
        .await;
        self.writer.abort();
    }

    async fn broadcast_room_message(
        &mut self,
        protocol: &str,
        payload: Value,
    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        let envelope = StateEnvelope::RoomRelay {
            protocol: protocol.to_string(),
            version: 1,
            message_id: Uuid::new_v4().to_string(),
            from_player: Some(self.username.clone()),
            target_player: None,
            room_id: None,
            payload,
        };
        self.send(&ClientMessage::BroadcastState {
            state: serde_json::to_value(envelope)?,
            target_player: None,
        })
        .await
    }
}

fn log_room_update(observer: &str, room: &RoomInfo) {
    let players = room
        .players
        .iter()
        .map(|player| {
            format!(
                "{}{}{}",
                player.username,
                if player.ready { ":ready" } else { "" },
                if player.connected { "" } else { ":offline" },
            )
        })
        .collect::<Vec<_>>()
        .join(",");
    info!(
        observer,
        room_id = %room.room_id,
        room_name = %room.room_name,
        host = %room.host,
        players,
        "room update"
    );
}

#[cfg(test)]
mod local_bot_tests {
    use super::*;
    use manabrew_protocol::prompts::choose_action::ChooseActionInput;
    use manabrew_protocol::prompts::PromptInput;
    use manabrew_protocol::transport::AgentPrompt;
    use std::time::Duration;

    const HUMAN: &str = "alice";
    const BOT: &str = "forge-bot";

    fn config(bot_enabled: bool, local: bool, forge_ai: bool) -> Config {
        let mut config = Config::for_hosted_room(
            "ws://127.0.0.1:1".to_string(),
            String::new(),
            "t".to_string(),
            GameFormat::Any,
            2,
            None,
            None,
            None,
        );
        config.bot_enabled = bot_enabled;
        config.bot_local_answers = local;
        config.forge_ai = forge_ai;
        config
    }

    fn prompt(seat: usize, prompt_id: u32) -> AgentPrompt {
        AgentPrompt {
            prompt_id,
            deciding_player_id: player_slot(seat),
            source_card: None,
            source_ability_text: None,
            input: PromptInput::ChooseAction(ChooseActionInput {
                actions: Vec::new(),
            }),
        }
    }

    /// A hosted 2-seat game: seat 0 human, seat 1 this node's bot.
    struct Rig {
        engine_session: SharedEngineSession,
        snapshot: SharedHostSnapshot,
        outbound_rx: tokio_mpsc::UnboundedReceiver<ClientMessage>,
        prompt_tx: std_mpsc::Sender<(usize, AgentMessage)>,
        bot_rx: std_mpsc::Receiver<ClientToServerMessage>,
    }

    fn rig(config: &Config) -> Rig {
        let names = vec![HUMAN.to_string(), BOT.to_string()];
        let bots: HashSet<String> = [BOT.to_string()].into_iter().collect();
        let (outbound_tx, outbound_rx) = tokio_mpsc::unbounded_channel();
        let (prompt_tx, prompt_rx) = std_mpsc::channel();
        let (bot_tx, bot_rx) = std_mpsc::channel();
        let (human_tx, _human_rx) = std_mpsc::channel();
        let mut txs = HashMap::new();
        txs.insert(0, human_tx);
        txs.insert(1, bot_tx);
        let engine_clock = EngineClock::default();
        let engine_session: SharedEngineSession =
            Arc::new(Mutex::new(Some(EngineSession::Forge {
                game_id: "g1".to_string(),
                remote_response_txs: txs,
                cancel: Arc::new(AtomicBool::new(false)),
                engine_clock: engine_clock.clone(),
                local_bot_answers: LocalBotAnswers::default(),
            })));
        let snapshot: SharedHostSnapshot = Arc::new(Mutex::new(HostSnapshot {
            game: Some(GameStart {
                game_id: "g1".to_string(),
                player_order: names.clone(),
                player_decks: Vec::new(),
                starting_life: 20,
            }),
            ..HostSnapshot::default()
        }));
        spawn_remote_prompt_forwarder(
            outbound_tx,
            snapshot.clone(),
            engine_session.clone(),
            "g1".to_string(),
            prompt_rx,
            Some(names.clone()),
            false,
            engine_clock,
            local_bot_seats(config, &names, &bots),
        );
        Rig {
            engine_session,
            snapshot,
            outbound_rx,
            prompt_tx,
            bot_rx,
        }
    }

    fn relay_response(rig: &Rig, prompt_id: u32) {
        let state = serde_json::to_value(StateEnvelope::Response {
            from_player: player_slot(1),
            prompt_id,
            action: json!({ "type": "chooseAction", "output": { "type": "pass" } }),
        })
        .unwrap();
        route_remote_response(&rig.engine_session, &rig.snapshot, BOT, &state);
    }

    #[test]
    fn local_bot_answers_its_prompt_and_the_relay_copy_is_dropped() {
        let rig = rig(&config(true, true, false));
        rig.prompt_tx
            .send((1, AgentMessage::Prompt(prompt(1, 7))))
            .unwrap();

        let answer = rig
            .bot_rx
            .recv_timeout(Duration::from_secs(5))
            .expect("bot seat answered in-process");
        assert!(matches!(
            answer,
            ClientToServerMessage::Response { prompt_id: 7, .. }
        ));

        // The prompt still went out to the relay, unchanged. The local answer
        // lands first, so give the forwarder a moment to send.
        let mut rig = rig;
        let sent = tokio::runtime::Runtime::new()
            .unwrap()
            .block_on(async {
                tokio::time::timeout(Duration::from_secs(5), rig.outbound_rx.recv()).await
            })
            .expect("prompt forwarded to relay")
            .expect("outbound channel open");
        let ClientMessage::BroadcastState { target_player, .. } = sent else {
            panic!("expected a broadcast, got {sent:?}");
        };
        assert_eq!(target_player.as_deref(), Some(BOT));

        // The bot's own relay connection answers the same prompt later: dropped.
        relay_response(&rig, 7);
        assert!(
            rig.bot_rx.try_recv().is_err(),
            "relay duplicate reached the engine"
        );

        // A newer prompt the node did not answer still routes.
        relay_response(&rig, 8);
        assert!(matches!(
            rig.bot_rx.try_recv(),
            Ok(ClientToServerMessage::Response { prompt_id: 8, .. })
        ));
    }

    #[test]
    fn local_answers_are_off_without_bots_or_when_forge_plays_them() {
        for config in [
            config(false, true, false),
            config(true, false, false),
            config(true, true, true),
        ] {
            let rig = rig(&config);
            rig.prompt_tx
                .send((1, AgentMessage::Prompt(prompt(1, 7))))
                .unwrap();
            assert!(
                rig.bot_rx.recv_timeout(Duration::from_millis(200)).is_err(),
                "bot seat was answered in-process"
            );
            relay_response(&rig, 7);
            assert!(rig.bot_rx.try_recv().is_ok(), "relay answer did not route");
        }
    }
}