feat(simulation): demux inbound bridge stream — receive() + atlas routing (#969, D-225)

Replaces the fixed-type SimBridge::receive_inputs() with a tagged
receive() -> Option<Inbound>, where Inbound is Inputs(Vec<PlayerInput>) or
AtlasRequest(AtlasLayerRequest). A shared decode_inbound() demuxes a frame by
shape (msgpack array = inputs, map = atlas request) — additive, no wire change
to existing input/snapshot frames. Adds send_atlas_response() to the trait
(both TcpBridge + LocalBridge impls). receive_bridge_inputs routes inputs to
the InputQueue as before; atlas requests to a new AtlasRequestBuffer (drained
by the serve system next). Integration tests (bridge_tcp/bridge_ipc) updated to
the tagged receive(); a demux unit test covers all three branches.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

server/src/bridge/local.rs

@@ -2,7 +2,8 @@
 // Implements D-020 subprocess/IPC architecture
 // Deterministic client-server communication via Unix domain sockets
 
-use super::{BridgeError, ObserverSnapshot, PlayerInput, SimBridge};
+use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge};
+use crate::atlas::layer_proxy::AtlasLayerResponse;
 use crate::bridge::framing::{read_framed, write_framed};
 use std::fs;
 use std::io::{BufReader, BufWriter};
@@ -123,37 +124,27 @@ impl SimBridge for LocalBridge {
         Ok(())
     }
 
-    fn receive_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError> {
+    fn receive(&self) -> Result<Option<Inbound>, 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::<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()
-                    )))
-                }
-            },
+            Some(payload) => decode_inbound(&payload).map(Some),
             None => Err(BridgeError::Disconnected),
         }
     }
+
+    fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> {
+        let payload = rmp_serde::to_vec_named(resp)?;
+        let mut writer = self
+            .writer
+            .lock()
+            .map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?;
+        write_framed(writer.get_mut(), &payload)?;
+        Ok(())
+    }
 }
 
 impl Drop for LocalBridge {

server/src/bridge/mod.rs

@@ -6,6 +6,8 @@ use bevy_app::prelude::*;
 use bevy_ecs::prelude::*;
 use bevy_ecs::schedule::IntoScheduleConfigs;
 
+use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse};
+
 pub mod debug;
 pub mod framing;
 pub mod local;
@@ -33,6 +35,44 @@ pub enum BridgeError {
     MutexPoisoned(String),
 }
 
+/// One decoded inbound message. The client→server stream is a single demuxed
+/// channel (D-225): a `Vec<PlayerInput>` frame is a MessagePack *array* and an
+/// `AtlasLayerRequest` frame is a *map*, so they are distinguishable without a
+/// wire-level type tag (existing frames are byte-unchanged — additive).
+#[derive(Debug)]
+pub enum Inbound {
+    /// A batch of player inputs (the gameplay path).
+    Inputs(Vec<PlayerInput>),
+    /// An atlas layer-stream request (#969, D-225).
+    AtlasRequest(AtlasLayerRequest),
+}
+
+/// Demux a received frame payload into an [`Inbound`] (D-225). Tries
+/// `Vec<PlayerInput>` (array), then `AtlasLayerRequest` (map); a frame that is
+/// neither is a genuinely malformed input frame.
+pub fn decode_inbound(payload: &[u8]) -> Result<Inbound, BridgeError> {
+    if let Ok(inputs) = rmp_serde::from_slice::<Vec<PlayerInput>>(payload) {
+        return Ok(Inbound::Inputs(inputs));
+    }
+    match rmp_serde::from_slice::<AtlasLayerRequest>(payload) {
+        Ok(req) => Ok(Inbound::AtlasRequest(req)),
+        Err(e) => {
+            let dump_len = payload.len().min(256);
+            tracing::error!(
+                "inbound decode failed (neither inputs nor atlas request): {}. Raw ({} of {} bytes): {:02x?}",
+                e,
+                dump_len,
+                payload.len(),
+                &payload[..dump_len]
+            );
+            Err(BridgeError::DeserializationWithDump(format!(
+                "{e} (payload {} bytes)",
+                payload.len()
+            )))
+        }
+    }
+}
+
 /// Abstracts transport layer (D-020)
 /// Implemented by LocalBridge (stdio) and future NetworkBridge
 pub trait SimBridge: Send + Sync {
@@ -49,8 +89,12 @@ pub trait SimBridge: Send + Sync {
     /// 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>;
+    /// Receive one inbound message, or `None` if no frame is ready this tick.
+    /// The single client→server stream is demuxed by frame shape (D-225).
+    fn receive(&self) -> Result<Option<Inbound>, BridgeError>;
+
+    /// Send an atlas layer-stream response to the client (#969, D-225).
+    fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError>;
 }
 
 /// BridgeResource: Bevy Resource wrapper for SimBridge trait object
@@ -78,8 +122,12 @@ impl BridgeResource {
         self.inner.send_snapshot(snapshot)
     }
 
-    pub fn receive_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError> {
-        self.inner.receive_inputs()
+    pub fn receive(&self) -> Result<Option<Inbound>, BridgeError> {
+        self.inner.receive()
+    }
+
+    pub fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> {
+        self.inner.send_atlas_response(resp)
     }
 }
 
@@ -105,13 +153,14 @@ pub fn receive_bridge_inputs(
     mut running: ResMut<ServerRunning>,
     handshake: Res<HandshakeState>,
     mut error_buffer: ResMut<SimErrorBuffer>,
+    mut atlas_requests: ResMut<AtlasRequestBuffer>,
     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) => {
+    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)",
@@ -129,6 +178,10 @@ pub fn receive_bridge_inputs(
                 input_queue.push(input);
             }
         }
