fix(bridge): add error variants and diagnostic logging for IPC

Implements ticket #341 for Sprint 3:
- Add DeserializationWithDump and MutexPoisoned variants to BridgeError
- Replace .expect("mutex poisoned") with graceful error propagation
  in LocalBridge and TcpBridge (4 locations)
- Log first 256 bytes as hex dump on deserialization failure for
  debugging malformed payloads
- Classify errors in receive_bridge_inputs: BrokenPipe/ConnectionReset
  → clean shutdown, MutexPoisoned → shutdown, DeserializationWithDump
  → skip frame (recoverable)

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

server/src/bridge/local.rs

@@ -83,7 +83,10 @@ impl SimBridge for LocalBridge {
     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");
+        let mut writer = self
+            .writer
+            .lock()
+            .map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?;
         write_framed(writer.get_mut(), &payload)?;
 
         tracing::trace!("sent snapshot: tick={}", snapshot.tick);
@@ -91,14 +94,33 @@ impl SimBridge for LocalBridge {
     }
 
     fn receive_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError> {
-        let mut reader = self.reader.lock().expect("reader mutex poisoned");
+        let mut reader = self
+            .reader
+            .lock()
+            .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?;
 
         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)
-            }
+            Some(payload) => match rmp_serde::from_slice::<Vec<PlayerInput>>(&payload) {
+                Ok(inputs) => {
+                    tracing::trace!("received {} inputs", inputs.len());
+                    Ok(inputs)
+                }
+                Err(e) => {
+                    let dump_len = payload.len().min(256);
+                    tracing::error!(
+                        "deserialization failed: {}. Raw bytes ({} of {} total): {:02x?}",
+                        e,
+                        dump_len,
+                        payload.len(),
+                        &payload[..dump_len]
+                    );
+                    Err(BridgeError::DeserializationWithDump(format!(
+                        "{} (payload {} bytes)",
+                        e,
+                        payload.len()
+                    )))
+                }
+            },
             None => Err(BridgeError::Disconnected),
         }
     }

server/src/bridge/mod.rs

@@ -19,12 +19,16 @@ pub enum BridgeError {
     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)
@@ -133,13 +137,29 @@ pub fn receive_bridge_inputs(
             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, don't shut down
+            tracing::error!("Skipping malformed input frame: {}", msg);
+        }
         Err(e) => {
             tracing::error!("Bridge receive error: {}", e);
         }
     }
 }
 
-/// Send snapshot from buffer to bridge
+/// 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>,
@@ -148,7 +168,17 @@ pub fn send_bridge_snapshot(
     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);
+            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/src/bridge/tcp.rs

@@ -122,7 +122,10 @@ 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");
+        let mut writer = self
+            .writer
+            .lock()
+            .map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?;
         write_framed(writer.get_mut(), &payload)?;
 
         tracing::trace!("sent snapshot: tick={}", snapshot.tick);
@@ -130,14 +133,33 @@ impl SimBridge for TcpBridge {
     }
 
     fn receive_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError> {
-        let mut reader = self.reader.lock().expect("reader mutex poisoned");
+        let mut reader = self
+            .reader
+            .lock()
+            .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?;
 
         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)
-            }
+            Some(payload) => match rmp_serde::from_slice::<Vec<PlayerInput>>(&payload) {
+                Ok(inputs) => {
+                    tracing::trace!("received {} inputs", inputs.len());
+                    Ok(inputs)
+                }
+                Err(e) => {
+                    let dump_len = payload.len().min(256);
+                    tracing::error!(
+                        "deserialization failed: {}. Raw bytes ({} of {} total): {:02x?}",
+                        e,
+                        dump_len,
+                        payload.len(),
+                        &payload[..dump_len]
+                    );
+                    Err(BridgeError::DeserializationWithDump(format!(
+                        "{} (payload {} bytes)",
+                        e,
+                        payload.len()
+                    )))
+                }
+            },
             None => Err(BridgeError::Disconnected),
         }
     }