fix(engine): TCP partial-read frame desync + per-tick inbound drain (T-1045)

FrameAccumulator state machine in framing.rs holds partial prefix/payload
bytes across non-blocking receive() calls — a frame split across TCP
segments no longer desyncs the stream (read_exact previously discarded
partially-consumed bytes on WouldBlock). receive_bridge_inputs now drains
all ready frames per tick (capped) instead of exactly one, covering input
batches + atlas requests in the same window.

Review hardening: EOF mid-frame escalates to Disconnected like clean EOF
(peer died with a truncated stream) instead of logging an Io error every
tick. Tests: frame split inside prefix / inside payload, multi-frame
drain, mid-frame-EOF disconnect. Corrupt-stream escalation tracked as
T-1072.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

server/src/bridge/framing.rs

@@ -60,6 +60,110 @@ pub fn read_framed(reader: &mut impl Read) -> io::Result<Option<Vec<u8>>> {
     Ok(Some(payload))
 }
 
+/// Incremental frame reader for non-blocking streams (T-1045).
+///
+/// `read_framed` uses `read_exact`, which consumes bytes before erroring — a
+/// `WouldBlock` mid-frame on a non-blocking socket loses the bytes already
+/// read and permanently desyncs the stream (the next read treats mid-frame
+/// bytes as a length prefix). The accumulator instead retains partial
+/// prefix/payload bytes across calls, so a frame split across TCP segments
+/// is reassembled over multiple polls.
+pub struct FrameAccumulator {
+    state: AccumState,
+}
+
+enum AccumState {
+    /// Reading the 4-byte big-endian length prefix.
+    Prefix { buf: [u8; 4], filled: usize },
+    /// Prefix complete; reading `buf.len()` payload bytes.
+    Payload { buf: Vec<u8>, filled: usize },
+}
+
+impl Default for FrameAccumulator {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl FrameAccumulator {
+    pub fn new() -> Self {
+        Self {
+            state: AccumState::Prefix {
+                buf: [0u8; 4],
+                filled: 0,
+            },
+        }
+    }
+
+    /// Pump bytes from `reader` into the current frame.
+    ///
+    /// - `Ok(Some(payload))` — one complete frame.
+    /// - `Ok(None)` — clean EOF at a frame boundary (same as `read_framed`).
+    /// - `Err(WouldBlock)` — no complete frame yet; accumulated bytes are
+    ///   retained and the next call resumes mid-frame.
+    /// - `Err(UnexpectedEof)` — EOF mid-frame (truncated stream).
+    pub fn poll_frame(&mut self, reader: &mut impl Read) -> io::Result<Option<Vec<u8>>> {
+        loop {
+            match &mut self.state {
+                AccumState::Prefix { buf, filled } => {
+                    while *filled < buf.len() {
+                        match reader.read(&mut buf[*filled..]) {
+                            Ok(0) if *filled == 0 => return Ok(None), // clean EOF
+                            Ok(0) => {
+                                return Err(io::Error::new(
+                                    io::ErrorKind::UnexpectedEof,
+                                    "EOF inside length prefix",
+                                ));
+                            }
+                            Ok(n) => *filled += n,
+                            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
+                            Err(e) => return Err(e),
+                        }
+                    }
+                    let len = u32::from_be_bytes(*buf);
+                    if len > MAX_MESSAGE_SIZE {
+                        // State is left with the prefix filled: the stream is
+                        // corrupt, so every subsequent poll re-reports the
+                        // error instead of misreading payload bytes as a prefix.
+                        return Err(io::Error::new(
+                            io::ErrorKind::InvalidData,
+                            format!(
+                                "message too large: {} bytes (max {})",
+                                len, MAX_MESSAGE_SIZE
+                            ),
+                        ));
+                    }
+                    self.state = AccumState::Payload {
+                        buf: vec![0u8; len as usize],
+                        filled: 0,
+                    };
+                }
+                AccumState::Payload { buf, filled } => {
+                    while *filled < buf.len() {
+                        match reader.read(&mut buf[*filled..]) {
+                            Ok(0) => {
+                                return Err(io::Error::new(
+                                    io::ErrorKind::UnexpectedEof,
+                                    "EOF inside payload",
+                                ));
+                            }
+                            Ok(n) => *filled += n,
+                            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
+                            Err(e) => return Err(e),
+                        }
+                    }
+                    let payload = std::mem::take(buf);
+                    self.state = AccumState::Prefix {
+                        buf: [0u8; 4],
+                        filled: 0,
+                    };
+                    return Ok(Some(payload));
+                }
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -109,4 +213,125 @@ mod tests {
         let result = read_framed(&mut cursor).expect("read failed");
         assert_eq!(result, None);
     }
+
+    /// Scripted reader for FrameAccumulator tests: each `Some(bytes)` step is
+    /// delivered (possibly partially) by `read`; each `None` step yields one
+    /// WouldBlock; an exhausted script yields EOF.
+    struct ScriptedReader {
+        steps: std::collections::VecDeque<Option<Vec<u8>>>,
+    }
+
+    impl ScriptedReader {
+        fn new(steps: Vec<Option<Vec<u8>>>) -> Self {
+            Self {
+                steps: steps.into(),
+            }
+        }
+    }
+
+    impl Read for ScriptedReader {
+        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+            match self.steps.pop_front() {
+                Some(Some(mut bytes)) => {
+                    let n = bytes.len().min(buf.len());
+                    buf[..n].copy_from_slice(&bytes[..n]);
+                    if n < bytes.len() {
+                        self.steps.push_front(Some(bytes.split_off(n)));
+                    }
+                    Ok(n)
+                }
+                Some(None) => Err(io::Error::new(io::ErrorKind::WouldBlock, "no data")),
+                None => Ok(0), // EOF
+            }
+        }
+    }
+
+    fn frame_bytes(payload: &[u8]) -> Vec<u8> {
+        let mut frame = Vec::new();
+        write_framed(&mut frame, payload).expect("write failed");
+        frame
+    }
+
+    #[test]
+    fn accumulator_survives_split_inside_prefix() {
+        let frame = frame_bytes(b"split prefix");
+        let mut reader = ScriptedReader::new(vec![
+            Some(frame[..2].to_vec()), // half the length prefix
+            None,                      // WouldBlock mid-prefix
+            Some(frame[2..].to_vec()),
+        ]);
+
+        let mut accum = FrameAccumulator::new();
+        let err = accum.poll_frame(&mut reader).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::WouldBlock);
+
+        let payload = accum
+            .poll_frame(&mut reader)
+            .expect("read failed")
+            .expect("expected frame");
+        assert_eq!(payload, b"split prefix");
+    }
+
+    #[test]
+    fn accumulator_survives_split_inside_payload() {
+        let frame = frame_bytes(b"split payload");
+        let mut reader = ScriptedReader::new(vec![
+            Some(frame[..7].to_vec()), // prefix + 3 payload bytes
+            None,                      // WouldBlock mid-payload
+            Some(frame[7..].to_vec()),
+        ]);
+
+        let mut accum = FrameAccumulator::new();
+        let err = accum.poll_frame(&mut reader).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::WouldBlock);
+
+        let payload = accum
+            .poll_frame(&mut reader)
+            .expect("read failed")
+            .expect("expected frame");
+        assert_eq!(payload, b"split payload");
+    }
+
+    #[test]
+    fn accumulator_reads_consecutive_frames_from_one_chunk() {
+        let mut bytes = frame_bytes(b"first");
+        bytes.extend_from_slice(&frame_bytes(b"second"));
+        let mut reader = ScriptedReader::new(vec![Some(bytes)]);
+
+        let mut accum = FrameAccumulator::new();
+        let first = accum.poll_frame(&mut reader).unwrap().unwrap();
+        let second = accum.poll_frame(&mut reader).unwrap().unwrap();
+        assert_eq!(first, b"first");
+        assert_eq!(second, b"second");
+    }
+
+    #[test]
+    fn accumulator_clean_eof_at_frame_boundary_returns_none() {
+        let frame = frame_bytes(b"only");
+        let mut reader = ScriptedReader::new(vec![Some(frame)]);
+
+        let mut accum = FrameAccumulator::new();
+        assert_eq!(accum.poll_frame(&mut reader).unwrap().unwrap(), b"only");
+        assert_eq!(accum.poll_frame(&mut reader).unwrap(), None);
+    }
+
+    #[test]
+    fn accumulator_eof_mid_frame_is_error() {
+        let frame = frame_bytes(b"truncated");
+        let mut reader = ScriptedReader::new(vec![Some(frame[..6].to_vec())]);
+
+        let mut accum = FrameAccumulator::new();
+        let err = accum.poll_frame(&mut reader).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof);
+    }
+
+    #[test]
+    fn accumulator_rejects_oversized_message() {
+        let prefix = (MAX_MESSAGE_SIZE + 1).to_be_bytes().to_vec();
+        let mut reader = ScriptedReader::new(vec![Some(prefix)]);
+
+        let mut accum = FrameAccumulator::new();
+        let err = accum.poll_frame(&mut reader).unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+    }
 }