+        Ok(Some(Inbound::AtlasRequest(req))) => {
+            atlas_requests.0.push(req);
+        }
+        Ok(None) => {}
         Err(BridgeError::Disconnected) => {
             tracing::info!("Client disconnected, shutting down");
             running.0 = false;
@@ -207,6 +260,11 @@ impl Default for ServerRunning {
     }
 }
 
+/// Inbound atlas layer requests routed off the bridge (#969, D-225), drained by
+/// the proxy serve system in `PreInput`.
+#[derive(Resource, Default)]
+pub struct AtlasRequestBuffer(pub Vec<AtlasLayerRequest>);
+
 /// Bridge plugin for client-server communication
 /// Abstracts transport layer (LocalBridge/NetworkBridge)
 pub struct BridgePlugin;
@@ -222,6 +280,7 @@ impl Plugin for BridgePlugin {
             .init_resource::<debug::DebugCommandBuffer>()
             .init_resource::<DebugEnabled>()
             .init_resource::<crate::perception::query::ActivePerceptionMode>()
+            .init_resource::<AtlasRequestBuffer>()
             // 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))
@@ -274,3 +333,30 @@ impl Plugin for BridgePlugin {
         tracing::debug!("BridgePlugin initialized");
     }
 }
+
+#[cfg(test)]
+mod inbound_tests {
+    use super::*;
+    use crate::atlas::cascade::CascadeLayer;
+
+    #[test]
+    fn demux_routes_inputs_and_atlas_requests() {
+        // A Vec<PlayerInput> frame (msgpack array) → Inbound::Inputs.
+        let inputs: Vec<PlayerInput> = vec![];
+        let frame = rmp_serde::to_vec_named(&inputs).unwrap();
+        assert!(matches!(decode_inbound(&frame), Ok(Inbound::Inputs(v)) if v.is_empty()));
+
+        // An AtlasLayerRequest frame (msgpack map) → Inbound::AtlasRequest.
+        let req = AtlasLayerRequest {
+            body_id: "GJ1c".into(),
+            up_to: CascadeLayer::Topography,
+        };
+        let frame = rmp_serde::to_vec_named(&req).unwrap();
+        assert!(
+            matches!(decode_inbound(&frame), Ok(Inbound::AtlasRequest(r)) if r.body_id == "GJ1c")
+        );
+
+        // Neither shape → a malformed-frame error.
+        assert!(decode_inbound(&[0xff, 0xff]).is_err());
+    }
+}

server/src/bridge/tcp.rs

@@ -3,7 +3,8 @@
 // Deterministic client-server communication via TCP sockets
 // Used for Godot client which lacks Unix socket support
 
