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;

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),
}

/// 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 player inputs from the client
    fn receive_inputs(&self) -> Result<Vec<PlayerInput>, 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_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError> {
        self.inner.receive_inputs()
    }
}

/// 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>,
    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_inputs() {
        Ok(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);
            }
        }
        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)
    }
}

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