server/src/bridge/mod.rs

@@ -89,8 +89,11 @@ pub trait SimBridge: Send + Sync {
     /// 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.
+    /// Receive one inbound message, or `None` if no complete frame is ready.
     /// The single client→server stream is demuxed by frame shape (D-225).
+    /// `receive_bridge_inputs` loops this until `None` (T-1045), so the
+    /// transport behind `BridgeResource` must not block when no frame is
+    /// buffered (TcpBridge is non-blocking; LocalBridge blocks — test-only).
     fn receive(&self) -> Result<Option<Inbound>, BridgeError>;
 
     /// Send an atlas layer-stream response to the client (#969, D-225).
@@ -144,7 +147,15 @@ pub enum HandshakeState {
     Complete,
 }
 
+/// Per-tick cap on drained inbound frames (T-1045) — a safety valve so a
+/// client flooding the stream cannot starve the simulation tick. Generous:
+/// normal traffic is one input batch plus the occasional atlas request.
+const MAX_INBOUND_FRAMES_PER_TICK: usize = 64;
+
 /// Receive inputs from bridge and push to InputQueue.
+/// Drains every complete frame buffered this tick (T-1045) — a single
+/// receive() per tick would backlog mixed input/atlas traffic at one frame
+/// per 50 ms. Relies on receive() being non-blocking (Ok(None) = no frame).
 /// 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(
@@ -159,64 +170,77 @@ pub fn receive_bridge_inputs(
     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 _ in 0..MAX_INBOUND_FRAMES_PER_TICK {
+        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);
+                }
             }
