// LocalBridge - Unix socket IPC implementation // Implements D-020 subprocess/IPC architecture // Deterministic client-server communication via Unix domain sockets use super::{BridgeError, ObserverSnapshot, PlayerInput, SimBridge}; use crate::bridge::framing::{read_framed, write_framed}; use std::fs; use std::io::{BufReader, BufWriter}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; use std::sync::Mutex; /// LocalBridge: Unix socket transport for client-server IPC pub struct LocalBridge { reader: Mutex>, writer: Mutex>, socket_path: PathBuf, } impl LocalBridge { /// Server-side: create a Unix socket listener and accept one connection. /// Removes any stale socket file before binding. pub fn accept(path: &Path) -> Result { // Remove stale socket if it exists if path.exists() { fs::remove_file(path).map_err(|e| { BridgeError::Transport(format!("failed to remove stale socket: {}", e)) })?; } tracing::info!("LocalBridge listening on {:?}", path); let listener = UnixListener::bind(path) .map_err(|e| BridgeError::Transport(format!("failed to bind Unix socket: {}", e)))?; // Accept one connection let (stream, _addr) = listener .accept() .map_err(|e| BridgeError::Transport(format!("failed to accept connection: {}", e)))?; tracing::info!("LocalBridge accepted connection on {:?}", path); // 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)), socket_path: path.to_path_buf(), }) } /// Client-side: connect to an existing Unix socket. pub fn connect(path: &Path) -> Result { tracing::info!("LocalBridge connecting to {:?}", path); let stream = UnixStream::connect(path).map_err(|e| { BridgeError::Transport(format!("failed to connect to Unix socket: {}", e)) })?; tracing::trace!("LocalBridge connected to {:?}", path); // 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)), socket_path: path.to_path_buf(), }) } pub fn socket_path(&self) -> &Path { &self.socket_path } } 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() .map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?; write_framed(writer.get_mut(), &payload)?; tracing::trace!("sent snapshot: tick={}", snapshot.tick); Ok(()) } fn receive_inputs(&self) -> Result, BridgeError> { let mut reader = self .reader .lock() .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?; match read_framed(reader.get_mut())? { Some(payload) => match rmp_serde::from_slice::>(&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), } } } impl Drop for LocalBridge { fn drop(&mut self) { // Best-effort socket cleanup if self.socket_path.exists() { let _ = fs::remove_file(&self.socket_path); tracing::trace!("removed socket file {:?}", self.socket_path); } } }