feat(engine): T-1130 reader connections — role-gated multi-connection bridge (D-254 SS1/SS2)

ConnectionRole (Player|Reader, serde-default Player for wire back-compat;
shaped for a future TradingReader) on StartupMessage. BridgeResource
rewritten as 0-1 Player + 0-N Readers with ConnectionId; per-tick
accept_new_connections loop replaces the single blocking accept (the
listener is cloned non-blocking into ConnectionListener). First
connection installs per startup.role — spawn-mode Readers are often the
only connection a server gets.

Permitted-message matrix: readers may handshake and issue
Atlas/StarMap/CityNames requests (responses connection-tagged, own
requests only); PlayerInput from a reader is dropped with a strike
(disconnect at 3); ObserverSnapshot has no reader-facing path at all.
Shutdown scope: only a PLAYER send failure flips ServerRunning —
reader-only servers idle with the snapshot queued; reader disconnects
never kill the session.

Deliberately out of scope, documented at the call site: character spawn
stays in monolithic world-setup (reader-first servers carry an inert
unpiloted PlayerCharacter); no total-reader-count cap (per-reader frame
cap only, loopback-only scope).

6 new bridge_tcp integration tests; layer3 subprocess test exercises the
accept-loop end to end.

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

server/src/atlas/plugin.rs

