Merge branch 'server' into main (PR #6)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

CHANGELOG.md

@@ -14,6 +14,14 @@ Format based on [Keep a Changelog](https://keepachangelog.com/).
 - 12 LocalBridge tests (framing roundtrips, cross-layer Protocol+framing, diagonal wire mapping)
 - 4 diagonal movement cross-language fixtures (Rust → GDScript, D-030 Layer 1)
 - 33 total client tests passing (up from 20)
+- TcpBridge transport for Godot client connection — TCP localhost IPC alongside existing Unix socket LocalBridge
+- Input processing system (process_player_input) — drains InputQueue, converts PlayerActions to MoveIntent components, handles pause/unpause
+- Snapshot generation system (generate_snapshot) — builds ObserverSnapshot from ECS state with render coordinate conversion
+- Bridge I/O systems (receive_bridge_inputs, send_bridge_snapshot) — wire bridge to ECS pipeline with graceful disconnect detection
+- Full game loop in main.rs — TCP accept, tick loop with ServerRunning resource, CLI/env addr config
+- PlayerCharacter marker component, Player EntityKind variant, SnapshotBuffer resource
+- E2E game_loop integration test verifying player movement through full pipeline
+- 8 new tests (3 TCP bridge + 4 input processing + 1 E2E game loop), total 53
 - v0.1 content gap analysis workshop — 6 agents, 2 rounds, 9 content layers, 8 new decisions (D-032 through D-039)
 - D-032: Separate monologue pools per playable character (hard partition, not filter)
 - D-033: Entity color represents relationship to player character (asymmetric per character)
@@ -103,6 +111,8 @@ Format based on [Keep a Changelog](https://keepachangelog.com/).
 - Rejected alternatives R-004 through R-010 documented (pure Bevy, pure Godot, GDExtension, C++ GDExtension, Fyrox, custom framework, protobuf)
 
 ### Fixed
+- Wire format mismatch: LocalBridge now uses rmp_serde::to_vec_named() (named maps) matching client Protocol.gd expectations
+- EOF on bridge read now returns BridgeError::Transport for disconnect detection instead of empty Vec
 - WalkabilityMap rewritten from flat Vec to chunk-based HashMap per D-012 architecture (Tyre review)
 - LocalBridge mutex .unwrap() → .expect() for clearer panic messages (Hoshe review)
 - Documented 16MB MAX_MESSAGE_SIZE rationale in framing.rs (Hoshe review)

server/src/bridge/local.rs

@@ -81,7 +81,7 @@ impl LocalBridge {
 
 impl SimBridge for LocalBridge {
     fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError> {
-        let payload = rmp_serde::to_vec(snapshot)?;
+        let payload = rmp_serde::to_vec_named(snapshot)?;
 
         let mut writer = self.writer.lock().expect("writer mutex poisoned");
         write_framed(writer.get_mut(), &payload)?;
@@ -99,10 +99,7 @@ impl SimBridge for LocalBridge {
                 tracing::trace!("received {} inputs", inputs.len());
                 Ok(inputs)
             }
-            None => {
+            None => Err(BridgeError::Disconnected),
-                tracing::trace!("received EOF, returning empty input vec");
-                Ok(Vec::new())
-            }
         }
     }
 }

server/src/bridge/mod.rs

@@ -4,9 +4,11 @@
 
 use bevy_app::prelude::*;
 use bevy_ecs::prelude::*;
+use bevy_ecs::schedule::IntoScheduleConfigs;
 
 pub mod framing;
 pub mod local;
+pub mod tcp;
 pub mod types;
 pub use types::*;
 
@@ -21,6 +23,8 @@ pub enum BridgeError {
     Io(#[from] std::io::Error),
     #[error("transport error: {0}")]
     Transport(String),
+    #[error("client disconnected")]
+    Disconnected,
 }
 
 /// Abstracts transport layer (D-020)
@@ -55,13 +59,112 @@ impl BridgeResource {
     }
 }
 
+/// Generate ObserverSnapshot from ECS state
+pub fn generate_snapshot(
+    time: Res<crate::simulation::time::SimulationTime>,
+    entities: Query<(
+        Entity,
+        &crate::simulation::movement::TilePosition,
+        Option<&crate::simulation::movement::PlayerCharacter>,
+        Option<&crate::npc::Npc>,
+    )>,
+    mut buffer: ResMut<SnapshotBuffer>,
+) {
+    let mut visible = Vec::new();
+    for (entity, pos, is_player, is_npc) in entities.iter() {
+        let (x, y, z) = pos.to_render_coords();
+        let kind = if is_player.is_some() {
+            EntityKind::Player
+        } else if is_npc.is_some() {
+            EntityKind::Npc
+        } else {
+            EntityKind::Object
+        };
+        visible.push(VisibleEntity {
+            entity_id: entity.to_bits(),
+            x,
+            y,
+            z,
+            kind,
+        });
+    }
+    tracing::trace!(
+        "generate_snapshot: tick={}, entities={}",
+        time.tick,
+        visible.len()
+    );
+    buffer.snapshot = Some(ObserverSnapshot {
+        tick: time.tick,
+        entities: visible,
+    });
+}
+
+/// Receive inputs from bridge and push to InputQueue
+pub fn receive_bridge_inputs(
+    bridge: Option<Res<BridgeResource>>,
+    mut input_queue: ResMut<crate::simulation::input::InputQueue>,
+    mut running: ResMut<ServerRunning>,
+) {
+    let Some(bridge) = bridge else { return };
+    match bridge.receive_inputs() {
+        Ok(inputs) => {
+            for input in inputs {
+                input_queue.push(input);
+            }
+        }
+        Err(BridgeError::Disconnected) => {
+            tracing::info!("Client disconnected, shutting down");
+            running.0 = false;
+        }
+        Err(e) => {
+            tracing::error!("Bridge receive error: {}", e);
+        }
+    }
+}
+
+/// Send snapshot from buffer to bridge
+pub fn send_bridge_snapshot(
+    bridge: Option<Res<BridgeResource>>,
+    mut buffer: ResMut<SnapshotBuffer>,
+    mut running: ResMut<ServerRunning>,
+) {
+    let Some(bridge) = bridge else { return };
+    if let Some(snapshot) = buffer.snapshot.take() {
+        if let Err(e) = bridge.send_snapshot(&snapshot) {
+            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) {
+    fn build(&self, app: &mut App) {
-        // Stub implementation - will be populated in phase 2
+        app.init_resource::<SnapshotBuffer>()
+            .init_resource::<ServerRunning>()
+            .add_systems(
+                Update,
+                (
+                    receive_bridge_inputs.before(crate::simulation::input::process_player_input),
+                    generate_snapshot
+                        .after(crate::simulation::movement::validate_movement)
+                        .before(crate::simulation::time::advance_tick),
+                    send_bridge_snapshot.after(generate_snapshot),
+                ),
+            );
         tracing::debug!("BridgePlugin initialized");
     }
 }

server/src/bridge/tcp.rs

@@ -0,0 +1,144 @@
+// TcpBridge - TCP localhost IPC implementation
+// Implements D-020 subprocess/IPC architecture
+// Deterministic client-server communication via TCP sockets
+// Used for Godot client which lacks Unix socket support
+
+use super::{BridgeError, ObserverSnapshot, PlayerInput, SimBridge};
+use crate::bridge::framing::{read_framed, write_framed};
+use std::io::{BufReader, BufWriter};
+use std::net::{SocketAddr, TcpListener, TcpStream};
+use std::sync::Mutex;
+
+/// TcpBridge: TCP transport for client-server IPC
+/// Same semantics as LocalBridge but over TCP localhost
+pub struct TcpBridge {
+    reader: Mutex<BufReader<TcpStream>>,
+    writer: Mutex<BufWriter<TcpStream>>,
+    local_addr: SocketAddr,
+}
+
+impl TcpBridge {
+    /// Server-side: bind TCP listener and accept one connection.
+    /// Binds to the specified address (e.g., "127.0.0.1:0" for OS-assigned port).
+    /// Returns the bridge with the actual bound address available via local_addr().
+    pub fn accept(addr: &str) -> Result<Self, BridgeError> {
+        tracing::info!("TcpBridge binding to {}", addr);
+
+        let listener = TcpListener::bind(addr)
+            .map_err(|e| BridgeError::Transport(format!("failed to bind TCP socket: {}", e)))?;
+
+        let local_addr = listener
+            .local_addr()
+            .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?;
+
+        tracing::info!("TcpBridge listening on {}", local_addr);
+
+        // Accept one connection
+        let (stream, peer_addr) = listener
+            .accept()
+            .map_err(|e| BridgeError::Transport(format!("failed to accept connection: {}", e)))?;
+
+        tracing::info!(
+            "TcpBridge accepted connection from {} on {}",
+            peer_addr,
+            local_addr
+        );
+
+        // Clone stream for reader and writer
+        let reader_stream = stream.try_clone().map_err(|e| {
+            BridgeError::Transport(format!("failed to clone stream for reader: {}", e))
+        })?;
+
+        Ok(Self {
+            reader: Mutex::new(BufReader::new(reader_stream)),
+            writer: Mutex::new(BufWriter::new(stream)),
+            local_addr,
+        })
+    }
+
+    /// Server-side: accept one connection on an existing TcpListener.
+    /// Avoids race conditions in tests by separating bind from accept.
+    pub fn accept_on(listener: TcpListener) -> Result<Self, BridgeError> {
+        let local_addr = listener
+            .local_addr()
+            .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?;
+
+        tracing::info!("TcpBridge accepting on {}", local_addr);
+
+        let (stream, peer_addr) = listener
+            .accept()
+            .map_err(|e| BridgeError::Transport(format!("failed to accept connection: {}", e)))?;
+
+        tracing::info!(
+            "TcpBridge accepted connection from {} on {}",
+            peer_addr,
+            local_addr
+        );
+
+        let reader_stream = stream.try_clone().map_err(|e| {
+            BridgeError::Transport(format!("failed to clone stream for reader: {}", e))
+        })?;
+
+        Ok(Self {
+            reader: Mutex::new(BufReader::new(reader_stream)),
+            writer: Mutex::new(BufWriter::new(stream)),
+            local_addr,
+        })
+    }
+
+    /// Client-side: connect to TCP address (for tests).
+    pub fn connect(addr: &str) -> Result<Self, BridgeError> {
+        tracing::info!("TcpBridge connecting to {}", addr);
+
+        let stream = TcpStream::connect(addr).map_err(|e| {
+            BridgeError::Transport(format!("failed to connect to TCP socket: {}", e))
+        })?;
+
+        let local_addr = stream
+            .local_addr()
+            .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?;
+
+        tracing::trace!("TcpBridge connected to {}", addr);
+
+        // Clone stream for reader and writer
+        let reader_stream = stream.try_clone().map_err(|e| {
+            BridgeError::Transport(format!("failed to clone stream for reader: {}", e))
+        })?;
+
+        Ok(Self {
+            reader: Mutex::new(BufReader::new(reader_stream)),
+            writer: Mutex::new(BufWriter::new(stream)),
+            local_addr,
+        })
+    }
+
+    /// Get the local address (useful for OS-assigned port discovery in tests).
+    pub fn local_addr(&self) -> SocketAddr {
+        self.local_addr
+    }
+}
+
+impl SimBridge for TcpBridge {
+    fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError> {
+        let payload = rmp_serde::to_vec_named(snapshot)?;
+
+        let mut writer = self.writer.lock().expect("writer mutex poisoned");
+        write_framed(writer.get_mut(), &payload)?;
+
+        tracing::trace!("sent snapshot: tick={}", snapshot.tick);
+        Ok(())
+    }
+
+    fn receive_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError> {
+        let mut reader = self.reader.lock().expect("reader mutex poisoned");
+
+        match read_framed(reader.get_mut())? {
+            Some(payload) => {
+                let inputs: Vec<PlayerInput> = rmp_serde::from_slice(&payload)?;
+                tracing::trace!("received {} inputs", inputs.len());
+                Ok(inputs)
+            }
+            None => Err(BridgeError::Disconnected),
+        }
+    }
+}

server/src/bridge/types.rs

@@ -2,6 +2,7 @@
 // ObserverSnapshot: data crossing the client-server boundary
 // PlayerInput: semantic actions from client
 
+use bevy_ecs::prelude::*;
 use serde::{Deserialize, Serialize};
 
 /// The ONLY data structure crossing the client-server boundary (D-020)
@@ -32,6 +33,7 @@ pub struct VisibleEntity {
 /// Category of visible entity
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum EntityKind {
+    Player,
     Npc,
     Object,
     Terrain,
@@ -63,3 +65,9 @@ pub enum PlayerAction {
     Pause,
     Unpause,
 }
+
+/// Snapshot buffer resource for staging outgoing ObserverSnapshots
+#[derive(Resource, Debug, Default)]
+pub struct SnapshotBuffer {
+    pub snapshot: Option<ObserverSnapshot>,
+}

server/src/main.rs

@@ -4,7 +4,9 @@
 use bevy_app::prelude::*;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
-use settled_reach_server::bridge::BridgePlugin;
+use settled_reach_server::bridge::tcp::TcpBridge;
+use settled_reach_server::bridge::{BridgePlugin, BridgeResource, ServerRunning};
+use settled_reach_server::simulation::movement::{PlayerCharacter, TilePosition, WalkabilityMap};
 use settled_reach_server::simulation::SimulationPlugin;
 
 fn main() {
@@ -17,15 +19,39 @@ fn main() {
         .with(tracing_subscriber::fmt::layer())
         .init();
 
+    let addr = std::env::args()
+        .nth(1)
+        .or_else(|| std::env::var("SR_ADDR").ok())
+        .unwrap_or_else(|| "127.0.0.1:9876".to_string());
+
     tracing::info!("The Settled Reach - Simulation Server starting");
+    tracing::info!("Waiting for client connection on {}", addr);
+
+    let bridge = TcpBridge::accept(&addr).unwrap_or_else(|e| {
+        tracing::error!("Failed to accept client connection on {}: {}", addr, e);
+        std::process::exit(1);
+    });
+    tracing::info!("Client connected, initializing simulation");
 
     // Create the bevy App and add plugins
     let mut app = App::new();
     app.add_plugins(SimulationPlugin);
     app.add_plugins(BridgePlugin);
+    app.insert_resource(BridgeResource::new(bridge));
+    app.insert_resource(WalkabilityMap::new(32, 32, 1));
+    app.world_mut()
+        .spawn((PlayerCharacter, TilePosition::new(16, 16, 0)));
 
-    // Single tick for smoke verification; real game loop in phase 2
+    tracing::info!("Simulation initialized, entering game loop");
-    app.update();
 
-    tracing::info!("Simulation server update complete");
+    // Game loop: run until client disconnects
+    loop {
+        app.update();
+        // Check ServerRunning resource
+        if !app.world().resource::<ServerRunning>().0 {
+            break;
+        }
+    }
+
+    tracing::info!("Simulation server shutting down");
 }

server/src/simulation/input.rs

@@ -2,7 +2,9 @@
 // Timestamped player input events for deterministic simulation (D-010 principle 4)
 // PlayerInput: semantic actions (MoveNorth, Interact, UsePerceptionMode)
 
-use crate::bridge::types::PlayerInput;
+use crate::bridge::types::{PlayerAction, PlayerInput};
+use crate::simulation::movement::{MoveIntent, PlayerCharacter, TilePosition};
+use crate::simulation::time::SimulationTime;
 use bevy_ecs::prelude::*;
 use std::collections::VecDeque;
 
@@ -51,10 +53,62 @@ impl InputQueue {
     }
 }
 
+/// Drains InputQueue for the current tick, converts PlayerActions to ECS components.
+pub fn process_player_input(
+    mut input_queue: ResMut<InputQueue>,
+    mut time: ResMut<SimulationTime>,
+    mut commands: Commands,
+    player_query: Query<(Entity, &TilePosition), With<PlayerCharacter>>,
+) {
+    let current_tick = time.tick;
+    let inputs = input_queue.drain_for_tick(current_tick);
+
+    for input in inputs {
+        match input.action {
+            PlayerAction::MoveNorth => apply_move(&player_query, &mut commands, 0, -1),
+            PlayerAction::MoveSouth => apply_move(&player_query, &mut commands, 0, 1),
+            PlayerAction::MoveEast => apply_move(&player_query, &mut commands, 1, 0),
+            PlayerAction::MoveWest => apply_move(&player_query, &mut commands, -1, 0),
+            PlayerAction::MoveNortheast => apply_move(&player_query, &mut commands, 1, -1),
+            PlayerAction::MoveNorthwest => apply_move(&player_query, &mut commands, -1, -1),
+            PlayerAction::MoveSoutheast => apply_move(&player_query, &mut commands, 1, 1),
+            PlayerAction::MoveSouthwest => apply_move(&player_query, &mut commands, -1, 1),
+            PlayerAction::Pause => {
+                time.paused = true;
+                tracing::debug!("Simulation paused by player input");
+            }
+            PlayerAction::Unpause => {
+                time.paused = false;
+                tracing::debug!("Simulation unpaused by player input");
+            }
+            PlayerAction::Interact => {
+                tracing::trace!("Interact action — no-op for Sprint 1");
+            }
+            PlayerAction::UsePerceptionMode(ref mode) => {
+                tracing::trace!("UsePerceptionMode({}) — no-op for Sprint 1", mode);
+            }
+        }
+    }
+}
+
+fn apply_move(
+    player_query: &Query<(Entity, &TilePosition), With<PlayerCharacter>>,
+    commands: &mut Commands,
+    dx: i32,
+    dy: i32,
+) {
+    if let Ok((entity, pos)) = player_query.single() {
+        commands.entity(entity).insert(MoveIntent {
+            target: TilePosition::new(pos.x + dx, pos.y + dy, pos.z),
+        });
+    } else {
+        tracing::warn!("No player entity found for movement input");
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::bridge::types::PlayerAction;
 
     #[test]
     fn drain_returns_inputs_up_to_tick() {
@@ -96,4 +150,97 @@ mod tests {
             action: PlayerAction::MoveSouth,
         });
     }
+
+    #[test]
+    fn process_input_move_creates_intent() {
+        let mut world = bevy_ecs::world::World::new();
+        world.insert_resource(InputQueue::default());
+        world.insert_resource(SimulationTime {
+            tick: 0,
+            paused: false,
+        });
+
+        let player = world
+            .spawn((PlayerCharacter, TilePosition::new(5, 5, 0)))
+            .id();
+
+        world.resource_mut::<InputQueue>().push(PlayerInput {
+            tick: 0,
+            action: PlayerAction::MoveNorth,
+        });
+
+        let mut schedule = bevy_ecs::schedule::Schedule::default();
+        schedule.add_systems(process_player_input);
+        schedule.run(&mut world);
+
+        let intent = world.get::<MoveIntent>(player).unwrap();
+        assert_eq!(intent.target, TilePosition::new(5, 4, 0));
+    }
+
+    #[test]
+    fn process_input_pause_toggles() {
+        let mut world = bevy_ecs::world::World::new();
+        world.insert_resource(InputQueue::default());
+        world.insert_resource(SimulationTime {
+            tick: 0,
+            paused: false,
+        });
+
+        world.resource_mut::<InputQueue>().push(PlayerInput {
+            tick: 0,
+            action: PlayerAction::Pause,
+        });
+
+        let mut schedule = bevy_ecs::schedule::Schedule::default();
+        schedule.add_systems(process_player_input);
+        schedule.run(&mut world);
+
+        assert!(world.resource::<SimulationTime>().paused);
+    }
+
+    #[test]
+    fn process_input_no_player_no_panic() {
+        let mut world = bevy_ecs::world::World::new();
+        world.insert_resource(InputQueue::default());
+        world.insert_resource(SimulationTime {
+            tick: 0,
+            paused: false,
+        });
+
+        world.resource_mut::<InputQueue>().push(PlayerInput {
+            tick: 0,
+            action: PlayerAction::MoveNorth,
+        });
+
+        let mut schedule = bevy_ecs::schedule::Schedule::default();
+        schedule.add_systems(process_player_input);
+        // Should not panic
+        schedule.run(&mut world);
+    }
+
+    #[test]
+    fn process_input_future_tick_ignored() {
+        let mut world = bevy_ecs::world::World::new();
+        world.insert_resource(InputQueue::default());
+        world.insert_resource(SimulationTime {
+            tick: 0,
+            paused: false,
+        });
+
+        let player = world
+            .spawn((PlayerCharacter, TilePosition::new(5, 5, 0)))
+            .id();
+
+        world.resource_mut::<InputQueue>().push(PlayerInput {
+            tick: 5,
+            action: PlayerAction::MoveNorth,
+        });
+
+        let mut schedule = bevy_ecs::schedule::Schedule::default();
+        schedule.add_systems(process_player_input);
+        schedule.run(&mut world);
+
+        // No MoveIntent should be created (input for future tick)
+        assert!(world.get::<MoveIntent>(player).is_none());
+    }
 }

server/src/simulation/mod.rs

@@ -20,10 +20,13 @@ impl Plugin for SimulationPlugin {
         app.init_resource::<time::SimulationTime>()
             .insert_resource(rng::SimRng::new(0))
             .init_resource::<input::InputQueue>()
-            .add_systems(Update, time::advance_tick)
             .add_systems(
                 Update,
-                movement::validate_movement.after(time::advance_tick),
+                (
+                    input::process_player_input,
+                    movement::validate_movement.after(input::process_player_input),
+                    time::advance_tick.after(movement::validate_movement),
+                ),
             );
 
         tracing::debug!("SimulationPlugin initialized");

server/src/simulation/movement.rs

@@ -10,6 +10,10 @@ use std::collections::HashMap;
 /// Chunk size in tiles (32x32 per chunk)
 pub const CHUNK_SIZE: i32 = 32;
 
+/// Marker component identifying the player-controlled entity.
+#[derive(Component, Debug)]
+pub struct PlayerCharacter;
+
 /// Tile position component for grid-based movement.
 /// Discrete integer coordinates used in simulation; converted to f32
 /// at the bridge boundary for VisibleEntity wire format.

server/tests/bridge_ipc.rs

@@ -106,7 +106,7 @@ fn input_roundtrip_over_unix_socket() {
         },
     ];
 
-    let payload = rmp_serde::to_vec(&inputs).expect("failed to serialize");
+    let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize");
     write_framed(&mut writer, &payload).expect("failed to write frame");
 
     // Drop writer to close connection and signal EOF to server

server/tests/bridge_tcp.rs

@@ -0,0 +1,131 @@
+//! Integration tests for TcpBridge over TCP localhost (D-030 Layer 2: IPC roundtrip).
+
+use settled_reach_server::bridge::framing::{read_framed, write_framed};
+use settled_reach_server::bridge::tcp::TcpBridge;
+use settled_reach_server::bridge::types::*;
+use settled_reach_server::bridge::SimBridge;
+use std::net::{TcpListener, TcpStream};
+use std::thread;
+
+#[test]
+fn snapshot_roundtrip_over_tcp() {
+    // Bind listener first — port is guaranteed ready before spawning threads
+    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
+    let server_addr = listener.local_addr().expect("failed to get local address");
+
+    // Server thread: accept on pre-bound listener and send snapshot
+    let server_handle = thread::spawn(move || {
+        let bridge = TcpBridge::accept_on(listener).expect("failed to accept");
+
+        let snapshot = ObserverSnapshot {
+            tick: 42,
+            entities: vec![VisibleEntity {
+                entity_id: 100,
+                x: 10.5,
+                y: 20.3,
+                z: 0,
+                kind: EntityKind::Npc,
+            }],
+        };
+
+        bridge
+            .send_snapshot(&snapshot)
+            .expect("failed to send snapshot");
+    });
+
+    // Client: connect and receive snapshot (no sleep needed — listener already bound)
+    let stream = TcpStream::connect(server_addr).expect("failed to connect");
+    let mut reader = std::io::BufReader::new(stream);
+
+    let payload = read_framed(&mut reader)
+        .expect("failed to read frame")
+        .expect("unexpected EOF");
+
+    let snapshot: ObserverSnapshot =
+        rmp_serde::from_slice(&payload).expect("failed to deserialize");
+
+    assert_eq!(snapshot.tick, 42);
+    assert_eq!(snapshot.entities.len(), 1);
+    assert_eq!(snapshot.entities[0].entity_id, 100);
+    assert_eq!(snapshot.entities[0].x, 10.5);
+    assert_eq!(snapshot.entities[0].y, 20.3);
+
+    server_handle.join().expect("server thread panicked");
+}
+
+#[test]
+fn input_roundtrip_over_tcp() {
+    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
+    let server_addr = listener.local_addr().expect("failed to get local address");
+
+    let server_handle = thread::spawn(move || {
+        let bridge = TcpBridge::accept_on(listener).expect("failed to accept");
+
+        let inputs = bridge.receive_inputs().expect("failed to receive inputs");
+
+        assert_eq!(inputs.len(), 2);
+        assert_eq!(inputs[0].tick, 10);
+        assert_eq!(inputs[1].tick, 11);
+
+        inputs
+    });
+
+    // Client: connect and send inputs
+    let stream = TcpStream::connect(server_addr).expect("failed to connect");
+    let mut writer = std::io::BufWriter::new(stream);
+
+    let inputs = vec![
+        PlayerInput {
+            tick: 10,
+            action: PlayerAction::MoveNorth,
+        },
+        PlayerInput {
+            tick: 11,
+            action: PlayerAction::Interact,
+        },
+    ];
+
+    let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize");
+    write_framed(&mut writer, &payload).expect("failed to write frame");
+
+    // Drop writer to close connection and signal EOF to server
+    drop(writer);
+
+    let received_inputs = server_handle.join().expect("server thread panicked");
+
+    match &received_inputs[0].action {
+        PlayerAction::MoveNorth => {}
+        _ => panic!("expected MoveNorth action"),
+    }
+    match &received_inputs[1].action {
+        PlayerAction::Interact => {}
+        _ => panic!("expected Interact action"),
+    }
+}
+
+#[test]
+fn tcp_bridge_eof_returns_error() {
+    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
+    let server_addr = listener.local_addr().expect("failed to get local address");
+
+    let server_handle = thread::spawn(move || {
+        let bridge = TcpBridge::accept_on(listener).expect("failed to accept");
+
+        let result = bridge.receive_inputs();
+
+        assert!(result.is_err(), "expected error on EOF");
+        assert!(
+            matches!(
+                result,
+                Err(settled_reach_server::bridge::BridgeError::Disconnected)
+            ),
+            "expected Disconnected error"
+        );
+    });
+
+    // Client: connect and immediately disconnect without sending data
+    let stream = TcpStream::connect(server_addr).expect("failed to connect");
+    drop(stream);
+
+    server_handle.join().expect("server thread panicked");
+}

server/tests/game_loop.rs

@@ -0,0 +1,74 @@
+//! E2E integration test: full game loop with input processing and snapshot generation
+//! Tests the complete pipeline: client sends input -> server processes -> server sends snapshot
+
+use bevy_app::prelude::*;
+use settled_reach_server::bridge::framing::{read_framed, write_framed};
+use settled_reach_server::bridge::tcp::TcpBridge;
+use settled_reach_server::bridge::types::*;
+use settled_reach_server::bridge::{BridgePlugin, BridgeResource};
+use settled_reach_server::simulation::movement::{PlayerCharacter, TilePosition, WalkabilityMap};
+use settled_reach_server::simulation::SimulationPlugin;
+use std::io::{BufReader, BufWriter};
+use std::net::{TcpListener, TcpStream};
+use std::thread;
+
+#[test]
+fn player_moves_north_through_full_pipeline() {
+    // Bind listener first — port guaranteed ready, no sleep needed
+    let listener = TcpListener::bind("127.0.0.1:0").expect("bind listener");
+    let server_addr = listener.local_addr().expect("get local addr");
+
+    // Spawn server thread
+    let server_handle = thread::spawn(move || {
+        let bridge = TcpBridge::accept_on(listener).expect("accept connection");
+
+        // Build app
+        let mut app = App::new();
+        app.add_plugins(SimulationPlugin);
+        app.add_plugins(BridgePlugin);
+        app.insert_resource(BridgeResource::new(bridge));
+        app.insert_resource(WalkabilityMap::new(32, 32, 1));
+        app.world_mut()
+            .spawn((PlayerCharacter, TilePosition::new(16, 16, 0)));
+
+        // Run one tick: receive input, process, validate movement, generate snapshot, send
+        app.update();
+    });
+
+    // Client: connect and send input (no sleep — listener was pre-bound)
+    let stream = TcpStream::connect(server_addr).expect("client connect");
+    let mut reader = BufReader::new(stream.try_clone().expect("clone for reader"));
+    let mut writer = BufWriter::new(stream);
+
+    // Send PlayerInput: MoveNorth at tick 0
+    let inputs = vec![PlayerInput {
+        tick: 0,
+        action: PlayerAction::MoveNorth,
+    }];
+    let payload = rmp_serde::to_vec_named(&inputs).expect("serialize inputs");
+    write_framed(&mut writer, &payload).expect("send inputs");
+
+    // Receive ObserverSnapshot
+    let response = read_framed(&mut reader)
+        .expect("read snapshot")
+        .expect("not EOF");
+    let snapshot: ObserverSnapshot =
+        rmp_serde::from_slice(&response).expect("deserialize snapshot");
+
+    // Snapshot captures state at end of tick 0 (before advance_tick increments to 1)
+    assert_eq!(snapshot.tick, 0);
+    assert_eq!(snapshot.entities.len(), 1);
+
+    let player_entity = &snapshot.entities[0];
+    // Player started at (16, 16, 0), moved north (y-1) to (16, 15, 0)
+    // Render coords: (16.5, 15.5, 0)
+    assert_eq!(player_entity.x, 16.5);
+    assert_eq!(player_entity.y, 15.5);
+    assert_eq!(player_entity.z, 0);
+    assert!(matches!(player_entity.kind, EntityKind::Player));
+
+    // Clean up
+    drop(reader);
+    drop(writer);
+    server_handle.join().expect("server thread panicked");
+}

server/tests/gen_fixtures.rs

@@ -28,7 +28,10 @@ fn generate_msgpack_fixtures() {
             kind: EntityKind::Npc,
         }],
     };
-    write_fixture("snapshot_one_npc", &rmp_serde::to_vec_named(&snapshot).unwrap());
+    write_fixture(
+        "snapshot_one_npc",
+        &rmp_serde::to_vec_named(&snapshot).unwrap(),
+    );
 
     // Empty snapshot
     let empty = ObserverSnapshot {
@@ -42,25 +45,52 @@ fn generate_msgpack_fixtures() {
         tick: 100,
         action: PlayerAction::MoveNorth,
     };
-    write_fixture("input_move_north", &rmp_serde::to_vec_named(&input_north).unwrap());
+    write_fixture(
+        "input_move_north",
+        &rmp_serde::to_vec_named(&input_north).unwrap(),
+    );
 
     // PlayerInput: UsePerceptionMode
     let input_perception = PlayerInput {
         tick: 200,
         action: PlayerAction::UsePerceptionMode("thermal".to_string()),
     };
-    write_fixture("input_perception_mode", &rmp_serde::to_vec_named(&input_perception).unwrap());
+    write_fixture(
+        "input_perception_mode",
+        &rmp_serde::to_vec_named(&input_perception).unwrap(),
+    );
 
     // Snapshot with multiple entities and all EntityKind variants
     let snapshot_multi = ObserverSnapshot {
         tick: 999,
         entities: vec![
-            VisibleEntity { entity_id: 1, x: 5.0, y: 10.0, z: 0, kind: EntityKind::Npc },
+            VisibleEntity {
-            VisibleEntity { entity_id: 2, x: 15.5, y: 3.0, z: 1, kind: EntityKind::Object },
+                entity_id: 1,
-            VisibleEntity { entity_id: 3, x: 0.0, y: 0.0, z: -1, kind: EntityKind::Terrain },
+                x: 5.0,
+                y: 10.0,
+                z: 0,
+                kind: EntityKind::Npc,
+            },
+            VisibleEntity {
+                entity_id: 2,
+                x: 15.5,
+                y: 3.0,
+                z: 1,
+                kind: EntityKind::Object,
+            },
+            VisibleEntity {
+                entity_id: 3,
+                x: 0.0,
+                y: 0.0,
+                z: -1,
+                kind: EntityKind::Terrain,
+            },
         ],
     };
-    write_fixture("snapshot_multi_entity", &rmp_serde::to_vec_named(&snapshot_multi).unwrap());
+    write_fixture(
+        "snapshot_multi_entity",
+        &rmp_serde::to_vec_named(&snapshot_multi).unwrap(),
+    );
 
     // Diagonal movement fixtures (clockwise: NE, SE, SW, NW)
     for (name, action) in [

server/tests/serialization.rs

@@ -15,7 +15,7 @@ fn observer_snapshot_roundtrip() {
         }],
     };
 
-    let bytes = rmp_serde::to_vec(&snapshot).expect("serialize");
+    let bytes = rmp_serde::to_vec_named(&snapshot).expect("serialize");
     let decoded: ObserverSnapshot = rmp_serde::from_slice(&bytes).expect("deserialize");
 
     assert_eq!(decoded.tick, 42);
@@ -30,7 +30,7 @@ fn player_input_roundtrip() {
         action: PlayerAction::MoveNorth,
     };
 
-    let bytes = rmp_serde::to_vec(&input).expect("serialize");
+    let bytes = rmp_serde::to_vec_named(&input).expect("serialize");
     let decoded: PlayerInput = rmp_serde::from_slice(&bytes).expect("deserialize");
 
     assert_eq!(decoded.tick, 100);
@@ -43,7 +43,7 @@ fn empty_snapshot_roundtrip() {
         entities: vec![],
     };
 
-    let bytes = rmp_serde::to_vec(&snapshot).expect("serialize");
+    let bytes = rmp_serde::to_vec_named(&snapshot).expect("serialize");
     let decoded: ObserverSnapshot = rmp_serde::from_slice(&bytes).expect("deserialize");
 
     assert_eq!(decoded.tick, 0);
@@ -73,11 +73,11 @@ fn all_player_action_variants_roundtrip() {
             tick: 1,
             action: action.clone(),
         };
-        let bytes = rmp_serde::to_vec(&input).expect("serialize");
+        let bytes = rmp_serde::to_vec_named(&input).expect("serialize");
         let decoded: PlayerInput = rmp_serde::from_slice(&bytes).expect("deserialize");
         assert_eq!(decoded.tick, 1);
         // Verify the variant survived by re-serializing and comparing bytes
-        let re_bytes = rmp_serde::to_vec(&decoded).expect("re-serialize");
+        let re_bytes = rmp_serde::to_vec_named(&decoded).expect("re-serialize");
         assert_eq!(bytes, re_bytes, "round-trip mismatch for action variant");
     }
 }
@@ -85,7 +85,12 @@ fn all_player_action_variants_roundtrip() {
 /// All EntityKind variants must survive MessagePack round-trip (D-030 Layer 1)
 #[test]
 fn all_entity_kind_variants_roundtrip() {
-    let kinds = vec![EntityKind::Npc, EntityKind::Object, EntityKind::Terrain];
+    let kinds = vec![
+        EntityKind::Player,
+        EntityKind::Npc,
+        EntityKind::Object,
+        EntityKind::Terrain,
+    ];
 
     for (i, kind) in kinds.into_iter().enumerate() {
         let entity = VisibleEntity {
@@ -99,9 +104,9 @@ fn all_entity_kind_variants_roundtrip() {
             tick: 0,
             entities: vec![entity],
         };
-        let bytes = rmp_serde::to_vec(&snapshot).expect("serialize");
+        let bytes = rmp_serde::to_vec_named(&snapshot).expect("serialize");
         let decoded: ObserverSnapshot = rmp_serde::from_slice(&bytes).expect("deserialize");
-        let re_bytes = rmp_serde::to_vec(&decoded).expect("re-serialize");
+        let re_bytes = rmp_serde::to_vec_named(&decoded).expect("re-serialize");
         assert_eq!(
             bytes, re_bytes,
             "round-trip mismatch for EntityKind variant"