-            for input in &inputs {
-                tracing::trace!(
-                    "Received input: tick={} action={:?}",
-                    input.tick,
-                    input.action
-                );
+            Ok(Some(Inbound::AtlasRequest(req))) => {
+                atlas_requests.0.push(req);
             }
-            for input in inputs {
-                input_queue.push(input);
+            // No complete frame ready — the backlog is drained.
+            Ok(None) => break,
+            Err(BridgeError::Disconnected) => {
+                tracing::info!("Client disconnected, shutting down");
+                running.0 = false;
+                break;
+            }
+            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;
+                break;
+            }
+            Err(BridgeError::MutexPoisoned(ref msg)) => {
+                tracing::error!("Bridge mutex poisoned: {}. Shutting down.", msg);
+                running.0 = false;
+                break;
+            }
+            Err(BridgeError::DeserializationWithDump(ref msg)) => {
+                // Recoverable: skip this frame's input, report to client (#85),
+                // keep draining — the frame was consumed, later ones may be fine.
+                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) => {
+                // Unknown error: log once per tick instead of hammering a
+                // persistently failing stream within one tick. A permanently
+                // corrupt stream (e.g. the oversized-prefix poison state)
+                // therefore logs every tick without escalation — follow-up
+                // ticket covers shutdown-after-N-consecutive-errors.
+                tracing::error!("Bridge receive error: {}", e);
+                break;
             }
-        }
-        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);
         }
     }
 }

server/src/bridge/tcp.rs

@@ -5,15 +5,33 @@
 
 use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge};
 use crate::atlas::layer_proxy::AtlasLayerResponse;
-use crate::bridge::framing::{read_framed, write_framed};
-use std::io::{BufReader, BufWriter};
+use crate::bridge::framing::{read_framed, write_framed, FrameAccumulator};
+use std::io::BufWriter;
 use std::net::{SocketAddr, TcpListener, TcpStream};
 use std::sync::Mutex;
 