@@ -94,7 +94,11 @@ fn serve_atlas_requests(
     let reader = city_reader.as_ref().map(|r| &r.0);
     let params_reader = body_params_reader.as_ref().map(|r| &r.0);
     let pending: Vec<_> = requests.0.drain(..).collect();
-    for req in pending {
+    // D-254 §2: 1:1, in-order request->response — the connection id rides
+    // alongside the request untouched by handle_atlas_request (which has no
+    // notion of connections) and is re-attached to the response so the
+    // bridge's send_atlas_responses routes it back to only that connection.
+    for (conn_id, req) in pending {
         let resp = match resolver.as_ref() {
             Some(r) => handle_atlas_request(
                 &req,
@@ -116,7 +120,7 @@ fn serve_atlas_requests(
                 region_grid: None,
             },
         };
-        responses.0.push(resp);
+        responses.0.push((conn_id, resp));
     }
 }
 
@@ -133,7 +137,7 @@ fn serve_star_map_requests(
         return;
     }
     let pending: Vec<_> = requests.0.drain(..).collect();
-    for req in pending {
+    for (conn_id, req) in pending {
         let resp = match path.as_ref() {
             Some(p) => handle_star_map_request(&req, &p.0),
             None => crate::atlas::atlas_data_proxy::StarMapResponse {
@@ -143,7 +147,7 @@ fn serve_star_map_requests(
                 data: None,
             },
         };
-        responses.0.push(resp);
+        responses.0.push((conn_id, resp));
     }
 }
 
@@ -159,8 +163,10 @@ fn serve_city_names_requests(
     }
     let reader = city_reader.as_ref().map(|r| &r.0);
     let pending: Vec<_> = requests.0.drain(..).collect();
-    for req in pending {
-        responses.0.push(handle_city_names_request(&req, reader));
+    for (conn_id, req) in pending {
+        responses
+            .0
+            .push((conn_id, handle_city_names_request(&req, reader)));
     }
 }
 
@@ -760,6 +766,7 @@ mod tests {
     use super::*;
     use crate::atlas::gen_queue::{GenPriority, GenWorkItem};
     use crate::atlas::road_graph::{RoadEdge, RoadNode, RoadNodeKind};
+    use crate::bridge::ConnectionId;
     use crate::seed::SeedChain;
     use crate::simulation::generator::{
         ArrangementPattern, AttractorType, FoundingOrientation, MaintenanceAuthority,
@@ -834,10 +841,13 @@ mod tests {
         use crate::atlas::layer_proxy::AtlasLayerRequest;
 
         let mut world = World::new();
-        world.insert_resource(AtlasRequestBuffer(vec![AtlasLayerRequest {
-            body_id: "GJ1c".to_string(),
-            up_to: CascadeLayer::Topography,
-        }]));
+        world.insert_resource(AtlasRequestBuffer(vec![(
+            ConnectionId(0),
+            AtlasLayerRequest {
+                body_id: "GJ1c".to_string(),
+                up_to: CascadeLayer::Topography,
+            },
+        )]));
         world.insert_resource(AtlasResponseBuffer::default());
         world.insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY));
         world.insert_resource(GenerationQueue::with_threads(1));
@@ -849,9 +859,13 @@ mod tests {
 
         let responses = world.resource::<AtlasResponseBuffer>();
         assert_eq!(responses.0.len(), 1, "request should produce one response");
-        assert_eq!(responses.0[0].body_id, "GJ1c");
+        assert_eq!(responses.0[0].0, ConnectionId(0), "connection id preserved");
+        assert_eq!(responses.0[0].1.body_id, "GJ1c");
         // No resolver wired → Error status (exercises the drain + push path).
-        assert!(matches!(responses.0[0].status, AtlasLayerStatus::Error(_)));
+        assert!(matches!(
+            responses.0[0].1.status,
+            AtlasLayerStatus::Error(_)
+        ));
         // The request buffer was drained.
         assert!(world.resource::<AtlasRequestBuffer>().0.is_empty());
     }
@@ -869,9 +883,10 @@ mod tests {
         std::fs::write(&path, r#"{"_meta": {}, "nodes": [], "edges": []}"#).unwrap();
 
         let mut world = World::new();
-        world.insert_resource(StarMapRequestBuffer(vec![StarMapRequest {
-            star_map: true,
-        }]));
+        world.insert_resource(StarMapRequestBuffer(vec![(
+            ConnectionId(0),
+            StarMapRequest { star_map: true },
+        )]));
         world.insert_resource(StarMapResponseBuffer::default());
         world.insert_resource(StarMapDataPath(path.clone()));
 
@@ -881,7 +896,8 @@ mod tests {
 
         let responses = world.resource::<StarMapResponseBuffer>();
         assert_eq!(responses.0.len(), 1);
-        assert_eq!(responses.0[0].status, StarMapStatus::Ready);
+        assert_eq!(responses.0[0].0, ConnectionId(0), "connection id preserved");
+        assert_eq!(responses.0[0].1.status, StarMapStatus::Ready);
         assert!(world.resource::<StarMapRequestBuffer>().0.is_empty());
 
         let _ = std::fs::remove_file(&path);
@@ -894,9 +910,10 @@ mod tests {
         use crate::atlas::atlas_data_proxy::{StarMapRequest, StarMapStatus};
 
         let mut world = World::new();
-        world.insert_resource(StarMapRequestBuffer(vec![StarMapRequest {
-            star_map: true,
-        }]));
+        world.insert_resource(StarMapRequestBuffer(vec![(
+            ConnectionId(0),
+            StarMapRequest { star_map: true },
+        )]));
         world.insert_resource(StarMapResponseBuffer::default());
         // No StarMapDataPath resource.
 
@@ -906,7 +923,7 @@ mod tests {
 
         let responses = world.resource::<StarMapResponseBuffer>();
         assert_eq!(responses.0.len(), 1);
-        assert!(matches!(responses.0[0].status, StarMapStatus::Error(_)));
+        assert!(matches!(responses.0[0].1.status, StarMapStatus::Error(_)));
     }
 
     /// T-949b: without `CityContextReaderResource` wired, the serve system
@@ -917,10 +934,13 @@ mod tests {
         use crate::atlas::atlas_data_proxy::{CityNamesRequest, CityNamesStatus};
 
         let mut world = World::new();
-        world.insert_resource(CityNamesRequestBuffer(vec![CityNamesRequest {
-            city_names: true,
-            body_id: "GJ1c".to_string(),
-        }]));
+        world.insert_resource(CityNamesRequestBuffer(vec![(
+            ConnectionId(0),
+            CityNamesRequest {
+                city_names: true,
+                body_id: "GJ1c".to_string(),
+            },
+        )]));
         world.insert_resource(CityNamesResponseBuffer::default());
         // No CityContextReaderResource.
 
@@ -930,8 +950,9 @@ mod tests {
 
         let responses = world.resource::<CityNamesResponseBuffer>();
         assert_eq!(responses.0.len(), 1);
-        assert_eq!(responses.0[0].body_id, "GJ1c");
-        assert!(matches!(responses.0[0].status, CityNamesStatus::Error(_)));
+        assert_eq!(responses.0[0].0, ConnectionId(0), "connection id preserved");
+        assert_eq!(responses.0[0].1.body_id, "GJ1c");
+        assert!(matches!(responses.0[0].1.status, CityNamesStatus::Error(_)));
         assert!(world.resource::<CityNamesRequestBuffer>().0.is_empty());
     }
 

server/src/bridge/mod.rs

@@ -10,6 +10,7 @@ use crate::atlas::atlas_data_proxy::{
     CityNamesRequest, CityNamesResponse, StarMapRequest, StarMapResponse,
 };
 use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse};
+use crate::bridge::tcp::TcpBridge;
 
 pub mod debug;
 pub mod framing;
@@ -190,45 +191,247 @@ pub trait SimBridge: Send + Sync {
     fn send_city_names_response(&self, resp: &CityNamesResponse) -> Result<(), BridgeError>;
 }
 
-/// BridgeResource: Bevy Resource wrapper for SimBridge trait object
-#[derive(Resource)]
+/// Identifies one connection for response-tagging and role-lookup purposes
+/// (D-254 §2, T-1130). Assigned at accept time by `BridgeResource`; never
+/// reused within a server process lifetime (monotonic counter), so a stale
+/// id from a disconnected connection can never collide with a live one.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ConnectionId(pub u64);
+
+/// One live connection: its transport, role, and (for readers) a violation
+/// strike counter (D-254 §2 — "log + drop on first offense, disconnect on
+/// repeated violations").
+struct Connection {
+    id: ConnectionId,
+    bridge: Box<dyn SimBridge>,
+    /// Not read today — `player`/`readers` (which list a `Connection` lives
+    /// in) already fully determines role-gated behavior in this ticket's
+    /// scope. Kept because D-254 §6's future `TradingReader` widening wants
+    /// a role-keyed per-verb match (`Player | Reader | TradingReader`) on
+    /// exactly this field rather than a third top-level `Vec` — carrying it
+    /// now avoids a struct-shape change alongside that later widening.
+    #[allow(dead_code)]
+    role: ConnectionRole,
+    /// Count of role-violating frames seen from this connection (e.g. a
+    /// Reader sending `Vec<PlayerInput>`). Always 0 for `Player` — the
+    /// violation path only applies to non-Player roles.
+    violation_strikes: u32,
+}
+
+/// Strikes tolerated before a violating reader is disconnected (D-254 §2).
+/// First offense logs + drops the frame; this is the ceiling before the
+/// connection itself is torn down. Deliberately small — a well-behaved
+/// reader client sends zero disallowed frames ever, so any nonzero count is
+/// already a bug or hostile probe, not normal traffic.
+const READER_VIOLATION_DISCONNECT_THRESHOLD: u32 = 3;
+
+/// Per-reader per-tick inbound frame cap (D-254 §2) — lower than the Player
+/// cap (`MAX_INBOUND_FRAMES_PER_TICK`, 64): a reader has no legitimate reason
+/// to send that volume of atlas/star-map/city-names requests in one 50ms
+/// tick. Cheap insurance against a runaway or misbehaving companion client;
+/// does not affect determinism either way (reader frames never reach
+/// InputQueue/SimRng regardless of how many are drained).
+const MAX_READER_INBOUND_FRAMES_PER_TICK: usize = 8;
+
+/// BridgeResource: Bevy Resource holding the server's connection set
+/// (D-254 §2, T-1130).
+///
+/// Scoped honestly as **0-1 Player + 0-N Readers** — not general N-player
+/// (D-009's separate, larger, out-of-scope ambition). `player` is the
+/// original single connection this resource used to wrap directly;
+/// `readers` is new. `pending` holds accepted-but-not-yet-handshaken
+/// connections (see `tcp::PendingConnection`) — polled non-blockingly each
+/// tick by `main.rs`'s accept-loop system until they either promote into
+/// `readers` (or, in the anomalous case of a second Player attempt, get
+/// cleanly rejected — see `main.rs`) or fail and are dropped.
+#[derive(Resource, Default)]
 pub struct BridgeResource {
-    inner: Box<dyn SimBridge>,
+    player: Option<Connection>,
+    readers: Vec<Connection>,
+    next_id: u64,
 }
 
 impl BridgeResource {
+    /// Construct a `BridgeResource` with `bridge` installed as the Player
+    /// connection. This is the pre-D-254 constructor signature, preserved
+    /// byte-for-byte so every existing call site (`main.rs`, and the
+    /// single-connection test suites in `server/tests/`) needs no change —
+    /// "install this bridge" always meant "install the Player" before
+    /// readers existed, and still does when called this way.
     pub fn new(bridge: impl SimBridge + 'static) -> Self {
-        Self {
-            inner: Box::new(bridge),
+        let mut resource = Self::default();
+        resource.insert_player(bridge);
+        resource
+    }
+
+    /// Install `bridge` as the Player connection, assigning it the next
+    /// `ConnectionId`. Overwrites any existing Player connection (there is
+    /// never more than one — `main.rs`'s accept-loop rejects a second Player
+    /// attempt before calling this).
+    pub fn insert_player(&mut self, bridge: impl SimBridge + 'static) -> ConnectionId {
+        let id = ConnectionId(self.next_id);
+        self.next_id += 1;
+        self.player = Some(Connection {
+            id,
+            bridge: Box::new(bridge),
+            role: ConnectionRole::Player,
+            violation_strikes: 0,
+        });
+        id
+    }
+
+    /// Install `bridge` as a new Reader connection, assigning it the next
+    /// `ConnectionId`.
+    pub fn insert_reader(&mut self, bridge: impl SimBridge + 'static) -> ConnectionId {
+        let id = ConnectionId(self.next_id);
+        self.next_id += 1;
+        self.readers.push(Connection {
+            id,
+            bridge: Box::new(bridge),
+            role: ConnectionRole::Reader,
+            violation_strikes: 0,
+        });
+        id
+    }
+
+    /// True if a Player connection is currently installed (D-254 §2: used by
+    /// the accept-loop to cleanly reject a second Player attempt instead of
+    /// silently starving it the way the pre-D-254 single-`accept()` did).
+    pub fn has_player(&self) -> bool {
+        self.player.is_some()
+    }
+
+    /// Number of currently-installed Reader connections (test/observability
+    /// helper, D-254 §2 — the 0-N reader count this ticket's scope is built
+    /// around).
+    pub fn reader_count(&self) -> usize {
+        self.readers.len()
+    }
+
+    /// `ConnectionId`s of every currently-installed reader, in insertion
+    /// order (test/observability helper, D-254 §2).
+    pub fn reader_ids(&self) -> Vec<ConnectionId> {
+        self.readers.iter().map(|c| c.id).collect()
+    }
+
+    /// Remove and drop the Player connection (its `TcpBridge`/socket is
+    /// dropped, closing the TCP connection).
+    pub fn remove_player(&mut self) {
+        self.player = None;
+    }
+
+    /// Remove and drop the reader with the given id, if present. A no-op if
+    /// the id doesn't match any current reader (already removed, or was
+    /// never a reader — e.g. the Player's own id).
+    pub fn remove_reader(&mut self, id: ConnectionId) {
+        self.readers.retain(|c| c.id != id);
+    }
+
+    /// Send the protocol handshake on the Player connection, if any.
+    /// Used only by `main.rs`'s original single-connection startup path —
+    /// reader connections send their own handshake as part of
+    /// `tcp::PendingConnection`'s state machine, not through this resource.
+    pub fn send_handshake(&self) -> Result<(), BridgeError> {
+        match &self.player {
+            Some(c) => c.bridge.send_handshake(),
+            None => Err(BridgeError::Transport("no player connection".into())),
         }
     }
 
-    pub fn send_handshake(&self) -> Result<(), BridgeError> {
-        self.inner.send_handshake()
-    }
-
+    /// Receive the Player's startup message. Used only by `main.rs`'s
+    /// original single-connection startup path (the first accept, before
+    /// the tick loop and its non-blocking accept-loop begin).
     pub fn receive_startup(&self) -> Result<StartupMessage, BridgeError> {
-        self.inner.receive_startup()
+        match &self.player {
+            Some(c) => c.bridge.receive_startup(),
+            None => Err(BridgeError::Transport("no player connection".into())),
+        }
     }
 
+    /// Send an `ObserverSnapshot` to the Player connection ONLY (D-254 §2 —
+    /// structural enforcement: this method has no reader-facing counterpart
+    /// at all, so a future edit cannot accidentally start broadcasting
+    /// snapshots to readers by forgetting a role check — there is no code
+    /// path here that could reach a reader's `bridge.send_snapshot`).
     pub fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError> {
-        self.inner.send_snapshot(snapshot)
+        match &self.player {
+            Some(c) => c.bridge.send_snapshot(snapshot),
+            None => Err(BridgeError::Disconnected),
+        }
     }
 
-    pub fn receive(&self) -> Result<Option<Inbound>, BridgeError> {
-        self.inner.receive()
+    /// Look up the connection (Player or Reader) matching `id`, for
+    /// per-connection response routing (D-254 §2: atlas/star-map/city-names
+    /// responses must go back to only the connection that asked).
+    fn connection(&self, id: ConnectionId) -> Option<&Connection> {
+        if let Some(p) = &self.player {
+            if p.id == id {
+                return Some(p);
+            }
+        }
+        self.readers.iter().find(|c| c.id == id)
     }
 
-    pub fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError> {
-        self.inner.send_atlas_response(resp)
+    /// Send an atlas layer-stream response to exactly the connection that
+    /// requested it (#969, D-225 original; D-254 §2 adds the routing — the
+    /// connection may be the Player or any Reader). `Ok(())` with a debug
+    /// log if the connection has since disconnected (the response is simply
+    /// dropped — not an error condition, the requester is gone).
+    pub fn send_atlas_response_to(
+        &self,
+        id: ConnectionId,
+        resp: &AtlasLayerResponse,
+    ) -> Result<(), BridgeError> {
+        match self.connection(id) {
+            Some(c) => c.bridge.send_atlas_response(resp),
+            None => {
+                tracing::debug!(
+                    "atlas response for {:?} dropped — connection {:?} no longer present",
+                    resp.body_id,
+                    id
+                );
+                Ok(())
+            }
+        }
     }
 
-    pub fn send_star_map_response(&self, resp: &StarMapResponse) -> Result<(), BridgeError> {
-        self.inner.send_star_map_response(resp)
+    /// Send a star-map response to exactly the connection that requested it
+    /// (T-949a original; D-254 §2 adds the routing).
+    pub fn send_star_map_response_to(
+        &self,
+        id: ConnectionId,
+        resp: &StarMapResponse,
+    ) -> Result<(), BridgeError> {
+        match self.connection(id) {
+            Some(c) => c.bridge.send_star_map_response(resp),
+            None => {
+                tracing::debug!(
+                    "star map response dropped — connection {:?} no longer present",
+                    id
+                );
+                Ok(())
+            }
+        }
     }
 
-    pub fn send_city_names_response(&self, resp: &CityNamesResponse) -> Result<(), BridgeError> {
-        self.inner.send_city_names_response(resp)
+    /// Send a city-names response to exactly the connection that requested
+    /// it (T-949b original; D-254 §2 adds the routing).
+    pub fn send_city_names_response_to(
+        &self,
+        id: ConnectionId,
+        resp: &CityNamesResponse,
+    ) -> Result<(), BridgeError> {
+        match self.connection(id) {
+            Some(c) => c.bridge.send_city_names_response(resp),
+            None => {
+                tracing::debug!(
+                    "city names response for {:?} dropped — connection {:?} no longer present",
+                    resp.body_id,
+                    id
+                );
+                Ok(())
+            }
+        }
     }
 }
 
@@ -250,14 +453,30 @@ pub enum HandshakeState {
 /// 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).
+/// Receive inputs/requests from every connection and route by role (D-254
+/// §2, T-1130). Drains every complete frame buffered this tick per
+/// connection (T-1045) — a single receive() per connection per tick would
+/// backlog mixed input/atlas traffic at one frame per 50 ms. Relies on
+/// receive() being non-blocking (Ok(None) = no frame).
+///
+/// Role gate (D-254 §2 permitted-message matrix): `Vec<PlayerInput>` is only
+/// ever pushed to `InputQueue` from the Player connection. A Reader sending
+/// inputs is syntactically valid (the D-225 demux parses it fine) but
+/// role-disallowed — logged, dropped, and struck; `READER_VIOLATION_
+/// DISCONNECT_THRESHOLD` repeated violations disconnect that reader (never
+/// the Player, never other readers). Atlas/star-map/city-names requests are
+/// accepted from ANY connection and tagged with the sender's `ConnectionId`
+/// so `send_*_responses` can route the reply back to only that connection.
+///
+/// THE CRITICAL FIX (D-254 §2): only the Player connection's disconnect
+/// flips `ServerRunning` — a reader disconnecting (or never having
+/// connected) must never affect a running player session. Reader
+/// disconnects just remove that reader from `BridgeResource` and continue.
+///
 /// 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(
-    bridge: Option<Res<BridgeResource>>,
+    bridge: Option<ResMut<BridgeResource>>,
     mut input_queue: ResMut<crate::simulation::input::InputQueue>,
     mut running: ResMut<ServerRunning>,
     handshake: Res<HandshakeState>,
@@ -267,115 +486,266 @@ pub fn receive_bridge_inputs(
     mut city_names_requests: ResMut<CityNamesRequestBuffer>,
     time: Option<Res<crate::simulation::time::SimulationTime>>,
 ) {
-    let Some(bridge) = bridge else { return };
+    let Some(mut bridge) = bridge else { return };
     let current_tick = time.as_ref().map(|t| t.tick).unwrap_or(0);
 
-    for _ in 0..MAX_INBOUND_FRAMES_PER_TICK {
-        match bridge.receive() {
-            Ok(Some(Inbound::Inputs(inputs))) => {
-                if !inputs.is_empty() && *handshake == HandshakeState::Pending {
+    // -- Player connection ------------------------------------------------
+    // Unchanged behavior from before D-254: the Player's Inputs go to
+    // InputQueue, its atlas/etc. requests get tagged with its ConnectionId,
+    // and ITS disconnect (and only its disconnect) shuts the server down.
+    if let Some(player) = bridge.player.as_mut() {
+        let player_id = player.id;
+        let mut player_disconnected = false;
+
+        for _ in 0..MAX_INBOUND_FRAMES_PER_TICK {
+            match player.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);
+                    }
+                }
+                Ok(Some(Inbound::AtlasRequest(req))) => {
+                    atlas_requests.0.push((player_id, req));
+                }
+                Ok(Some(Inbound::StarMapRequest(req))) => {
+                    star_map_requests.0.push((player_id, req));
+                }
+                Ok(Some(Inbound::CityNamesRequest(req))) => {
+                    city_names_requests.0.push((player_id, req));
+                }
+                // No complete frame ready — the backlog is drained.
+                Ok(None) => break,
+                Err(BridgeError::Disconnected) => {
+                    tracing::info!("Player disconnected, shutting down");
+                    running.0 = false;
+                    player_disconnected = true;
+                    break;
+                }
+                Err(BridgeError::Io(ref e))
+                    if e.kind() == std::io::ErrorKind::BrokenPipe
+                        || e.kind() == std::io::ErrorKind::ConnectionReset =>
+                {
+                    tracing::info!("Player pipe broken, shutting down cleanly");
+                    running.0 = false;
+                    player_disconnected = true;
+                    break;
+                }
+                Err(BridgeError::MutexPoisoned(ref msg)) => {
+                    tracing::error!("Player bridge mutex poisoned: {}. Shutting down.", msg);
+                    running.0 = false;
+                    player_disconnected = true;
+                    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;
+                }
+            }
+        }
+
+        if player_disconnected {
+            bridge.player = None;
+        }
+    }
+
+    // -- Reader connections -------------------------------------------------
+    // Role-gated: Inputs are never forwarded to InputQueue from a reader —
+    // logged, dropped, and struck instead. Atlas/star-map/city-names
+    // requests ARE forwarded, tagged with the reader's own ConnectionId.
+    // A reader's disconnect only removes that one reader — ServerRunning is
+    // untouched, and other connections (Player, other readers) are unaffected.
+    let mut disconnected_readers: Vec<ConnectionId> = Vec::new();
+    let mut to_disconnect_for_violations: Vec<ConnectionId> = Vec::new();
+
+    for reader in bridge.readers.iter_mut() {
+        let reader_id = reader.id;
+        for _ in 0..MAX_READER_INBOUND_FRAMES_PER_TICK {
+            match reader.bridge.receive() {
+                Ok(Some(Inbound::Inputs(inputs))) => {
+                    // D-254 §2 permitted-message matrix: Reader -> Inputs is
+                    // disallowed. Syntactically valid, role-forbidden — log,
+                    // drop the frame (never reaches InputQueue/SimRng, so
+                    // determinism is unaffected by construction), and strike.
+                    reader.violation_strikes += 1;
                     tracing::warn!(
-                        "Received {} input(s) before handshake completed — processing anyway (forward-compatible)",
-                        inputs.len()
+                        "Reader connection {:?} sent {} disallowed PlayerInput(s) — dropped (strike {}/{})",
+                        reader_id,
+                        inputs.len(),
+                        reader.violation_strikes,
+                        READER_VIOLATION_DISCONNECT_THRESHOLD
                     );
+                    error_buffer.push(SimError {
+                        kind: SimErrorKind::ProtocolError,
+                        message: format!(
+                            "Reader connection {:?} sent disallowed PlayerInput (role violation)",
+                            reader_id
+                        ),
+                        tick: current_tick,
+                    });
+                    if reader.violation_strikes >= READER_VIOLATION_DISCONNECT_THRESHOLD {
+                        tracing::warn!(
+                            "Reader connection {:?} exceeded violation threshold — disconnecting",
+                            reader_id
+                        );
+                        to_disconnect_for_violations.push(reader_id);
+                        break;
+                    }
                 }
-                for input in &inputs {
-                    tracing::trace!(
-                        "Received input: tick={} action={:?}",
-                        input.tick,
-                        input.action
+                Ok(Some(Inbound::AtlasRequest(req))) => {
+                    atlas_requests.0.push((reader_id, req));
+                }
+                Ok(Some(Inbound::StarMapRequest(req))) => {
+                    star_map_requests.0.push((reader_id, req));
+                }
+                Ok(Some(Inbound::CityNamesRequest(req))) => {
+                    city_names_requests.0.push((reader_id, req));
+                }
+                Ok(None) => break,
+                Err(BridgeError::Disconnected) => {
+                    tracing::info!("Reader connection {:?} disconnected", reader_id);
+                    disconnected_readers.push(reader_id);
+                    break;
+                }
+                Err(BridgeError::Io(ref e))
+                    if e.kind() == std::io::ErrorKind::BrokenPipe
+                        || e.kind() == std::io::ErrorKind::ConnectionReset =>
+                {
+                    tracing::info!("Reader connection {:?} pipe broken", reader_id);
+                    disconnected_readers.push(reader_id);
+                    break;
+                }
+                Err(BridgeError::MutexPoisoned(ref msg)) => {
+                    tracing::error!(
+                        "Reader connection {:?} bridge mutex poisoned: {}",
+                        reader_id,
+                        msg
                     );
+                    disconnected_readers.push(reader_id);
+                    break;
                 }
-                for input in inputs {
-                    input_queue.push(input);
+                Err(BridgeError::DeserializationWithDump(ref msg)) => {
+                    tracing::error!(
+                        "Reader connection {:?}: skipping malformed frame: {}",
+                        reader_id,
+                        msg
+                    );
+                    error_buffer.push(SimError {
+                        kind: SimErrorKind::ProtocolError,
+                        message: format!("Malformed reader frame: {}", msg),
+                        tick: current_tick,
+                    });
+                }
+                Err(ref e @ BridgeError::Deserialization(_)) => {
+                    tracing::error!(
+                        "Reader connection {:?}: skipping malformed frame: {}",
+                        reader_id,
+                        e
+                    );
+                    error_buffer.push(SimError {
+                        kind: SimErrorKind::ProtocolError,
+                        message: format!("Deserialization error: {}", e),
+                        tick: current_tick,
+                    });
+                }
+                Err(e) => {
+                    tracing::error!("Reader connection {:?} receive error: {}", reader_id, e);
+                    break;
                 }
-            }
-            Ok(Some(Inbound::AtlasRequest(req))) => {
-                atlas_requests.0.push(req);
-            }
-            Ok(Some(Inbound::StarMapRequest(req))) => {
-                star_map_requests.0.push(req);
-            }
-            Ok(Some(Inbound::CityNamesRequest(req))) => {
-                city_names_requests.0.push(req);
-            }
-            // 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;
             }
         }
     }
+
+    for id in disconnected_readers
+        .into_iter()
+        .chain(to_disconnect_for_violations)
+    {
+        bridge.remove_reader(id);
+    }
 }
 
-/// Send snapshot from buffer to bridge.
-/// Any send error is fatal — the client cannot proceed without snapshots.
+/// Send snapshot from buffer to the Player connection ONLY (D-254 §2 — a
+/// Reader receives no `ObserverSnapshot` at all, not even filtered; this
+/// function never touches `bridge.readers`, structurally).
+///
+/// A send error on an EXISTING Player connection is fatal for that
+/// connection — the client cannot proceed without snapshots — and shuts the
+/// server down (THE CRITICAL FIX: this is the only way this function ever
+/// touches `ServerRunning`, and it is scoped correctly, because a reader was
+/// never a snapshot recipient to begin with).
+///
+/// Having NO Player connection at all is a DIFFERENT, valid case (D-254 §1:
+/// a spawn-mode server whose sole connection is a Reader) — `bridge.player`
+/// being `None` here is not an error and must never shut the server down;
+/// the snapshot is simply not delivered anywhere (there is currently no
+/// character-controlling connection to deliver it to) and stays queued in
+/// `buffer` for whenever a Player does connect, if ever.
 pub fn send_bridge_snapshot(
-    bridge: Option<Res<BridgeResource>>,
+    bridge: Option<ResMut<BridgeResource>>,
     mut buffer: ResMut<SnapshotBuffer>,
     mut running: ResMut<ServerRunning>,
 ) {
-    let Some(bridge) = bridge else {
+    let Some(mut bridge) = bridge else {
         tracing::error!("send_bridge_snapshot: no BridgeResource");
         return;
     };
+    if !bridge.has_player() {
+        // Reader-only server (D-254 §1 spawn-mode) or a Player that hasn't
+        // finished its accept-loop handshake yet — neither is an error.
+        // Leave the snapshot queued; it is simply not deliverable this tick.
+        return;
+    }
     if let Some(snapshot) = buffer.snapshot.take() {
         if let Err(e) = bridge.send_snapshot(&snapshot) {
             match &e {
                 BridgeError::Disconnected => {
-                    tracing::info!("Client disconnected during send, shutting down");
+                    tracing::info!("Player disconnected during send, shutting down");
                 }
                 BridgeError::MutexPoisoned(msg) => {
-                    tracing::error!("Bridge mutex poisoned during send: {}", msg);
+                    tracing::error!("Player bridge mutex poisoned during send: {}", msg);
                 }
                 _ => {
-                    tracing::error!("Bridge send error: {}", e);
+                    tracing::error!("Player bridge send error: {}", e);
                 }
             }
             running.0 = false;
+            bridge.player = None;
         }
     }
 }
@@ -391,81 +761,230 @@ impl Default for ServerRunning {
 }
 
 /// Inbound atlas layer requests routed off the bridge (#969, D-225), drained by
-/// the proxy serve system in `PreInput`.
+/// the proxy serve system in `PreInput`. Each entry is tagged with the
+/// requesting connection's id (D-254 §2) so the matching response — pushed
+/// 1:1 and in order by `serve_atlas_requests` in `atlas/plugin.rs` — routes
+/// back to only that connection, never a broadcast.
 #[derive(Resource, Default)]
-pub struct AtlasRequestBuffer(pub Vec<AtlasLayerRequest>);
+pub struct AtlasRequestBuffer(pub Vec<(ConnectionId, AtlasLayerRequest)>);
 
 /// Outbound atlas layer responses, filled by the proxy serve system and flushed
-/// to the client in `PostSnapshot` (#969, D-225).
+/// to the client in `PostSnapshot` (#969, D-225). Connection-tagged (D-254 §2).
 #[derive(Resource, Default)]
-pub struct AtlasResponseBuffer(pub Vec<AtlasLayerResponse>);
+pub struct AtlasResponseBuffer(pub Vec<(ConnectionId, AtlasLayerResponse)>);
 
-/// Flush buffered atlas responses to the client (#969, D-225). A failed send is
-/// logged but not fatal — an atlas response is not load-bearing like a snapshot.
+/// Flush buffered atlas responses to their requesting connections (#969,
+/// D-225 original; D-254 §2 adds per-connection routing). A failed send is
+/// logged but not fatal — an atlas response is not load-bearing like a
+/// snapshot, and a stale/disconnected recipient is not an error (see
+/// `BridgeResource::send_atlas_response_to`).
 pub fn send_atlas_responses(
     bridge: Option<Res<BridgeResource>>,
     mut buffer: ResMut<AtlasResponseBuffer>,
 ) {
     let Some(bridge) = bridge else { return };
-    for resp in buffer.0.drain(..) {
-        if let Err(e) = bridge.send_atlas_response(&resp) {
-            tracing::warn!("failed to send atlas response for {}: {}", resp.body_id, e);
+    for (id, resp) in buffer.0.drain(..) {
+        if let Err(e) = bridge.send_atlas_response_to(id, &resp) {
+            tracing::warn!(
+                "failed to send atlas response for {} to {:?}: {}",
+                resp.body_id,
+                id,
+                e
+            );
         }
     }
 }
 
 /// Inbound star-map requests routed off the bridge (T-949a), drained by the
-/// proxy serve system in `PreInput`.
+/// proxy serve system in `PreInput`. Connection-tagged (D-254 §2).
 #[derive(Resource, Default)]
-pub struct StarMapRequestBuffer(pub Vec<StarMapRequest>);
+pub struct StarMapRequestBuffer(pub Vec<(ConnectionId, StarMapRequest)>);
 
 /// Outbound star-map responses, filled by the proxy serve system and flushed
-/// to the client in `PostSnapshot` (T-949a).
+/// to the client in `PostSnapshot` (T-949a). Connection-tagged (D-254 §2).
 #[derive(Resource, Default)]
-pub struct StarMapResponseBuffer(pub Vec<StarMapResponse>);
+pub struct StarMapResponseBuffer(pub Vec<(ConnectionId, StarMapResponse)>);
 
-/// Flush buffered star-map responses to the client (T-949a). A failed send is
-/// logged but not fatal.
+/// Flush buffered star-map responses to their requesting connections
+/// (T-949a original; D-254 §2 adds per-connection routing). A failed send
+/// is logged but not fatal.
 pub fn send_star_map_responses(
     bridge: Option<Res<BridgeResource>>,
     mut buffer: ResMut<StarMapResponseBuffer>,
 ) {
     let Some(bridge) = bridge else { return };
-    for resp in buffer.0.drain(..) {
-        if let Err(e) = bridge.send_star_map_response(&resp) {
-            tracing::warn!("failed to send star map response: {}", e);
+    for (id, resp) in buffer.0.drain(..) {
+        if let Err(e) = bridge.send_star_map_response_to(id, &resp) {
+            tracing::warn!("failed to send star map response to {:?}: {}", id, e);
         }
     }
 }
 
 /// Inbound city-names requests routed off the bridge (T-949b), drained by the
-/// proxy serve system in `PreInput`.
+/// proxy serve system in `PreInput`. Connection-tagged (D-254 §2).
 #[derive(Resource, Default)]
-pub struct CityNamesRequestBuffer(pub Vec<CityNamesRequest>);
+pub struct CityNamesRequestBuffer(pub Vec<(ConnectionId, CityNamesRequest)>);
 
 /// Outbound city-names responses, filled by the proxy serve system and
-/// flushed to the client in `PostSnapshot` (T-949b).
+/// flushed to the client in `PostSnapshot` (T-949b). Connection-tagged
+/// (D-254 §2).
 #[derive(Resource, Default)]
-pub struct CityNamesResponseBuffer(pub Vec<CityNamesResponse>);
+pub struct CityNamesResponseBuffer(pub Vec<(ConnectionId, CityNamesResponse)>);
 
-/// Flush buffered city-names responses to the client (T-949b). A failed send
+/// Flush buffered city-names responses to their requesting connections
+/// (T-949b original; D-254 §2 adds per-connection routing). A failed send
 /// is logged but not fatal.
 pub fn send_city_names_responses(
     bridge: Option<Res<BridgeResource>>,
     mut buffer: ResMut<CityNamesResponseBuffer>,
 ) {
     let Some(bridge) = bridge else { return };
-    for resp in buffer.0.drain(..) {
-        if let Err(e) = bridge.send_city_names_response(&resp) {
+    for (id, resp) in buffer.0.drain(..) {
+        if let Err(e) = bridge.send_city_names_response_to(id, &resp) {
             tracing::warn!(
-                "failed to send city names response for {}: {}",
+                "failed to send city names response for {} to {:?}: {}",
                 resp.body_id,
+                id,
                 e
             );
         }
     }
 }
 
+/// Holds the server's TCP listener for accepting connections AFTER the
+/// first Player connection (D-254 §2, T-1130).
+///
+/// The first connection is still accepted by `main.rs`'s original
+/// blocking `listener.accept()` before the tick loop begins (unchanged —
+/// see `main.rs`), matching a normal game launch exactly byte-for-byte
+/// when nobody else ever connects. This resource wraps the SAME listener
+/// (moved into it after that first accept) so `accept_new_connections` can
+/// keep accepting *additional* connections once the tick loop is running —
+/// this is what fixes the original starvation bug (a second client used to
+/// hang forever waiting for an `accept()` call that would never come).
+///
+/// `None` when no listener is wired (e.g. most existing unit/integration
+/// tests that construct a `BridgeResource` directly and never spawn a real
+/// listener) — `accept_new_connections` is a no-op in that case, so it is
+/// always safe to add to any `App`/`World` without also wiring a listener.
+#[derive(Resource, Default)]
+pub struct ConnectionListener(pub Option<std::net::TcpListener>);
+
+/// Connections that have been TCP-accepted but have not yet completed their
+/// handshake/startup exchange (D-254 §2, T-1130). Polled non-blockingly
+/// every tick by `accept_new_connections` — see `tcp::PendingConnection`'s
+/// doc for why this can never stall the tick loop.
+#[derive(Resource, Default)]
+pub struct PendingConnections(pub Vec<crate::bridge::tcp::PendingConnection>);
+
+/// Accept new TCP connections and advance in-progress handshakes, without
+/// ever blocking the tick loop (D-254 §2, T-1130).
+///
+/// Two independent, non-blocking steps each tick:
+/// 1. Try to accept any newly-arrived TCP connection on `ConnectionListener`
+///    (the listener itself is non-blocking — `main.rs` sets this before
+///    wrapping it in the resource). A `WouldBlock`/no-pending-connection
+///    result is the overwhelmingly common case (no new client this tick)
+///    and is silently ignored, not logged.
+/// 2. Poll every connection in `PendingConnections`. `PendingPoll::Waiting`
+///    connections stay queued for next tick. `PendingPoll::Ready`
+///    connections are promoted based on `startup.role`:
+///    - `Player`, and `BridgeResource` has no Player yet → installed as the
+///      Player connection.
+///    - `Player`, and a Player already exists → THE SECOND-PLAYER CASE
+///      (D-254 §2/T-1130 scope: "a second Player attempt gets a clean
+///      rejection, not a hang"). The connection is dropped immediately
+///      after the handshake completes — no silent starvation (the original
+///      bug), and no impact on the existing Player's session (its
+///      connection is never touched). A client attempting to connect as a
+///      second Player sees a clean disconnect right after startup, which
+///      is a well-defined, discoverable failure — the correct behavior for
+///      an out-of-scope case (general N-player is explicitly D-009's
+///      separate ambition, not this ticket's).
+///    - `Reader` → installed as a new Reader connection, always (0-N
+///      readers is the whole point of this ticket).
+///    `PendingPoll::Failed` connections (EOF before completing handshake,
+///    malformed startup, etc.) are simply dropped — never logged as errors
+///    at more than `warn` level, since an incomplete handshake from a
+///    probing/misbehaving client is an expected occurrence, not a bug.
+pub fn accept_new_connections(
+    listener: Option<Res<ConnectionListener>>,
+    mut pending: Option<ResMut<PendingConnections>>,
+    bridge: Option<ResMut<BridgeResource>>,
+) {
+    let (Some(listener), Some(pending), Some(mut bridge)) = (listener, pending.as_mut(), bridge)
+    else {
+        return;
+    };
+
+    // Step 1: accept any newly-arrived connection (non-blocking listener).
+    if let Some(l) = listener.0.as_ref() {
+        match l.accept() {
+            Ok((stream, peer_addr)) => {
+                tracing::info!("accepted new connection from {}", peer_addr);
+                match crate::bridge::tcp::PendingConnection::new(stream) {
+                    Ok(conn) => pending.0.push(conn),
+                    Err(e) => tracing::warn!("failed to wrap accepted connection: {}", e),
+                }
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                // No connection pending — the overwhelmingly common case.
+            }
+            Err(e) => {
+                tracing::warn!("accept() failed: {}", e);
+            }
+        }
+    }
+
+    // Step 2: advance every in-progress handshake by one non-blocking poll.
+    let mut still_pending = Vec::with_capacity(pending.0.len());
+    for mut conn in pending.0.drain(..) {
+        match conn.poll() {
+            tcp::PendingPoll::Waiting => still_pending.push(conn),
+            tcp::PendingPoll::Ready { stream, startup } => match startup.role {
+                ConnectionRole::Player => {
+                    if bridge.has_player() {
+                        tracing::warn!(
+                            "second Player connection attempt rejected (0-1 Player + 0-N Reader scope, D-254 §2) — disconnecting"
+                        );
+                        // Dropping `stream` closes the TCP connection — a
+                        // clean, immediate disconnect, not a hang.
+                        drop(stream);
+                    } else {
+                        match TcpBridge::from_connected_stream(stream) {
+                            Ok(tcp_bridge) => {
+                                let id = bridge.insert_player(tcp_bridge);
+                                tracing::info!("Player connection established: {:?}", id);
+                            }
+                            Err(e) => {
+                                tracing::warn!(
+                                    "failed to promote pending Player connection: {}",
+                                    e
+                                );
+                            }
+                        }
+                    }
+                }
+                ConnectionRole::Reader => match TcpBridge::from_connected_stream(stream) {
+                    Ok(tcp_bridge) => {
+                        let id = bridge.insert_reader(tcp_bridge);
+                        tracing::info!("Reader connection established: {:?}", id);
+                    }
+                    Err(e) => {
+                        tracing::warn!("failed to promote pending Reader connection: {}", e);
+                    }
+                },
+            },
+            tcp::PendingPoll::Failed => {
+                // Handshake never completed (EOF, malformed startup, etc.) —
+                // drop silently at info level. Not a server error.
+                tracing::info!("pending connection failed to complete handshake");
+            }
+        }
+    }
+    pending.0 = still_pending;
+}
+
 /// Bridge plugin for client-server communication
 /// Abstracts transport layer (LocalBridge/NetworkBridge)
 pub struct BridgePlugin;
@@ -488,6 +1007,18 @@ impl Plugin for BridgePlugin {
             .init_resource::<StarMapResponseBuffer>()
             .init_resource::<CityNamesRequestBuffer>()
             .init_resource::<CityNamesResponseBuffer>()
+            .init_resource::<ConnectionListener>()
+            .init_resource::<PendingConnections>()
+            // Multi-connection accept-loop (D-254 §2, T-1130) — must run
+            // before receive_bridge_inputs so a connection whose handshake
+            // completes this tick has its first frame drained the same
+            // tick, not next tick.
+            .add_systems(
+                Update,
+                accept_new_connections
+                    .before(receive_bridge_inputs)
+                    .in_set(TickPhase::PreInput),
+            )
             // 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))

server/src/bridge/tcp.rs

@@ -143,6 +143,35 @@ impl TcpBridge {
         })
     }
 
+    /// Server-side: promote an already-connected, already-handshaken stream
+    /// into a full `TcpBridge` (D-254 §2, T-1130).
+    ///
+    /// Used by [`PendingConnection`] once its non-blocking handshake/startup
+    /// exchange completes — unlike `accept`/`accept_on`, this does not call
+    /// `listener.accept()` itself; the stream is already connected and (per
+    /// `PendingConnection`'s contract) already sent `HandshakeMessage` and
+    /// received a valid `StartupMessage`. Sets the stream non-blocking for
+    /// the steady-state tick loop, same as every other constructor here.
+    pub fn from_connected_stream(stream: TcpStream) -> Result<Self, BridgeError> {
+        let local_addr = stream
+            .local_addr()
+            .map_err(|e| BridgeError::Transport(format!("failed to get local address: {}", e)))?;
+
+        stream
+            .set_nonblocking(true)
+            .map_err(|e| BridgeError::Transport(format!("failed to set non-blocking: {}", e)))?;
+
+        let reader_stream = stream.try_clone().map_err(|e| {
+            BridgeError::Transport(format!("failed to clone stream for reader: {}", e))
+        })?;
+
+        Ok(Self {
+            reader: Mutex::new(ReadHalf::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
@@ -285,3 +314,139 @@ impl SimBridge for TcpBridge {
         Ok(())
     }
 }
+
+/// A connection that has been TCP-accepted but has not yet completed the
+/// handshake/startup exchange (D-254 §2, T-1130).
+///
+/// Exists because the multi-connection accept-loop in `main.rs` runs inside
+/// the live tick loop — unlike the original single-connection startup
+/// sequence (`TcpBridge::accept` + `receive_startup`, which blocks freely
+/// because nothing else is running yet), a connection accepted *after* the
+/// server is already ticking must never stall other connections while it
+/// completes its own handshake. `PendingConnection` is polled once per tick,
+/// non-blockingly, exactly like `TcpBridge::receive()` already is — a slow
+/// or hostile client sits here indefinitely, consuming no thread and
+/// blocking nothing, until it finishes or disconnects.
+///
+/// State machine: `AwaitingHandshakeSend` (send the empty `HandshakeMessage`
+/// — reused from `send_handshake`'s wire shape) → `AwaitingStartup` (poll for
+/// a complete `StartupMessage` frame via the same `FrameAccumulator` the
+/// steady-state `receive()` path uses, so a startup message split across TCP
+/// segments reassembles correctly here too).
+pub struct PendingConnection {
+    stream: TcpStream,
+    accum: FrameAccumulator,
+    state: PendingState,
+}
+
+enum PendingState {
+    AwaitingHandshakeSend,
+    AwaitingStartup,
+}
+
+/// Result of one `PendingConnection::poll()` call.
+pub enum PendingPoll {
+    /// Handshake not yet complete — keep polling next tick.
+    Waiting,
+    /// Startup message received and decoded. Caller promotes this into a
+    /// full `TcpBridge` via `TcpBridge::from_connected_stream`.
+    Ready {
+        stream: TcpStream,
+        startup: super::StartupMessage,
+    },
+    /// The connection died before completing its handshake (EOF or a fatal
+    /// I/O error). Caller drops this pending connection.
+    Failed,
+}
+
+impl PendingConnection {
+    /// Wrap a freshly-`accept()`-ed stream. Sets non-blocking immediately —
+    /// this type never blocks the tick loop, by construction.
+    pub fn new(stream: TcpStream) -> Result<Self, BridgeError> {
+        stream
+            .set_nonblocking(true)
+            .map_err(|e| BridgeError::Transport(format!("failed to set non-blocking: {}", e)))?;
+        Ok(Self {
+            stream,
+            accum: FrameAccumulator::new(),
+            state: PendingState::AwaitingHandshakeSend,
+        })
+    }
+
+    /// Advance the handshake by one tick's worth of non-blocking I/O.
+    ///
+    /// Never blocks: a `WouldBlock` on either the handshake write or the
+    /// startup read simply returns `PendingPoll::Waiting` for another tick
+    /// to retry. `HandshakeMessage` is a fixed few bytes (an empty
+    /// MessagePack map plus the 4-byte length prefix) — in practice it
+    /// completes in a single non-blocking write, but the retry path exists
+    /// for the same reason `send_handshake`'s blocking toggle exists on the
+    /// first connection: TCP send buffers are not guaranteed instantaneous.
+    pub fn poll(&mut self) -> PendingPoll {
+        use super::types::HandshakeMessage;
+        use std::io::Write;
+
+        if matches!(self.state, PendingState::AwaitingHandshakeSend) {
+            let msg = HandshakeMessage {};
+            let payload = match rmp_serde::to_vec_named(&msg) {
+                Ok(p) => p,
+                Err(e) => {
+                    tracing::error!("pending connection: failed to encode handshake: {}", e);
+                    return PendingPoll::Failed;
+                }
+            };
+            match write_framed(&mut self.stream, &payload) {
+                Ok(()) => {
+                    self.state = PendingState::AwaitingStartup;
+                }
+                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                    return PendingPoll::Waiting;
+                }
+                Err(e) => {
+                    tracing::warn!("pending connection: handshake send failed: {}", e);
+                    return PendingPoll::Failed;
+                }
+            }
+            // write_framed calls flush() internally — no separate flush needed.
+            let _ = self.stream.flush();
+        }
+
+        match self.accum.poll_frame(&mut self.stream) {
+            Ok(Some(payload)) => match rmp_serde::from_slice::<super::StartupMessage>(&payload) {
+                Ok(startup) => {
+                    tracing::info!(
+                        "pending connection: startup received, role={:?}",
+                        startup.role
+                    );
+                    // try_clone so the caller gets an owned stream; self.stream
+                    // is dropped with this PendingConnection once the caller
+                    // promotes the clone into a TcpBridge.
+                    match self.stream.try_clone() {
+                        Ok(stream) => PendingPoll::Ready { stream, startup },
+                        Err(e) => {
+                            tracing::error!(
+                                "pending connection: failed to clone stream for promotion: {}",
+                                e
+                            );
+                            PendingPoll::Failed
+                        }
+                    }
+                }
+                Err(e) => {
+                    tracing::warn!("pending connection: malformed startup message: {}", e);
+                    PendingPoll::Failed
+                }
+            },
+            Ok(None) => {
+                // Clean EOF — the peer disconnected before sending startup.
+                PendingPoll::Failed
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => PendingPoll::Waiting,
+            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => PendingPoll::Failed,
+            Err(e) => {
+                tracing::warn!("pending connection: startup read failed: {}", e);
+                PendingPoll::Failed
+            }
+        }
+    }
+}

server/src/bridge/types.rs

@@ -20,6 +20,29 @@ pub use crate::simulation::time::{DayPhase, TickRate};
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct HandshakeMessage {}
 
+/// Connection role, gating what a connection may send/receive (D-254 §2).
+///
+/// `Player` is the sole role that may enter the character-spawn path, send
+/// `PlayerInput`, or receive `ObserverSnapshot`. `Reader` is a genuinely new
+/// class: no character, no inputs, no per-tick snapshot — structurally
+/// enforced server-side (`main.rs`, `bridge::mod`), never client courtesy.
+///
+/// Shaped for growth per D-254 §6: a future `TradingReader` variant is a
+/// strict superset of `Reader` (everything `Reader` gets, plus a narrow
+/// per-verb `PlayerInput` allowlist) — not a replacement, and not added by
+/// this ticket (T-1130 scope is `Player`/`Reader` only).
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
+pub enum ConnectionRole {
+    /// The single character-controlling connection. Only role that existed
+    /// before D-254. Default for back-compat (see `StartupMessage::role`).
+    #[default]
+    Player,
+    /// Read-only observer: no character, no inputs, no `ObserverSnapshot`.
+    /// May request install-static/world-public data (atlas/star-map/
+    /// city-names) — see the permitted-message matrix in `bridge::mod`.
+    Reader,
+}
+
 /// Startup message sent by the client after receiving HandshakeMessage (#175).
 /// Contains the world seed for deterministic simulation (D-010, D-029).
 ///
@@ -35,7 +58,22 @@ pub struct StartupMessage {
     /// World seed for SimRng initialization.
     /// Generated by SessionManager.new_game() on the client.
     /// Same seed → same EntanglementConfig → same NPC population (D-029).
+    ///
+    /// Ignored server-side for `Reader` connections (D-254 §2): a reader
+    /// inherits whatever `SimRng` state the server already has (from the
+    /// Player's own `StartupMessage`, or `--seed` in spawn-mode) — a second
+    /// StartupMessage must never re-seed `SimRng` after tick 0, or a
+    /// reader attaching mid-session would silently break determinism for
+    /// the Player already connected.
     pub world_seed: u64,
+    /// Connection role (D-254 §2). `#[serde(default)]` makes this field
+    /// optional on the wire: an old-format client that only ever sent
+    /// `{world_seed}` (every client before D-254) still decodes cleanly,
+    /// with `role` defaulting to `ConnectionRole::Player` — the same
+    /// connection behavior that client always got. Byte-compatible,
+    /// forward-compatible; no protocol version bump (D-192 precedent).
+    #[serde(default)]
+    pub role: ConnectionRole,
 }
 
 /// The ONLY data structure crossing the client-server boundary (D-020)
@@ -1101,6 +1139,7 @@ mod tests {
     fn startup_message_roundtrip() {
         let msg = StartupMessage {
             world_seed: 0xDEADBEEF,
+            role: ConnectionRole::Player,
         };
         let bytes = rmp_serde::to_vec_named(&msg).expect("serialize");
         let decoded: StartupMessage = rmp_serde::from_slice(&bytes).expect("deserialize");
@@ -1110,7 +1149,10 @@ mod tests {
 
     #[test]
     fn startup_message_zero_seed() {
-        let msg = StartupMessage { world_seed: 0 };
+        let msg = StartupMessage {
+            world_seed: 0,
+            role: ConnectionRole::Player,
+        };
         let bytes = rmp_serde::to_vec_named(&msg).expect("serialize");
         let decoded: StartupMessage = rmp_serde::from_slice(&bytes).expect("deserialize");
         assert_eq!(decoded.world_seed, 0);
@@ -1120,12 +1162,57 @@ mod tests {
     fn startup_message_max_seed() {
         let msg = StartupMessage {
             world_seed: u64::MAX,
+            role: ConnectionRole::Player,
         };
         let bytes = rmp_serde::to_vec_named(&msg).expect("serialize");
         let decoded: StartupMessage = rmp_serde::from_slice(&bytes).expect("deserialize");
         assert_eq!(decoded.world_seed, u64::MAX);
     }
 
+    /// D-254 §2 / T-1130 behavior 1: an old-format `StartupMessage` — the
+    /// exact wire shape every client before this ticket sent, `{world_seed}`
+    /// only, no `role` key at all — must still decode cleanly, with `role`
+    /// defaulting to `Player`. This is the byte-compatibility guarantee: no
+    /// already-shipped client encoder needs to change for this ticket to be
+    /// safe to deploy alongside it.
+    #[test]
+    fn old_format_startup_message_decodes_as_player() {
+        // Hand-encode the pre-D-254 shape directly (a single-key map), rather
+        // than deriving it from a struct literal, so this test can't
+        // accidentally pass just because both sides changed together.
+        #[derive(Serialize)]
+        struct OldStartupMessage {
+            world_seed: u64,
+        }
+        let old_msg = OldStartupMessage {
+            world_seed: 0x1234_5678,
+        };
+        let bytes = rmp_serde::to_vec_named(&old_msg).expect("serialize old-format message");
+
+        let decoded: StartupMessage =
+            rmp_serde::from_slice(&bytes).expect("old-format message must still decode");
+        assert_eq!(decoded.world_seed, 0x1234_5678);
+        assert_eq!(
+            decoded.role,
+            ConnectionRole::Player,
+            "role must default to Player when absent from the wire"
+        );
+    }
+
+    /// D-254 §2: a new-format message explicitly carrying `role: Reader`
+    /// must decode with that role preserved — the default only applies when
+    /// the field is genuinely absent, it must not clobber an explicit value.
+    #[test]
+    fn startup_message_reader_role_roundtrips() {
+        let msg = StartupMessage {
+            world_seed: 99,
+            role: ConnectionRole::Reader,
+        };
+        let bytes = rmp_serde::to_vec_named(&msg).expect("serialize");
+        let decoded: StartupMessage = rmp_serde::from_slice(&bytes).expect("deserialize");
+        assert_eq!(decoded.role, ConnectionRole::Reader);
+    }
+
     #[test]
     fn handshake_is_distinct_from_snapshot() {
         // HandshakeMessage and ObserverSnapshot are different types on the wire.

server/src/main.rs

@@ -11,7 +11,9 @@ use bevy_app::prelude::*;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
 use settled_reach_server::bridge::tcp::TcpBridge;
-use settled_reach_server::bridge::{BridgePlugin, BridgeResource, HandshakeState, ServerRunning};
+use settled_reach_server::bridge::{
+    BridgePlugin, BridgeResource, ConnectionListener, ConnectionRole, HandshakeState, ServerRunning,
+};
 use settled_reach_server::simulation::SimulationPlugin;
 
 fn main() {
@@ -117,6 +119,25 @@ fn main() {
     }
 
     tracing::info!("Waiting for client connection on port {}", actual_port);
+    // D-254 §2/T-1130: the FIRST connection is still accepted here, exactly
+    // as before — one blocking listener.accept() call, byte-identical to
+    // pre-D-254 behavior when nobody else ever connects. What changes is
+    // AFTER: the listener is set non-blocking and handed to
+    // ConnectionListener (inserted below) so accept_new_connections can
+    // keep accepting additional connections once the tick loop starts,
+    // instead of the original bug where a second accept() call never
+    // happened at all and a second client hung forever.
+    //
+    // accept_on() consumes the listener; clone it first so both the first
+    // accept AND the later non-blocking accept-loop have a working handle
+    // on the same underlying socket (TcpListener::try_clone shares the fd,
+    // not a new listener — connections queued on either handle are visible
+    // to both, same as TcpStream::try_clone is already used for read/write
+    // halves throughout this bridge).
+    let listener_for_loop = listener.try_clone().unwrap_or_else(|e| {
+        tracing::error!("Failed to clone listener for accept-loop: {}", e);
+        std::process::exit(1);
+    });
     let bridge = TcpBridge::accept_on(listener).unwrap_or_else(|e| {
         tracing::error!("Failed to accept: {}", e);
         std::process::exit(1);
@@ -291,9 +312,42 @@ fn main() {
         }
     }
 
-    app.insert_resource(BridgeResource::new(bridge));
+    // D-254 §2/T-1130: the FIRST connection's role, exactly as it does for
+    // every later accept-loop connection (main.rs's accept-loop handles
+    // connections 2+; this handles the honest first-connection case a
+    // spawn-mode Reader server actually needs — D-254 §1's spawn-mode
+    // Atlas companion connects as the ONLY connection to a freshly-spawned
+    // server, so "first connection" and "Reader" are not mutually
+    // exclusive). `BridgeResource::default()` + explicit insert_player/
+    // insert_reader replaces the old unconditional `BridgeResource::new`
+    // (which always meant "install as Player" — there was no other role
+    // before this ticket).
+    let mut bridge_resource = BridgeResource::default();
+    match startup.role {
+        ConnectionRole::Player => {
+            bridge_resource.insert_player(bridge);
+        }
+        ConnectionRole::Reader => {
+            tracing::info!(
+                "first connection is a Reader (D-254 §1 spawn-mode) — no character will be spawned for it"
+            );
+            bridge_resource.insert_reader(bridge);
+        }
+    }
+    app.insert_resource(bridge_resource);
     app.insert_resource(HandshakeState::Complete);
 
+    // D-254 §2/T-1130: wire the cloned listener non-blocking so
+    // accept_new_connections (BridgePlugin, PreInput) can accept additional
+    // connections every tick without ever blocking the tick loop. This is
+    // the actual fix for the original starvation bug — before this, there
+    // was exactly one listener.accept() call in the whole process lifetime.
+    listener_for_loop.set_nonblocking(true).unwrap_or_else(|e| {
+        tracing::error!("Failed to set accept-loop listener non-blocking: {}", e);
+        std::process::exit(1);
+    });
+    app.insert_resource(ConnectionListener(Some(listener_for_loop)));
+
     // SimulationPlugin { seed } already inserts SimRng with the correct seed
     // during plugin build. We re-insert here as a defensive override for one
     // specific ordering risk: any future plugin that registers *before*
@@ -316,6 +370,23 @@ fn main() {
     );
 
     // Gauntlet test world for --test-mode, proof room for normal mode.
+    //
+    // D-254 §2/T-1130 scope note: this call is NOT gated on the first
+    // connection's role, even though it spawns a `PlayerCharacter` entity
+    // unconditionally. A Reader-only spawned server (D-254 §1 spawn-mode)
+    // therefore has an inert, unpiloted `PlayerCharacter` entity sitting in
+    // its ECS world — nothing drives it (no Player connection exists to
+    // send it inputs), and the Reader never learns it exists: `send_
+    // bridge_snapshot` routes `ObserverSnapshot` to the Player connection
+    // ONLY and is a documented no-op with no Player installed (see
+    // `bridge::send_bridge_snapshot`), so this entity's existence has no
+    // observable effect on a Reader-only session. Splitting character-spawn
+    // out of `setup_proof_room`/`setup_gauntlet` (both of which also wire
+    // NPCs, the walkability map, and the relationship graph — genuinely
+    // "whole world setup", not just "spawn the player") into an optional
+    // step is real refactoring work, correctly out of scope for this
+    // gating ticket; tracked as follow-up, not required for the D-010
+    // information-boundary guarantee this ticket exists to establish.
     if test_mode {
         #[cfg(feature = "gauntlet")]
         settled_reach_server::test_world::setup_gauntlet(&mut app);

server/tests/bridge_tcp.rs

@@ -421,9 +421,521 @@ fn single_tick_drains_all_ready_inbound_frames() {
         1,
         "the city-names request must drain in the same tick (H6: real wire path)"
     );
-    assert_eq!(city_names[0].body_id, "GJ1c");
+    assert_eq!(city_names[0].1.body_id, "GJ1c");
     assert!(
         world.resource::<ServerRunning>().0,
         "draining must not shut the server down"
     );
 }
+
+// -- D-254 §2 / T-1130: multi-connection bridge + ConnectionRole gate ---------
+//
+// Six behaviors below (a 7th — old-format StartupMessage decodes as Player —
+// is a unit test in server/src/bridge/types.rs, next to the type itself).
+// Each test drives the real wire path: a genuine TcpStream client performs
+// the handshake/startup exchange a real Godot client (or PendingConnection's
+// server-side counterpart) would, against the real non-blocking accept-loop
+// and drain-loop systems via `run_system_once` — no mocks of the framing or
+// role-gate logic itself.
+
+/// Client-side test helper: perform one full handshake/startup exchange over
+/// an already-connected stream, exactly as `sim_bridge.gd`'s live-mode path
+/// does (read HandshakeMessage, write StartupMessage). Returns the stream so
+/// the caller can continue driving it (send inputs, read responses, etc.).
+fn client_handshake(mut stream: TcpStream, role: ConnectionRole) -> TcpStream {
+    let handshake_payload = read_framed(&mut stream)
+        .expect("failed to read handshake")
+        .expect("unexpected EOF reading handshake");
+    let _: HandshakeMessage =
+        rmp_serde::from_slice(&handshake_payload).expect("failed to decode HandshakeMessage");
+
+    let startup = StartupMessage {
+        world_seed: 12345,
+        role,
+    };
+    let payload = rmp_serde::to_vec_named(&startup).expect("failed to serialize StartupMessage");
+    write_framed(&mut stream, &payload).expect("failed to write StartupMessage");
+    stream
+}
+
+/// Drive `accept_new_connections` for up to `max_ticks` schedule passes, or
+/// until `bridge.readers().len() + bridge.has_player() as usize` (checked via
+/// the passed predicate) is satisfied — matches this test file's existing
+/// wall-clock-deadline-over-fixed-sleep hardening (see `collect_input_batches`
+/// above): connection promotion depends on TCP delivery timing, not a fixed
+/// tick count, so poll until true or time out rather than guessing a sleep.
+fn drive_accept_loop_until(
+    world: &mut bevy_ecs::world::World,
+    condition: impl Fn(&bevy_ecs::world::World) -> bool,
+) {
+    use bevy_ecs::system::RunSystemOnce;
+    use settled_reach_server::bridge::accept_new_connections;
+
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+    while !condition(world) {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "timed out waiting for accept-loop condition"
+        );
+        world
+            .run_system_once(accept_new_connections)
+            .expect("accept_new_connections failed to run");
+        thread::sleep(std::time::Duration::from_millis(1));
+    }
+}
+
+/// Build a `bevy_ecs::World` wired exactly like `BridgePlugin` wires it for
+/// the systems under test here (accept-loop + drain-loop + response
+/// buffers), bound to a fresh OS-assigned port.
+fn new_multi_connection_world() -> (bevy_ecs::world::World, std::net::SocketAddr) {
+    use settled_reach_server::bridge::{
+        AtlasRequestBuffer, AtlasResponseBuffer, BridgeResource, CityNamesRequestBuffer,
+        CityNamesResponseBuffer, ConnectionListener, HandshakeState, PendingConnections,
+        ServerRunning, SnapshotBuffer, StarMapRequestBuffer, StarMapResponseBuffer,
+    };
+    use settled_reach_server::simulation::input::InputQueue;
+
+    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
+    let addr = listener.local_addr().expect("failed to get local address");
+    listener
+        .set_nonblocking(true)
+        .expect("failed to set listener non-blocking");
+
+    let mut world = bevy_ecs::world::World::new();
+    world.insert_resource(ConnectionListener(Some(listener)));
+    world.init_resource::<PendingConnections>();
+    world.insert_resource(BridgeResource::default());
+    world.init_resource::<InputQueue>();
+    world.init_resource::<ServerRunning>();
+    world.insert_resource(HandshakeState::Complete);
+    world.init_resource::<SimErrorBuffer>();
+    world.init_resource::<AtlasRequestBuffer>();
+    world.init_resource::<AtlasResponseBuffer>();
+    world.init_resource::<StarMapRequestBuffer>();
+    world.init_resource::<StarMapResponseBuffer>();
+    world.init_resource::<CityNamesRequestBuffer>();
+    world.init_resource::<CityNamesResponseBuffer>();
+    world.init_resource::<SnapshotBuffer>();
+    (world, addr)
+}
+
+/// T-1130 behavior 2: a Reader connection's handshake succeeds — it reaches
+/// `BridgeResource`'s reader collection — without any character-spawn
+/// concept ever entering the picture. There is no `PlayerCharacter`-query
+/// assertion here because that's the correct proof of the D-254 §2
+/// guarantee: character-spawn is a `main.rs`/world-setup concern entirely
+/// disjoint from connection acceptance (see `main.rs`'s D-254 §2 scope-note
+/// comment at the `setup_proof_room`/`setup_gauntlet` call site) — a Reader
+/// reaching `BridgeResource.readers` never touches that code path at all,
+/// which this test demonstrates by never invoking it and the connection
+/// still working end-to-end.
+#[test]
+fn reader_handshake_succeeds_and_installs_as_reader() {
+    use settled_reach_server::bridge::BridgeResource;
+
+    let (mut world, addr) = new_multi_connection_world();
+
+    let client_handle = thread::spawn(move || {
+        let stream = TcpStream::connect(addr).expect("failed to connect");
+        client_handshake(stream, ConnectionRole::Reader)
+    });
+
+    drive_accept_loop_until(&mut world, |w| {
+        w.resource::<BridgeResource>().reader_count() == 1
+    });
+
+    let _stream = client_handle.join().expect("client thread panicked");
+
+    let bridge = world.resource::<BridgeResource>();
+    assert!(
+        !bridge.has_player(),
+        "a Reader-only connection must never be installed as Player"
+    );
+    assert_eq!(bridge.reader_count(), 1, "exactly one reader installed");
+}
+
+/// T-1130 behavior 3: an atlas/star-map/city-names response addressed to a
+/// Reader's `ConnectionId` reaches that Reader over the wire — the
+/// per-connection response-tagging plumbing (D-254 §2) actually delivers,
+/// not just tags in-memory. Drives `send_star_map_responses` directly
+/// (bypassing `serve_star_map_requests`/the atlas proxy, which are already
+/// covered elsewhere) to isolate exactly the tagging/routing behavior this
+/// ticket adds.
+#[test]
+fn reader_receives_tagged_star_map_response() {
+    use bevy_ecs::system::RunSystemOnce;
+    use settled_reach_server::atlas::atlas_data_proxy::{StarMapResponse, StarMapStatus};
+    use settled_reach_server::bridge::{
+        send_star_map_responses, BridgeResource, StarMapResponseBuffer,
+    };
+
+    let (mut world, addr) = new_multi_connection_world();
+
+    let client_handle = thread::spawn(move || {
+        let stream = TcpStream::connect(addr).expect("failed to connect");
+        client_handshake(stream, ConnectionRole::Reader)
+    });
+
+    drive_accept_loop_until(&mut world, |w| {
+        w.resource::<BridgeResource>().reader_count() == 1
+    });
+    let mut stream = client_handle.join().expect("client thread panicked");
+
+    let reader_id = world
+        .resource::<BridgeResource>()
+        .reader_ids()
+        .first()
+        .copied()
+        .expect("reader must be installed");
+
+    world.resource_mut::<StarMapResponseBuffer>().0.push((
+        reader_id,
+        StarMapResponse {
+            status: StarMapStatus::Ready,
+            data: Some(serde_json::json!({"nodes": [], "edges": []})),
+        },
+    ));
+
+    world
+        .run_system_once(send_star_map_responses)
+        .expect("send_star_map_responses failed to run");
+
+    let payload = read_framed(&mut stream)
+        .expect("failed to read response frame")
+        .expect("unexpected EOF reading response");
+    let resp: StarMapResponse = rmp_serde::from_slice(&payload).expect("failed to decode");
+    assert_eq!(resp.status, StarMapStatus::Ready);
+}
+
+/// T-1130 behavior 4: while a Player streams `ObserverSnapshot` every tick,
+/// a concurrently-connected Reader receives NOTHING on its socket — not a
+/// filtered snapshot, not an empty one, nothing at all (D-254 §2: "not even
+/// filtered" is structural, this test proves it holds over the wire with
+/// both connections genuinely live at once, not just by code inspection).
+#[test]
+fn reader_never_receives_observer_snapshot_while_player_streams() {
+    use bevy_ecs::system::RunSystemOnce;
+    use settled_reach_server::bridge::{send_bridge_snapshot, BridgeResource, SnapshotBuffer};
+
+    let (mut world, addr) = new_multi_connection_world();
+
+    // Player connects first (mirrors main.rs's structural first-connection
+    // path in spirit, though this test drives it through the SAME
+    // accept-loop the Reader below uses — the accept-loop must handle a
+    // Player exactly as well as a Reader, not just readers-after-a-
+    // pre-existing-Player).
+    let player_handle = thread::spawn(move || {
+        let stream = TcpStream::connect(addr).expect("failed to connect (player)");
+        client_handshake(stream, ConnectionRole::Player)
+    });
+    drive_accept_loop_until(&mut world, |w| w.resource::<BridgeResource>().has_player());
+    let mut player_stream = player_handle.join().expect("player thread panicked");
+
+    let reader_handle = thread::spawn(move || {
+        let stream = TcpStream::connect(addr).expect("failed to connect (reader)");
+        client_handshake(stream, ConnectionRole::Reader)
+    });
+    drive_accept_loop_until(&mut world, |w| {
+        w.resource::<BridgeResource>().reader_count() == 1
+    });
+    let mut reader_stream = reader_handle.join().expect("reader thread panicked");
+    // Non-blocking so a read that would otherwise hang forever (the whole
+    // point being tested — nothing ever arrives) returns WouldBlock instead.
+    reader_stream
+        .set_nonblocking(true)
+        .expect("failed to set reader stream non-blocking");
+
+    world.resource_mut::<SnapshotBuffer>().snapshot = Some(sample_snapshot(7));
+    world
+        .run_system_once(send_bridge_snapshot)
+        .expect("send_bridge_snapshot failed to run");
+
+    // Player DID get the snapshot — establishes the positive control so a
+    // trivially-broken send path (e.g. send_bridge_snapshot silently
+    // no-op'ing for everyone) can't masquerade as "reader correctly got
+    // nothing".
+    let payload = read_framed(&mut player_stream)
+        .expect("failed to read player frame")
+        .expect("unexpected EOF reading player snapshot");
+    let snapshot: ObserverSnapshot = rmp_serde::from_slice(&payload).expect("failed to decode");
+    assert_eq!(snapshot.tick, 7);
+
+    // Reader got NOTHING — not WouldBlock-then-eventually-something, a
+    // sustained absence over a real wall-clock window.
+    let check_until = std::time::Instant::now() + std::time::Duration::from_millis(300);
+    while std::time::Instant::now() < check_until {
+        let mut probe = [0u8; 1];
+        match std::io::Read::read(&mut reader_stream, &mut probe) {
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
+            Ok(0) => panic!("reader stream unexpectedly closed"),
+            other => panic!(
+                "reader received unexpected data/result while player streamed: {:?}",
+                other
+            ),
+        }
+        thread::sleep(std::time::Duration::from_millis(10));
+    }
+}
+
+/// T-1130 behavior 5: a Reader sending the forbidden `Vec<PlayerInput>`
+/// shape is dropped (not forwarded to `InputQueue`) on each offense, and
+/// disconnected once it crosses the strike threshold — never on the first
+/// offense (D-254 §2: "log + drop on first offense, repeated = disconnect").
+#[test]
+fn forbidden_reader_input_is_dropped_then_disconnected() {
+    use bevy_ecs::system::RunSystemOnce;
+    use settled_reach_server::bridge::{receive_bridge_inputs, BridgeResource};
+    use settled_reach_server::simulation::input::InputQueue;
+
+    let (mut world, addr) = new_multi_connection_world();
+
+    let client_handle = thread::spawn(move || {
+        let stream = TcpStream::connect(addr).expect("failed to connect");
+        client_handshake(stream, ConnectionRole::Reader)
+    });
+    drive_accept_loop_until(&mut world, |w| {
+        w.resource::<BridgeResource>().reader_count() == 1
+    });
+    let mut stream = client_handle.join().expect("client thread panicked");
+
+    let send_one_input_frame = |stream: &mut TcpStream| {
+        let inputs = vec![PlayerInput {
+            tick: 1,
+            action: PlayerAction::MoveNorth,
+        }];
+        let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize");
+        write_framed(stream, &payload).expect("failed to write input frame");
+    };
+
+    // Strikes 1 and 2: the reader survives, but nothing reaches InputQueue.
+    for strike in 1..=2 {
+        send_one_input_frame(&mut stream);
+        // Wall-clock settle for TCP delivery, then drain — mirrors this
+        // file's existing pattern (see single_tick_drains_all_ready_inbound_frames).
+        thread::sleep(std::time::Duration::from_millis(50));
+        world
+            .run_system_once(receive_bridge_inputs)
+            .expect("receive_bridge_inputs failed to run");
+        assert_eq!(
+            world.resource::<InputQueue>().len(),
+            0,
+            "forbidden reader input must never reach InputQueue (strike {})",
+            strike
+        );
+        assert_eq!(
+            world.resource::<BridgeResource>().reader_count(),
+            1,
+            "reader must survive strike {} (below disconnect threshold)",
+            strike
+        );
+    }
+
+    // Strike 3 crosses READER_VIOLATION_DISCONNECT_THRESHOLD (3) — the
+    // reader is disconnected.
+    send_one_input_frame(&mut stream);
+    thread::sleep(std::time::Duration::from_millis(50));
+    world
+        .run_system_once(receive_bridge_inputs)
+        .expect("receive_bridge_inputs failed to run");
+    assert_eq!(
+        world.resource::<InputQueue>().len(),
+        0,
+        "forbidden reader input must never reach InputQueue, even on the disconnecting strike"
+    );
+    assert_eq!(
+        world.resource::<BridgeResource>().reader_count(),
+        0,
+        "reader must be disconnected after crossing the violation threshold"
+    );
+}
+
+/// T-1130 behavior 6 (THE CRITICAL FIX): a Reader disconnecting must leave
+/// a running Player session completely unaffected — `ServerRunning` stays
+/// true, and the Player connection keeps working (proven by successfully
+/// sending it a snapshot AFTER the reader is gone, not just by inspecting
+/// the flag).
+#[test]
+fn reader_disconnect_does_not_affect_running_player_session() {
+    use bevy_ecs::system::RunSystemOnce;
+    use settled_reach_server::bridge::{
+        receive_bridge_inputs, send_bridge_snapshot, BridgeResource, ServerRunning, SnapshotBuffer,
+    };
+
+    let (mut world, addr) = new_multi_connection_world();
+
+    let player_handle = thread::spawn(move || {
+        let stream = TcpStream::connect(addr).expect("failed to connect (player)");
+        client_handshake(stream, ConnectionRole::Player)
+    });
+    drive_accept_loop_until(&mut world, |w| w.resource::<BridgeResource>().has_player());
+    let mut player_stream = player_handle.join().expect("player thread panicked");
+
+    let reader_handle = thread::spawn(move || {
+        let stream = TcpStream::connect(addr).expect("failed to connect (reader)");
+        client_handshake(stream, ConnectionRole::Reader)
+    });
+    drive_accept_loop_until(&mut world, |w| {
+        w.resource::<BridgeResource>().reader_count() == 1
+    });
+    let reader_stream = reader_handle.join().expect("reader thread panicked");
+
+    // The reader disconnects (drop closes the TCP connection — clean EOF).
+    drop(reader_stream);
+
+    // Drain: the server observes the reader's EOF and removes it.
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        world
+            .run_system_once(receive_bridge_inputs)
+            .expect("receive_bridge_inputs failed to run");
+        if world.resource::<BridgeResource>().reader_count() == 0 {
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "timed out waiting for reader disconnect to be observed"
+        );
+        thread::sleep(std::time::Duration::from_millis(10));
+    }
+
+    assert!(
+        world.resource::<ServerRunning>().0,
+        "a reader's disconnect must never flip ServerRunning"
+    );
+    assert!(
+        world.resource::<BridgeResource>().has_player(),
+        "the player connection must still be installed after the reader disconnects"
+    );
+
+    // Prove the player session is still genuinely functional, not just
+    // structurally present: send it a snapshot and read it back.
+    world.resource_mut::<SnapshotBuffer>().snapshot = Some(sample_snapshot(99));
+    world
+        .run_system_once(send_bridge_snapshot)
+        .expect("send_bridge_snapshot failed to run");
+    let payload = read_framed(&mut player_stream)
+        .expect("failed to read post-disconnect player frame")
+        .expect("unexpected EOF — player session was affected by reader disconnect");
+    let snapshot: ObserverSnapshot = rmp_serde::from_slice(&payload).expect("failed to decode");
+    assert_eq!(
+        snapshot.tick, 99,
+        "player session must remain fully functional after reader disconnect"
+    );
+    assert!(
+        world.resource::<ServerRunning>().0,
+        "sending to the still-live player must not flip ServerRunning either"
+    );
+}
+
+/// T-1130 behavior 7: a second connection attempting `role: Player` while a
+/// Player is already connected gets a clean, immediate rejection (the
+/// connection closes right after its handshake completes) — NOT the
+/// original starvation bug (silent hang forever), and NOT a crash or
+/// disruption to the existing Player's session.
+#[test]
+fn second_player_attempt_is_cleanly_rejected() {
+    use bevy_ecs::system::RunSystemOnce;
+    use settled_reach_server::bridge::BridgeResource;
+
+    let (mut world, addr) = new_multi_connection_world();
+
+    let first_handle = thread::spawn(move || {
+        let stream = TcpStream::connect(addr).expect("failed to connect (first player)");
+        client_handshake(stream, ConnectionRole::Player)
+    });
+    drive_accept_loop_until(&mut world, |w| w.resource::<BridgeResource>().has_player());
+    let _first_stream = first_handle.join().expect("first player thread panicked");
+
+    let second_handle = thread::spawn(move || {
+        let mut stream = TcpStream::connect(addr).expect("failed to connect (second player)");
+        // Read handshake and send StartupMessage{role: Player} exactly like
+        // a normal client — the rejection happens AFTER this, not by
+        // refusing the handshake itself (the connection is genuinely
+        // accepted and handshaken; it's the ROLE promotion that's refused).
+        stream = client_handshake(stream, ConnectionRole::Player);
+        // The server closes the connection right after — prove it's a
+        // clean disconnect (not the original silent-hang bug) by reading
+        // until EOF. read_framed blocks here (this stream is never set
+        // non-blocking), which is deliberate: it's this thread's own proof
+        // that the disconnect actually happens — if the original starvation
+        // bug were still present, this call would hang forever with no
+        // internal deadline of its own. The OUTER test loop below bounds
+        // total test time via `second_handle.is_finished()` polling against
+        // ITS OWN 5s deadline, so a regression here still fails the test in
+        // bounded time rather than hanging the test suite.
+        match read_framed(&mut stream) {
+            Ok(None) => {} // clean EOF — the expected rejection signal
+            Ok(Some(_)) => panic!("second player attempt must never receive a message"),
+            Err(_) => {} // connection reset also counts as "rejected, not hung"
+        }
+    });
+
+    // Drive the accept-loop until the second connection has been processed
+    // (it will never appear as a reader OR a second player — reader_count
+    // stays 0 and has_player stays true throughout, which is exactly the
+    // rejection this test verifies).
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+    while !second_handle.is_finished() {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "timed out waiting for second player rejection to complete"
+        );
+        world
+            .run_system_once(settled_reach_server::bridge::accept_new_connections)
+            .expect("accept_new_connections failed to run");
+        thread::sleep(std::time::Duration::from_millis(5));
+    }
+    second_handle.join().expect("second player thread panicked");
+
+    assert!(
+        world.resource::<BridgeResource>().has_player(),
+        "the original player connection must be untouched by the rejected second attempt"
+    );
+    assert_eq!(
+        world.resource::<BridgeResource>().reader_count(),
+        0,
+        "a rejected second-Player attempt must never be silently installed as a reader"
+    );
+}
+
+/// Minimal `ObserverSnapshot` for the tests above — same shape as
+/// `snapshot_roundtrip_over_tcp`'s at the top of this file, parameterized
+/// only by `tick` (the one field these tests assert on).
+fn sample_snapshot(tick: u64) -> ObserverSnapshot {
+    ObserverSnapshot {
+        tick,
+        game_time: GameTime {
+            day: 0,
+            time_of_day: 0,
+            day_phase: DayPhase::Morning,
+            tick_rate: TickRate::Full,
+        },
+        player_facing: FacingDirection::North,
+        player_stance: MovementStance::default(),
+        player_inventory: vec![],
+        entities: vec![],
+        visible_tiles: vec![],
+        nearby_interactions: vec![],
+        current_monologue: None,
+        pending_recognitions: vec![],
+        dialogue_response: None,
+        blocked_entities: vec![],
+        scan_events: vec![],
+        sound_events: vec![],
+        follow_state: None,
+        character_pressure: None,
+        rng_seed: None,
+        poi_list: vec![],
+        examine_result: None,
+        player_knowledge: None,
+        save_result: None,
+        triangle_crisis_events: vec![],
+        state_hash: None,
+        debug_response: None,
+        sim_errors: vec![],
+        current_ticker: None,
+        settings_response: None,
+        economy_snapshot: None,
+        bookmark_catalog: None,
+    }
+}

server/tests/layer3.rs

@@ -81,7 +81,10 @@ fn server_subprocess_sends_snapshot_on_connect() {
         rmp_serde::from_slice(&handshake_frame).expect("deserialize HandshakeMessage");
 
     // 5. Send StartupMessage with world_seed (#175)
-    let startup = StartupMessage { world_seed: 42 };
+    let startup = StartupMessage {
+        world_seed: 42,
+        role: ConnectionRole::Player,
+    };
     let startup_payload = rmp_serde::to_vec_named(&startup).expect("serialize StartupMessage");
     write_framed(&mut writer, &startup_payload).expect("send StartupMessage to server");