Merge remote-tracking branch 'origin/atlas-companion-app'

2026-07-17 09:32:07 +02:00
parent 7baac29861 9b5c113961
commit a334c7c237
19 changed files with 2260 additions and 165 deletions
@@ -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());
    }

@@ -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,231 @@ 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 +1008,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))
@@ -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
+            }
+        }
+    }
+}
@@ -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.
@@ -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);