+/// Read half of the bridge: the raw stream plus the partial-frame accumulator.
+/// The accumulator must persist across `receive()` calls (T-1045) — the socket
+/// is non-blocking, so a frame split across TCP segments is reassembled over
+/// multiple ticks instead of desyncing the stream.
+struct ReadHalf {
+    stream: TcpStream,
+    accum: FrameAccumulator,
+}
+
+impl ReadHalf {
+    fn new(stream: TcpStream) -> Self {
+        Self {
+            stream,
+            accum: FrameAccumulator::new(),
+        }
+    }
+}
+
 /// TcpBridge: TCP transport for client-server IPC
 /// Same semantics as LocalBridge but over TCP localhost
 pub struct TcpBridge {
-    reader: Mutex<BufReader<TcpStream>>,
+    reader: Mutex<ReadHalf>,
     writer: Mutex<BufWriter<TcpStream>>,
     local_addr: SocketAddr,
 }
@@ -45,8 +63,9 @@ impl TcpBridge {
             local_addr
         );
 
-        // Set non-blocking so receive_inputs doesn't stall the game loop.
-        // receive_inputs catches WouldBlock from read_framed and returns Ok(vec![]).
+        // Set non-blocking so receive() doesn't stall the game loop. A
+        // WouldBlock mid-frame is handled by the ReadHalf accumulator;
+        // receive() maps "no complete frame" to Ok(None).
         stream
             .set_nonblocking(true)
             .map_err(|e| BridgeError::Transport(format!("failed to set non-blocking: {}", e)))?;
@@ -57,7 +76,7 @@ impl TcpBridge {
         })?;
 
         Ok(Self {
-            reader: Mutex::new(BufReader::new(reader_stream)),
+            reader: Mutex::new(ReadHalf::new(reader_stream)),
             writer: Mutex::new(BufWriter::new(stream)),
             local_addr,
         })
@@ -91,7 +110,7 @@ impl TcpBridge {
         })?;
 
         Ok(Self {
-            reader: Mutex::new(BufReader::new(reader_stream)),
+            reader: Mutex::new(ReadHalf::new(reader_stream)),
             writer: Mutex::new(BufWriter::new(stream)),
             local_addr,
         })
@@ -117,7 +136,7 @@ impl TcpBridge {
         })?;
 
         Ok(Self {
-            reader: Mutex::new(BufReader::new(reader_stream)),
+            reader: Mutex::new(ReadHalf::new(reader_stream)),
             writer: Mutex::new(BufWriter::new(stream)),
             local_addr,
         })
@@ -137,15 +156,16 @@ impl SimBridge for TcpBridge {
             .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?;
         // Toggle to blocking for reliable startup message read.
         // The client sends StartupMessage immediately after handshake validation,
-        // so this read should complete quickly.
+        // so this read should complete quickly. Runs before the first receive(),
+        // so the frame accumulator is empty and read_framed sees a fresh stream.
         reader
-            .get_mut()
+            .stream
             .set_nonblocking(false)
             .map_err(BridgeError::Io)?;
-        let result = read_framed(reader.get_mut());
+        let result = read_framed(&mut reader.stream);
         // Restore non-blocking for the tick loop
         reader
-            .get_mut()
+            .stream
             .set_nonblocking(true)
             .map_err(BridgeError::Io)?;
         match result? {
@@ -204,11 +224,19 @@ impl SimBridge for TcpBridge {
             .lock()
             .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?;
 
-        match read_framed(reader.get_mut()) {
+        let ReadHalf { stream, accum } = &mut *reader;
+        match accum.poll_frame(stream) {
             Ok(Some(payload)) => decode_inbound(&payload).map(Some),
             Ok(None) => Err(BridgeError::Disconnected),
-            // Non-blocking socket: no data available this tick — not an error.
+            // Non-blocking socket: no complete frame yet — not an error. Any
+            // partial frame stays in the accumulator for the next tick.
             Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(None),
+            // EOF mid-frame: the peer died leaving a truncated stream. No data
+            // can ever complete this frame — escalate like the clean-EOF case
+            // above instead of surfacing an Io error every tick forever.
+            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                Err(BridgeError::Disconnected)
+            }
             Err(e) => Err(BridgeError::Io(e)),
         }
     }