-use super::{BridgeError, ObserverSnapshot, PlayerInput, SimBridge};
+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 std::net::{SocketAddr, TcpListener, TcpStream};
@@ -197,38 +198,33 @@ impl SimBridge for TcpBridge {
         Ok(())
     }
 
-    fn receive_inputs(&self) -> Result<Vec<PlayerInput>, BridgeError> {
+    fn receive(&self) -> Result<Option<Inbound>, BridgeError> {
         let mut reader = self
             .reader
             .lock()
             .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?;
 
         match read_framed(reader.get_mut()) {
-            Ok(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()
-                    )))
-                }
-            },
+            Ok(Some(payload)) => decode_inbound(&payload).map(Some),
             Ok(None) => Err(BridgeError::Disconnected),
             // Non-blocking socket: no data available this tick — not an error.
-            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(vec![]),
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(None),
             Err(e) => Err(BridgeError::Io(e)),
         }
     }
+
+    fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> {
+        let payload = rmp_serde::to_vec_named(resp)?;
+        let mut writer = self
+            .writer
+            .lock()
+            .map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?;
+        // Toggle to blocking for reliable write delivery (same as send_snapshot).
+        let stream = writer.get_mut();
+        stream.set_nonblocking(false).map_err(BridgeError::Io)?;
+        let result = write_framed(stream, &payload);
+        stream.set_nonblocking(true).map_err(BridgeError::Io)?;
+        result?;
+        Ok(())
+    }
 }

server/tests/bridge_ipc.rs

@@ -3,7 +3,7 @@
 use settled_reach_server::bridge::framing::{read_framed, write_framed};
 use settled_reach_server::bridge::local::LocalBridge;
 use settled_reach_server::bridge::types::*;
-use settled_reach_server::bridge::SimBridge;
+use settled_reach_server::bridge::{Inbound, SimBridge};
 use settled_reach_server::simulation::time::{DayPhase, TickRate};
 use std::os::unix::net::UnixStream;
 use std::path::PathBuf;
@@ -116,7 +116,10 @@ fn input_roundtrip_over_unix_socket() {
     let server_handle = thread::spawn(move || {
         let bridge = LocalBridge::accept(&server_path).expect("failed to accept");
 
-        let inputs = bridge.receive_inputs().expect("failed to receive inputs");
+        let inputs = match bridge.receive().expect("failed to receive") {
+            Some(Inbound::Inputs(inputs)) => inputs,
+            other => panic!("expected inputs, got {other:?}"),
+        };
 
         assert_eq!(inputs.len(), 2);
         assert_eq!(inputs[0].tick, 10);

server/tests/bridge_tcp.rs

@@ -3,7 +3,7 @@
 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 settled_reach_server::bridge::{Inbound, SimBridge};
 use settled_reach_server::simulation::time::{DayPhase, TickRate};
 use std::net::{TcpListener, TcpStream};
 use std::thread;
@@ -101,8 +101,8 @@ fn input_roundtrip_over_tcp() {
         // Non-blocking socket: retry until data arrives or timeout.
         let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
         let inputs = loop {
-            match bridge.receive_inputs() {
-                Ok(inputs) if !inputs.is_empty() => break inputs,
+            match bridge.receive() {
+                Ok(Some(Inbound::Inputs(inputs))) if !inputs.is_empty() => break inputs,
                 Ok(_) => {
                     assert!(
                         std::time::Instant::now() < deadline,
@@ -168,16 +168,16 @@ fn tcp_bridge_eof_returns_error() {
         // Non-blocking socket: retry until we get Disconnected or timeout.
         let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
         loop {
-            match bridge.receive_inputs() {
-                Ok(inputs) if inputs.is_empty() => {
-                    // WouldBlock — client hasn't disconnected yet, retry
+            match bridge.receive() {
+                Ok(None) => {
+                    // No data yet — client hasn't disconnected, retry
                     assert!(
                         std::time::Instant::now() < deadline,
                         "timed out waiting for EOF"
                     );
                     thread::sleep(std::time::Duration::from_millis(1));
                 }
-                Ok(inputs) => panic!("expected Disconnected error, got {} inputs", inputs.len()),
+                Ok(Some(_)) => panic!("expected Disconnected error, got a message"),
                 Err(settled_reach_server::bridge::BridgeError::Disconnected) => break,
                 Err(e) => panic!("expected Disconnected error, got: {}", e),
             }