settled-reach/server/src/bridge/mod.rs

// Bridge module - Client-server communication
// Implements D-020 subprocess/IPC architecture
// MessagePack serialization for Rust<->Godot communication

use bevy_app::prelude::*;
use bevy_ecs::prelude::*;
use bevy_ecs::schedule::IntoScheduleConfigs;

use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse};

pub mod debug;
pub mod framing;
pub mod local;
pub mod tcp;
pub mod text_renderer;
pub mod types;
pub use types::*;

/// Error type for bridge operations
#[derive(Debug, thiserror::Error)]
pub enum BridgeError {
    #[error("serialization error: {0}")]
    Serialization(#[from] rmp_serde::encode::Error),
    #[error("deserialization error: {0}")]
    Deserialization(#[from] rmp_serde::decode::Error),
    #[error("deserialization error (raw bytes logged): {0}")]
    DeserializationWithDump(String),
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
    #[error("transport error: {0}")]
    Transport(String),
    #[error("client disconnected")]
    Disconnected,
    #[error("internal mutex poisoned: {0}")]
    MutexPoisoned(String),
}

/// One decoded inbound message. The client→server stream is a single demuxed
/// channel (D-225): a `Vec<PlayerInput>` frame is a MessagePack *array* and an
/// `AtlasLayerRequest` frame is a *map*, so they are distinguishable without a
/// wire-level type tag (existing frames are byte-unchanged — additive).
#[derive(Debug)]
pub enum Inbound {
    /// A batch of player inputs (the gameplay path).
    Inputs(Vec<PlayerInput>),
    /// An atlas layer-stream request (#969, D-225).
    AtlasRequest(AtlasLayerRequest),
}

/// Demux a received frame payload into an [`Inbound`] (D-225). Tries
/// `Vec<PlayerInput>` (array), then `AtlasLayerRequest` (map); a frame that is
/// neither is a genuinely malformed input frame.
pub fn decode_inbound(payload: &[u8]) -> Result<Inbound, BridgeError> {
    if let Ok(inputs) = rmp_serde::from_slice::<Vec<PlayerInput>>(payload) {
        return Ok(Inbound::Inputs(inputs));
    }
    match rmp_serde::from_slice::<AtlasLayerRequest>(payload) {
        Ok(req) => Ok(Inbound::AtlasRequest(req)),
        Err(e) => {
            let dump_len = payload.len().min(256);
            tracing::error!(
                "inbound decode failed (neither inputs nor atlas request): {}. Raw ({} of {} bytes): {:02x?}",
                e,
                dump_len,
                payload.len(),
                &payload[..dump_len]
            );
            Err(BridgeError::DeserializationWithDump(format!(
                "{e} (payload {} bytes)",
                payload.len()
            )))
        }
    }
}

/// Abstracts transport layer (D-020)
/// Implemented by LocalBridge (stdio) and future NetworkBridge
pub trait SimBridge: Send + Sync {
    /// Send the protocol handshake as the first framed message (#555).
    /// Must be called exactly once, immediately after connection, before
    /// any ObserverSnapshot is sent.
    fn send_handshake(&self) -> Result<(), BridgeError>;

    /// Receive the client's startup message containing the world seed (#175).
    /// Called exactly once, after send_handshake(), before entering the tick loop.
    /// Blocks until the client sends the message.
    fn receive_startup(&self) -> Result<StartupMessage, BridgeError>;

    /// Send an observer snapshot to the client
    fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError>;

    /// Receive one inbound message, or `None` if no frame is ready this tick.
    /// The single client→server stream is demuxed by frame shape (D-225).
    fn receive(&self) -> Result<Option<Inbound>, BridgeError>;

    /// Send an atlas layer-stream response to the client (#969, D-225).
    fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError>;
}

/// BridgeResource: Bevy Resource wrapper for SimBridge trait object
#[derive(Resource)]
pub struct BridgeResource {
    inner: Box<dyn SimBridge>,
}

impl BridgeResource {
    pub fn new(bridge: impl SimBridge + 'static) -> Self {
        Self {
            inner: Box::new(bridge),
        }
    }

    pub fn send_handshake(&self) -> Result<(), BridgeError> {
        self.inner.send_handshake()
    }

    pub fn receive_startup(&self) -> Result<StartupMessage, BridgeError> {
        self.inner.receive_startup()
    }

    pub fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError> {
        self.inner.send_snapshot(snapshot)
    }

    pub fn receive(&self) -> Result<Option<Inbound>, BridgeError> {
        self.inner.receive()
    }

    pub fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> {
        self.inner.send_atlas_response(resp)
    }
}

/// Tracks whether the protocol handshake has been sent (#555).
/// Inserted by BridgePlugin as Pending. Set to Complete in main.rs after
/// `send_handshake()` succeeds. `receive_bridge_inputs` logs a warning
/// if inputs arrive while still Pending.
#[derive(Resource, Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum HandshakeState {
    /// Handshake not yet sent. Inputs arriving in this state trigger a warning.
    #[default]
    Pending,
    /// Handshake sent. Normal operation.
    Complete,
}

/// Receive inputs from bridge and push to InputQueue.
/// Protocol errors (malformed input) are recoverable: the frame is skipped
/// and a SimError is pushed to the SimErrorBuffer for client reporting (#85).
pub fn receive_bridge_inputs(
    bridge: Option<Res<BridgeResource>>,
    mut input_queue: ResMut<crate::simulation::input::InputQueue>,
    mut running: ResMut<ServerRunning>,
    handshake: Res<HandshakeState>,
    mut error_buffer: ResMut<SimErrorBuffer>,
    mut atlas_requests: ResMut<AtlasRequestBuffer>,
    time: Option<Res<crate::simulation::time::SimulationTime>>,
) {
    let Some(bridge) = bridge else { return };
    let current_tick = time.as_ref().map(|t| t.tick).unwrap_or(0);

    match bridge.receive() {
        Ok(Some(Inbound::Inputs(inputs))) => {
            if !inputs.is_empty() && *handshake == HandshakeState::Pending {
                tracing::warn!(
                    "Received {} input(s) before handshake completed — processing anyway (forward-compatible)",
                    inputs.len()
                );
            }
            for input in &inputs {
                tracing::trace!(
                    "Received input: tick={} action={:?}",
                    input.tick,
                    input.action
                );
            }
            for input in inputs {
                input_queue.push(input);
            }
        }
        Ok(Some(Inbound::AtlasRequest(req))) => {
            atlas_requests.0.push(req);
        }
        Ok(None) => {}
        Err(BridgeError::Disconnected) => {
            tracing::info!("Client disconnected, shutting down");
            running.0 = false;
        }
        Err(BridgeError::Io(ref e))
            if e.kind() == std::io::ErrorKind::BrokenPipe
                || e.kind() == std::io::ErrorKind::ConnectionReset =>
        {
            tracing::info!("Pipe broken, shutting down cleanly");
            running.0 = false;
        }
        Err(BridgeError::MutexPoisoned(ref msg)) => {
            tracing::error!("Bridge mutex poisoned: {}. Shutting down.", msg);
            running.0 = false;
        }
        Err(BridgeError::DeserializationWithDump(ref msg)) => {
            // Recoverable: skip this frame's input, report to client (#85)
            tracing::error!("Skipping malformed input frame: {}", msg);
            error_buffer.push(SimError {
                kind: SimErrorKind::ProtocolError,
                message: format!("Malformed input frame: {}", msg),
                tick: current_tick,
            });
        }
        Err(ref e @ BridgeError::Deserialization(_)) => {
            // Recoverable deserialization error without dump
            tracing::error!("Skipping malformed input: {}", e);
            error_buffer.push(SimError {
                kind: SimErrorKind::ProtocolError,
                message: format!("Deserialization error: {}", e),
                tick: current_tick,
            });
        }
        Err(e) => {
            tracing::error!("Bridge receive error: {}", e);
        }
    }
}

/// Send snapshot from buffer to bridge.
/// Any send error is fatal — the client cannot proceed without snapshots.
pub fn send_bridge_snapshot(
    bridge: Option<Res<BridgeResource>>,
    mut buffer: ResMut<SnapshotBuffer>,
    mut running: ResMut<ServerRunning>,
) {
    let Some(bridge) = bridge else {
        tracing::error!("send_bridge_snapshot: no BridgeResource");
        return;
    };
    if let Some(snapshot) = buffer.snapshot.take() {
        if let Err(e) = bridge.send_snapshot(&snapshot) {
            match &e {
                BridgeError::Disconnected => {
                    tracing::info!("Client disconnected during send, shutting down");
                }
                BridgeError::MutexPoisoned(msg) => {
                    tracing::error!("Bridge mutex poisoned during send: {}", msg);
                }
                _ => {
                    tracing::error!("Bridge send error: {}", e);
                }
            }
            running.0 = false;
        }
    }
}

/// Server running flag resource
#[derive(Resource, Debug, Clone)]
pub struct ServerRunning(pub bool);

impl Default for ServerRunning {
    fn default() -> Self {
        Self(true)
    }
}

/// Inbound atlas layer requests routed off the bridge (#969, D-225), drained by
/// the proxy serve system in `PreInput`.
#[derive(Resource, Default)]
pub struct AtlasRequestBuffer(pub Vec<AtlasLayerRequest>);

/// Outbound atlas layer responses, filled by the proxy serve system and flushed
/// to the client in `PostSnapshot` (#969, D-225).
#[derive(Resource, Default)]
pub struct AtlasResponseBuffer(pub Vec<AtlasLayerResponse>);

/// Flush buffered atlas responses to the client (#969, D-225). A failed send is
/// logged but not fatal — an atlas response is not load-bearing like a snapshot.
pub fn send_atlas_responses(
    bridge: Option<Res<BridgeResource>>,
    mut buffer: ResMut<AtlasResponseBuffer>,
) {
    let Some(bridge) = bridge else { return };
    for resp in buffer.0.drain(..) {
        if let Err(e) = bridge.send_atlas_response(&resp) {
            tracing::warn!("failed to send atlas response for {}: {}", resp.body_id, e);
        }
    }
}

/// Bridge plugin for client-server communication
/// Abstracts transport layer (LocalBridge/NetworkBridge)
pub struct BridgePlugin;

impl Plugin for BridgePlugin {
    fn build(&self, app: &mut App) {
        use crate::tick_phases::TickPhase;

        app.init_resource::<SnapshotBuffer>()
            .init_resource::<ServerRunning>()
            .init_resource::<HandshakeState>()
            .init_resource::<SimErrorBuffer>()
            .init_resource::<debug::DebugCommandBuffer>()
            .init_resource::<DebugEnabled>()
            .init_resource::<crate::perception::query::ActivePerceptionMode>()
            .init_resource::<AtlasRequestBuffer>()
            .init_resource::<AtlasResponseBuffer>()
            // Bridge I/O — PreInput (receive) and PostSnapshot (send)
            .add_systems(Update, receive_bridge_inputs.in_set(TickPhase::PreInput))
            .add_systems(Update, send_bridge_snapshot.in_set(TickPhase::PostSnapshot))
            .add_systems(Update, send_atlas_responses.in_set(TickPhase::PostSnapshot))
            // Debug commands — Snapshot phase
            .add_systems(
                Update,
                debug::handle_debug_commands.in_set(TickPhase::Snapshot),
            )
            // Monologue chain — Simulation phase, strict intra-phase sequence.
            // trigger_event_monologue must run after conversations + sound (also Simulation).
            .add_systems(
                Update,
                (
                    crate::simulation::monologue::trigger_monologue,
                    crate::simulation::monologue::trigger_recognition_monologue
                        .after(crate::simulation::monologue::trigger_monologue)
                        .after(crate::perception::anomaly::detect_anomalies),
                    crate::simulation::monologue::process_sprint_anomaly_monologue
                        .after(crate::simulation::monologue::trigger_recognition_monologue),
                    crate::simulation::monologue::trigger_event_monologue
                        .after(crate::simulation::monologue::process_sprint_anomaly_monologue)
                        .after(crate::simulation::sound::collect_sound_events)
                        .after(crate::simulation::dialogue::process_walk_away),
                    crate::simulation::monologue::process_contradiction_monologue
                        .after(crate::simulation::monologue::trigger_event_monologue),
                )
                    .in_set(TickPhase::Simulation),
            )
            // Observation systems — Simulation phase (reads positions, feeds snapshot)
            .add_systems(
                Update,
                (
                    crate::perception::observer::compute_visibility_geometry,
                    crate::simulation::interaction::compute_nearby_interactions,
                )
                    .in_set(TickPhase::Simulation),
            )
            // Observer snapshot assembly — Snapshot phase
            .add_systems(
                Update,
                crate::perception::observer::compute_observer_snapshot.in_set(TickPhase::Snapshot),
            )
            // Post-snapshot: emit observation events
            .add_systems(
                Update,
                crate::perception::observation::emit_observation_events
                    .in_set(TickPhase::PostSnapshot),
            );

        tracing::debug!("BridgePlugin initialized");
    }
}

#[cfg(test)]
mod inbound_tests {
    use super::*;
    use crate::atlas::cascade::CascadeLayer;

    #[test]
    fn demux_routes_inputs_and_atlas_requests() {
        // A Vec<PlayerInput> frame (msgpack array) → Inbound::Inputs.
        let inputs: Vec<PlayerInput> = vec![];
        let frame = rmp_serde::to_vec_named(&inputs).unwrap();
        assert!(matches!(decode_inbound(&frame), Ok(Inbound::Inputs(v)) if v.is_empty()));

        // An AtlasLayerRequest frame (msgpack map) → Inbound::AtlasRequest.
        let req = AtlasLayerRequest {
            body_id: "GJ1c".into(),
            up_to: CascadeLayer::Topography,
        };
        let frame = rmp_serde::to_vec_named(&req).unwrap();
        assert!(
            matches!(decode_inbound(&frame), Ok(Inbound::AtlasRequest(r)) if r.body_id == "GJ1c")
        );

        // Neither shape → a malformed-frame error.
        assert!(decode_inbound(&[0xff, 0xff]).is_err());
    }
}