settled-reach/server/src/bridge/mod.rs

// Bridge module - Client-server communication
// Implements D-020 subprocess/IPC architecture
// MessagePack serialization for Rust<->Godot communication

use bevy_app::prelude::*;
use bevy_ecs::prelude::*;
use bevy_ecs::schedule::IntoScheduleConfigs;

use crate::atlas::atlas_data_proxy::{
    CityNamesRequest, CityNamesResponse, StarMapRequest, StarMapResponse,
};
use crate::atlas::browse_proxy::{BrowseRequest, BrowseResponse};
use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse};
use crate::atlas::step_canvas::{StepCanvasRequest, StepCanvasResponse};
use crate::bridge::tcp::TcpBridge;

pub mod debug;
pub mod framing;
pub mod local;
pub mod tcp;
pub mod text_renderer;
pub mod types;
pub use types::*;

/// Error type for bridge operations
#[derive(Debug, thiserror::Error)]
pub enum BridgeError {
    #[error("serialization error: {0}")]
    Serialization(#[from] rmp_serde::encode::Error),
    #[error("deserialization error: {0}")]
    Deserialization(#[from] rmp_serde::decode::Error),
    #[error("deserialization error (raw bytes logged): {0}")]
    DeserializationWithDump(String),
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
    #[error("transport error: {0}")]
    Transport(String),
    #[error("client disconnected")]
    Disconnected,
    #[error("internal mutex poisoned: {0}")]
    MutexPoisoned(String),
}

/// One decoded inbound message. The client→server stream is a single demuxed
/// channel (D-225): a `Vec<PlayerInput>` frame is a MessagePack *array* and
/// every request type below is a *map*, so array vs. map alone separates
/// inputs from everything else without a wire-level type tag (existing frames
/// are byte-unchanged — additive).
///
/// **Disambiguating the three map shapes (D-225 extension, T-949):**
/// `AtlasLayerRequest{body_id, up_to}` was the only map shape until T-949
/// added `StarMapRequest`/`CityNamesRequest` alongside it. serde's derived
/// `Deserialize` silently ignores unknown fields by default, so "does this
/// struct parse at all" is not a safe discriminator once more than one map
/// shape can share a field name (`CityNamesRequest` and `AtlasLayerRequest`
/// both key on `body_id`) — a payload carrying every field either shape wants
/// would ambiguously satisfy both. Rather than retrofit
/// `#[serde(deny_unknown_fields)]` onto the existing `AtlasLayerRequest` (risking
/// breakage if any already-deployed client encoder harmlessly sends extra
/// fields), the two *new* map shapes each carry a mandatory boolean
/// discriminator field the others don't have at all (`star_map` /
/// `city_names`): a missing required field is a hard deserialize failure, not
/// a silent ignore, so no *minimal well-formed* instance of one shape
/// satisfies another — and [`decode_inbound`] additionally REJECTS union
/// frames that carry more than one shape's discriminators outright (PR #176
/// review H1). `AtlasLayerRequest` itself is untouched byte-for-byte.
///
/// **Ceiling (D-225 trajectory, DISCHARGED by D-255(c)):** [`BrowseRequest`]
/// (T-1131) was the FIFTH map shape and, per the ceiling this doc used to
/// call at four, the last one this hand-rolled scheme was meant to carry.
/// [`StepCanvasRequest`] (T-1181) is that predicted SIXTH shape — the
/// step-canvas payload is genuinely new (D-255(c): 21×-563× the legacy
/// `district_window` carrier's ~30 KB reference, a cell-count gap no
/// encoding closes), so per this doc's own prior instruction it extends the
/// SAME tagged-marker pattern one more time (`step_canvas: bool`) rather
/// than inventing a parallel envelope format — this IS the tagged-envelope
/// migration D-225's 2026-06-12 amendment deferred, now executed. The
/// pattern generalizes cleanly because every one of the five prior shapes
/// already carries (or, for `AtlasLayerRequest`, is disambiguated against)
/// exactly this kind of required marker field — there is no natural ceiling
/// on the DEMUX mechanism itself, only a discipline reminder that a new
/// shape should justify why it can't ride an existing one (as
/// `StepCanvasRequest` does, D-255(c)).
#[derive(Debug)]
pub enum Inbound {
    /// A batch of player inputs (the gameplay path).
    Inputs(Vec<PlayerInput>),
    /// An atlas layer-stream request (#969, D-225).
    AtlasRequest(AtlasLayerRequest),
    /// A star-map dataset request (T-949a).
    StarMapRequest(StarMapRequest),
    /// A per-body city-names request (T-949b).
    CityNamesRequest(CityNamesRequest),
    /// A data-browser request — one of the six D-254 §4 v1 entity kinds
    /// (T-1131).
    BrowseRequest(BrowseRequest),
    /// A D-255(a) step-canvas data-canvas request (T-1181, the D-225
    /// tagged-envelope migration, executed).
    StepCanvasRequest(StepCanvasRequest),
}

/// Key-presence probe for the defensive multi-shape check in
/// [`decode_inbound`]: `Option<IgnoredAny>` records whether a key exists
/// without caring about its value or type, so a union frame is detected even
/// when the individual values wouldn't parse as their target types.
#[derive(serde::Deserialize)]
struct ShapeProbe {
    body_id: Option<serde::de::IgnoredAny>,
    up_to: Option<serde::de::IgnoredAny>,
    star_map: Option<serde::de::IgnoredAny>,
    city_names: Option<serde::de::IgnoredAny>,
    browse: Option<serde::de::IgnoredAny>,
    step_canvas: Option<serde::de::IgnoredAny>,
}

/// Demux a received frame payload into an [`Inbound`] (D-225, T-949, T-1131,
/// T-1181). Tries, in order: `Vec<PlayerInput>` (array) → `AtlasLayerRequest`
/// (map, `body_id`+`up_to`) → `StarMapRequest` (map, `star_map`
/// discriminator) → `CityNamesRequest` (map, `city_names` discriminator +
/// `body_id`) → `BrowseRequest` (map, `browse` discriminator) →
/// `StepCanvasRequest` (map, `step_canvas` discriminator).
///
/// Mutual exclusivity is enforced, not assumed: no minimal well-formed
/// instance of one shape satisfies another (see the [`Inbound`] doc), and a
/// defensive pre-check rejects any map frame carrying the discriminators of
/// more than one shape — e.g. a buggy encoder emitting
/// `{"star_map": true, "city_names": true, ...}` — instead of silently
/// routing it to whichever shape is tried first (PR #176 review H1). A frame
/// satisfying none of the six shapes is a genuinely malformed input frame.
pub fn decode_inbound(payload: &[u8]) -> Result<Inbound, BridgeError> {
    if let Ok(inputs) = rmp_serde::from_slice::<Vec<PlayerInput>>(payload) {
        return Ok(Inbound::Inputs(inputs));
    }
    // Defensive multi-shape rejection: serde ignores unknown fields, so a
    // union frame would otherwise route silently by try-order. Unreachable
    // from the shipped client encoders (each sends one minimal shape) — this
    // guards buggy or adversarial frames.
    if let Ok(probe) = rmp_serde::from_slice::<ShapeProbe>(payload) {
        let atlas = probe.body_id.is_some() && probe.up_to.is_some();
        let star_map = probe.star_map.is_some();
        let city_names = probe.city_names.is_some();
        let browse = probe.browse.is_some();
        let step_canvas = probe.step_canvas.is_some();
        let shapes = usize::from(atlas)
            + usize::from(star_map)
            + usize::from(city_names)
            + usize::from(browse)
            + usize::from(step_canvas);
        if shapes > 1 {
            let dump_len = payload.len().min(256);
            tracing::error!(
                "inbound frame matches {} request shapes at once (atlas={}, star_map={}, city_names={}, browse={}, step_canvas={}) — rejecting ambiguous frame. Raw ({} of {} bytes): {:02x?}",
                shapes,
                atlas,
                star_map,
                city_names,
                browse,
                step_canvas,
                dump_len,
                payload.len(),
                &payload[..dump_len]
            );
            return Err(BridgeError::DeserializationWithDump(format!(
                "ambiguous inbound frame matches {shapes} request shapes (payload {} bytes)",
                payload.len()
            )));
        }
    }
    if let Ok(req) = rmp_serde::from_slice::<AtlasLayerRequest>(payload) {
        return Ok(Inbound::AtlasRequest(req));
    }
    if let Ok(req) = rmp_serde::from_slice::<StarMapRequest>(payload) {
        return Ok(Inbound::StarMapRequest(req));
    }
    if let Ok(req) = rmp_serde::from_slice::<CityNamesRequest>(payload) {
        return Ok(Inbound::CityNamesRequest(req));
    }
    if let Ok(req) = rmp_serde::from_slice::<BrowseRequest>(payload) {
        return Ok(Inbound::BrowseRequest(req));
    }
    match rmp_serde::from_slice::<StepCanvasRequest>(payload) {
        Ok(req) => Ok(Inbound::StepCanvasRequest(req)),
        Err(e) => {
            let dump_len = payload.len().min(256);
            tracing::error!(
                "inbound decode failed (matches no known frame shape): {}. Raw ({} of {} bytes): {:02x?}",
                e,
                dump_len,
                payload.len(),
                &payload[..dump_len]
            );
            Err(BridgeError::DeserializationWithDump(format!(
                "{e} (payload {} bytes)",
                payload.len()
            )))
        }
    }
}

/// Abstracts transport layer (D-020)
/// Implemented by LocalBridge (stdio) and future NetworkBridge
pub trait SimBridge: Send + Sync {
    /// Send the protocol handshake as the first framed message (#555).
    /// Must be called exactly once, immediately after connection, before
    /// any ObserverSnapshot is sent.
    fn send_handshake(&self) -> Result<(), BridgeError>;

    /// Receive the client's startup message containing the world seed (#175).
    /// Called exactly once, after send_handshake(), before entering the tick loop.
    /// Blocks until the client sends the message.
    fn receive_startup(&self) -> Result<StartupMessage, BridgeError>;

    /// Send an observer snapshot to the client
    fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError>;

    /// Receive one inbound message, or `None` if no complete frame is ready.
    /// The single client→server stream is demuxed by frame shape (D-225).
    /// `receive_bridge_inputs` loops this until `None` (T-1045), so the
    /// transport behind `BridgeResource` must not block when no frame is
    /// buffered (TcpBridge is non-blocking; LocalBridge blocks — test-only).
    fn receive(&self) -> Result<Option<Inbound>, BridgeError>;

    /// Send an atlas layer-stream response to the client (#969, D-225).
    fn send_atlas_response(&self, resp: &AtlasLayerResponse) -> Result<(), BridgeError>;

    /// Send a star-map response to the client (T-949a).
    fn send_star_map_response(&self, resp: &StarMapResponse) -> Result<(), BridgeError>;

    /// Send a city-names response to the client (T-949b).
    fn send_city_names_response(&self, resp: &CityNamesResponse) -> Result<(), BridgeError>;

    /// Send a browse response to the client (T-1131).
    fn send_browse_response(&self, resp: &BrowseResponse) -> Result<(), BridgeError>;

    /// Send a step-canvas response to the client (T-1181, D-255(c)).
    fn send_step_canvas_response(&self, resp: &StepCanvasResponse) -> Result<(), BridgeError>;
}

/// 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 {
    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 {
        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())),
        }
    }

    /// 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> {
        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> {
        match &self.player {
            Some(c) => c.bridge.send_snapshot(snapshot),
            None => Err(BridgeError::Disconnected),
        }
    }

    /// 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)
    }

    /// 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(())
            }
        }
    }

    /// 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(())
            }
        }
    }

    /// 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(())
            }
        }
    }

    /// Send a browse response to exactly the connection that requested it
    /// (T-1131 — same per-connection routing D-254 §2 established for
    /// atlas/star-map/city-names).
    pub fn send_browse_response_to(
        &self,
        id: ConnectionId,
        resp: &BrowseResponse,
    ) -> Result<(), BridgeError> {
        match self.connection(id) {
            Some(c) => c.bridge.send_browse_response(resp),
            None => {
                tracing::debug!(
                    "browse response for {:?} dropped — connection {:?} no longer present",
                    resp.kind,
                    id
                );
                Ok(())
            }
        }
    }

    /// Send a step-canvas response to exactly the connection that requested
    /// it (T-1181 — same per-connection routing D-254 §2 established for
    /// atlas/star-map/city-names/browse).
    pub fn send_step_canvas_response_to(
        &self,
        id: ConnectionId,
        resp: &StepCanvasResponse,
    ) -> Result<(), BridgeError> {
        match self.connection(id) {
            Some(c) => c.bridge.send_step_canvas_response(resp),
            None => {
                tracing::debug!(
                    "step canvas response for {} dropped — connection {:?} no longer present",
                    resp.body_id,
                    id
                );
                Ok(())
            }
        }
    }
}

/// Tracks whether the protocol handshake has been sent (#555).
/// Inserted by BridgePlugin as Pending. Set to Complete in main.rs after
/// `send_handshake()` succeeds. `receive_bridge_inputs` logs a warning
/// if inputs arrive while still Pending.
#[derive(Resource, Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum HandshakeState {
    /// Handshake not yet sent. Inputs arriving in this state trigger a warning.
    #[default]
    Pending,
    /// Handshake sent. Normal operation.
    Complete,
}

/// Per-tick cap on drained inbound frames (T-1045) — a safety valve so a
/// client flooding the stream cannot starve the simulation tick. Generous:
/// normal traffic is one input batch plus the occasional atlas request.
const MAX_INBOUND_FRAMES_PER_TICK: usize = 64;

/// Receive inputs/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<ResMut<BridgeResource>>,
    mut input_queue: ResMut<crate::simulation::input::InputQueue>,
    mut running: ResMut<ServerRunning>,
    handshake: Res<HandshakeState>,
    mut error_buffer: ResMut<SimErrorBuffer>,
    mut atlas_requests: ResMut<AtlasRequestBuffer>,
    mut star_map_requests: ResMut<StarMapRequestBuffer>,
    mut city_names_requests: ResMut<CityNamesRequestBuffer>,
    mut browse_requests: ResMut<BrowseRequestBuffer>,
    mut step_canvas_requests: ResMut<StepCanvasRequestBuffer>,
    time: Option<Res<crate::simulation::time::SimulationTime>>,
) {
    let Some(mut bridge) = bridge else { return };
    let current_tick = time.as_ref().map(|t| t.tick).unwrap_or(0);

    // -- 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));
                }
                Ok(Some(Inbound::BrowseRequest(req))) => {
                    browse_requests.0.push((player_id, req));
                }
                Ok(Some(Inbound::StepCanvasRequest(req))) => {
                    step_canvas_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!(
                        "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;
                    }
                }
                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(Some(Inbound::BrowseRequest(req))) => {
                    browse_requests.0.push((reader_id, req));
                }
                Ok(Some(Inbound::StepCanvasRequest(req))) => {
                    step_canvas_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;
                }
                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;
                }
            }
        }
    }

    for id in disconnected_readers
        .into_iter()
        .chain(to_disconnect_for_violations)
    {
        bridge.remove_reader(id);
    }
}

/// 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<ResMut<BridgeResource>>,
    mut buffer: ResMut<SnapshotBuffer>,
    mut running: ResMut<ServerRunning>,
) {
    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!("Player disconnected during send, shutting down");
                }
                BridgeError::MutexPoisoned(msg) => {
                    tracing::error!("Player bridge mutex poisoned during send: {}", msg);
                }
                _ => {
                    tracing::error!("Player bridge send error: {}", e);
                }
            }
            running.0 = false;
            bridge.player = None;
        }
    }
}

/// Server running flag resource
#[derive(Resource, Debug, Clone)]
pub struct ServerRunning(pub bool);

impl Default for ServerRunning {
    fn default() -> Self {
        Self(true)
    }
}

/// Inbound atlas layer requests routed off the bridge (#969, D-225), drained by
/// 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<(ConnectionId, AtlasLayerRequest)>);

/// Outbound atlas layer responses, filled by the proxy serve system and flushed
/// to the client in `PostSnapshot` (#969, D-225). Connection-tagged (D-254 §2).
#[derive(Resource, Default)]
pub struct AtlasResponseBuffer(pub Vec<(ConnectionId, AtlasLayerResponse)>);

/// 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 (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`. Connection-tagged (D-254 §2).
#[derive(Resource, Default)]
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). Connection-tagged (D-254 §2).
#[derive(Resource, Default)]
pub struct StarMapResponseBuffer(pub Vec<(ConnectionId, StarMapResponse)>);

/// 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 (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`. Connection-tagged (D-254 §2).
#[derive(Resource, Default)]
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). Connection-tagged
/// (D-254 §2).
#[derive(Resource, Default)]
pub struct CityNamesResponseBuffer(pub Vec<(ConnectionId, CityNamesResponse)>);

/// 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 (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 {} to {:?}: {}",
                resp.body_id,
                id,
                e
            );
        }
    }
}

/// Inbound data-browser requests routed off the bridge (T-1131), drained by
/// the proxy serve system in `PreInput`. Connection-tagged (D-254 §2).
#[derive(Resource, Default)]
pub struct BrowseRequestBuffer(pub Vec<(ConnectionId, BrowseRequest)>);

/// Outbound browse responses, filled by the proxy serve system and flushed
/// to the client in `PostSnapshot` (T-1131). Connection-tagged (D-254 §2).
#[derive(Resource, Default)]
pub struct BrowseResponseBuffer(pub Vec<(ConnectionId, BrowseResponse)>);

/// Flush buffered browse responses to their requesting connections (T-1131 —
/// same per-connection routing D-254 §2 established for
/// atlas/star-map/city-names). A failed send is logged but not fatal.
pub fn send_browse_responses(
    bridge: Option<Res<BridgeResource>>,
    mut buffer: ResMut<BrowseResponseBuffer>,
) {
    let Some(bridge) = bridge else { return };
    for (id, resp) in buffer.0.drain(..) {
        if let Err(e) = bridge.send_browse_response_to(id, &resp) {
            tracing::warn!(
                "failed to send browse response for {:?} to {:?}: {}",
                resp.kind,
                id,
                e
            );
        }
    }
}

/// Inbound step-canvas requests routed off the bridge (T-1181), drained by
/// the proxy serve system in `PreInput`. Connection-tagged (D-254 §2).
#[derive(Resource, Default)]
pub struct StepCanvasRequestBuffer(pub Vec<(ConnectionId, StepCanvasRequest)>);

/// Outbound step-canvas responses, filled by the proxy serve system and
/// flushed to the client in `PostSnapshot` (T-1181). Connection-tagged
/// (D-254 §2).
#[derive(Resource, Default)]
pub struct StepCanvasResponseBuffer(pub Vec<(ConnectionId, StepCanvasResponse)>);

/// Flush buffered step-canvas responses to their requesting connections
/// (T-1181 — same per-connection routing D-254 §2 established for
/// atlas/star-map/city-names/browse). A failed send is logged but not fatal.
pub fn send_step_canvas_responses(
    bridge: Option<Res<BridgeResource>>,
    mut buffer: ResMut<StepCanvasResponseBuffer>,
) {
    let Some(bridge) = bridge else { return };
    for (id, resp) in buffer.0.drain(..) {
        if let Err(e) = bridge.send_step_canvas_response_to(id, &resp) {
            tracing::warn!(
                "failed to send step canvas 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;

impl Plugin for BridgePlugin {
    fn build(&self, app: &mut App) {
        use crate::simulation::time::sim_not_paused;
        use crate::tick_phases::TickPhase;

        app.init_resource::<SnapshotBuffer>()
            .init_resource::<ServerRunning>()
            .init_resource::<HandshakeState>()
            .init_resource::<SimErrorBuffer>()
            .init_resource::<debug::DebugCommandBuffer>()
            .init_resource::<DebugEnabled>()
            .init_resource::<crate::perception::query::ActivePerceptionMode>()
            .init_resource::<AtlasRequestBuffer>()
            .init_resource::<AtlasResponseBuffer>()
            .init_resource::<StarMapRequestBuffer>()
            .init_resource::<StarMapResponseBuffer>()
            .init_resource::<CityNamesRequestBuffer>()
            .init_resource::<CityNamesResponseBuffer>()
            .init_resource::<BrowseRequestBuffer>()
            .init_resource::<BrowseResponseBuffer>()
            .init_resource::<StepCanvasRequestBuffer>()
            .init_resource::<StepCanvasResponseBuffer>()
            .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))
            .add_systems(Update, send_atlas_responses.in_set(TickPhase::PostSnapshot))
            .add_systems(
                Update,
                send_star_map_responses.in_set(TickPhase::PostSnapshot),
            )
            .add_systems(
                Update,
                send_city_names_responses.in_set(TickPhase::PostSnapshot),
            )
            .add_systems(
                Update,
                send_browse_responses.in_set(TickPhase::PostSnapshot),
            )
            .add_systems(
                Update,
                send_step_canvas_responses.in_set(TickPhase::PostSnapshot),
            )
            // Debug commands — Snapshot phase
            .add_systems(
                Update,
                debug::handle_debug_commands.in_set(TickPhase::Snapshot),
            )
            // Monologue chain — Simulation phase, strict intra-phase sequence.
            // trigger_event_monologue must run after conversations + sound (also Simulation).
            // T-970: TickPhase::Simulation is not set-gated (see
            // social_plugin.rs's collect_sound_events exemption) — this whole
            // chain is genuine world-advancing dialogue/monologue logic, so
            // it gates safely on its own. trigger_event_monologue's
            // .after(collect_sound_events) still holds while gated:
            // collect_sound_events itself is never gated, and an ordering
            // edge onto a skipped predecessor is trivially satisfied.
            .add_systems(
                Update,
                (
                    crate::simulation::monologue::trigger_monologue,
                    crate::simulation::monologue::trigger_recognition_monologue
                        .after(crate::simulation::monologue::trigger_monologue)
                        .after(crate::perception::anomaly::detect_anomalies),
                    crate::simulation::monologue::process_sprint_anomaly_monologue
                        .after(crate::simulation::monologue::trigger_recognition_monologue),
                    crate::simulation::monologue::trigger_event_monologue
                        .after(crate::simulation::monologue::process_sprint_anomaly_monologue)
                        .after(crate::simulation::sound::collect_sound_events)
                        .after(crate::simulation::dialogue::process_walk_away),
                    crate::simulation::monologue::process_contradiction_monologue
                        .after(crate::simulation::monologue::trigger_event_monologue),
                )
                    .run_if(sim_not_paused)
                    .in_set(TickPhase::Simulation),
            )
            // Observation systems — Simulation phase (reads positions, feeds snapshot).
            // T-970: gates safely on its own (see note above) — these compute
            // "current state" (visibility, nearby interactions) that's valid
            // as long as nothing moved, which holds while paused since
            // Movement is frozen too; unlike SoundEventQueue, nothing here
            // depends on being refreshed on a tick where nothing changed.
            .add_systems(
                Update,
                (
                    crate::perception::observer::compute_visibility_geometry,
                    crate::simulation::interaction::compute_nearby_interactions,
                )
                    .run_if(sim_not_paused)
                    .in_set(TickPhase::Simulation),
            )
            // Observer snapshot assembly — Snapshot phase
            .add_systems(
                Update,
                crate::perception::observer::compute_observer_snapshot.in_set(TickPhase::Snapshot),
            )
            // Post-snapshot: emit observation events
            .add_systems(
                Update,
                crate::perception::observation::emit_observation_events
                    .in_set(TickPhase::PostSnapshot),
            );

        tracing::debug!("BridgePlugin initialized");
    }
}

#[cfg(test)]
mod inbound_tests {
    use super::*;
    use crate::atlas::cascade::CascadeLayer;

    #[test]
    fn demux_routes_inputs_and_atlas_requests() {
        // A Vec<PlayerInput> frame (msgpack array) → Inbound::Inputs.
        let inputs: Vec<PlayerInput> = vec![];
        let frame = rmp_serde::to_vec_named(&inputs).unwrap();
        assert!(matches!(decode_inbound(&frame), Ok(Inbound::Inputs(v)) if v.is_empty()));

        // An AtlasLayerRequest frame (msgpack map) → Inbound::AtlasRequest.
        let req = AtlasLayerRequest {
            body_id: "GJ1c".into(),
            up_to: CascadeLayer::Topography,
            window_center: None,
            window_n: 0,
            window_granularity: 0,
            window_granularity_v2: None,
            window_min_wl_m: 0,
        };
        let frame = rmp_serde::to_vec_named(&req).unwrap();
        assert!(
            matches!(decode_inbound(&frame), Ok(Inbound::AtlasRequest(r)) if r.body_id == "GJ1c")
        );

        // Neither shape → a malformed-frame error.
        assert!(decode_inbound(&[0xff, 0xff]).is_err());
    }

    #[test]
    fn demux_routes_star_map_requests() {
        let req = StarMapRequest { star_map: true };
        let frame = rmp_serde::to_vec_named(&req).unwrap();
        assert!(matches!(
            decode_inbound(&frame),
            Ok(Inbound::StarMapRequest(r)) if r.star_map
        ));
    }

    #[test]
    fn demux_routes_city_names_requests() {
        let req = CityNamesRequest {
            city_names: true,
            body_id: "GJ1c".into(),
        };
        let frame = rmp_serde::to_vec_named(&req).unwrap();
        assert!(matches!(
            decode_inbound(&frame),
            Ok(Inbound::CityNamesRequest(r)) if r.body_id == "GJ1c"
        ));
    }

    /// T-949: the array-vs-map trick (D-225) still separates `Inputs` from
    /// everything else, and the three map shapes' discriminator fields keep
    /// them mutually exclusive — each of the four frame shapes decodes to
    /// exactly its own `Inbound` variant, never a neighbor's.
    #[test]
    fn inbound_disambiguation_is_unambiguous_across_all_four_shapes() {
        let inputs_frame = rmp_serde::to_vec_named(&Vec::<PlayerInput>::new()).unwrap();
        let atlas_frame = rmp_serde::to_vec_named(&AtlasLayerRequest {
            body_id: "GJ1c".into(),
            up_to: CascadeLayer::Topography,
            window_center: None,
            window_n: 0,
            window_granularity: 0,
            window_granularity_v2: None,
            window_min_wl_m: 0,
        })
        .unwrap();
        let star_map_frame = rmp_serde::to_vec_named(&StarMapRequest { star_map: true }).unwrap();
        let city_names_frame = rmp_serde::to_vec_named(&CityNamesRequest {
            city_names: true,
            body_id: "GJ1c".into(),
        })
        .unwrap();

        assert!(matches!(
            decode_inbound(&inputs_frame),
            Ok(Inbound::Inputs(_))
        ));
        assert!(matches!(
            decode_inbound(&atlas_frame),
            Ok(Inbound::AtlasRequest(_))
        ));
        assert!(matches!(
            decode_inbound(&star_map_frame),
            Ok(Inbound::StarMapRequest(_))
        ));
        assert!(matches!(
            decode_inbound(&city_names_frame),
            Ok(Inbound::CityNamesRequest(_))
        ));

        // Cross-check: an AtlasLayerRequest frame must NOT decode as
        // CityNamesRequest even though both key on `body_id` — the missing
        // `city_names` discriminator makes that a hard failure, not a silent
        // "extra field ignored" success either shape could show without it.
        assert!(rmp_serde::from_slice::<CityNamesRequest>(&atlas_frame).is_err());
        // And a CityNamesRequest frame must NOT decode as AtlasLayerRequest —
        // it's missing the required `up_to` field.
        assert!(rmp_serde::from_slice::<AtlasLayerRequest>(&city_names_frame).is_err());
    }

    /// PR #176 review H1: a union frame carrying more than one shape's
    /// discriminators must be REJECTED, not silently routed to whichever
    /// shape `decode_inbound` happens to try first.
    #[test]
    fn ambiguous_union_frame_is_rejected() {
        #[derive(serde::Serialize)]
        struct StarAndCity {
            star_map: bool,
            city_names: bool,
            body_id: String,
        }
        let frame = rmp_serde::to_vec_named(&StarAndCity {
            star_map: true,
            city_names: true,
            body_id: "GJ1c".into(),
        })
        .unwrap();
        assert!(
            decode_inbound(&frame).is_err(),
            "star_map+city_names union frame must be rejected"
        );

        #[derive(serde::Serialize)]
        struct AtlasAndStar {
            body_id: String,
            up_to: CascadeLayer,
            star_map: bool,
        }
        let frame = rmp_serde::to_vec_named(&AtlasAndStar {
            body_id: "GJ1c".into(),
            up_to: CascadeLayer::Topography,
            star_map: true,
        })
        .unwrap();
        assert!(
            decode_inbound(&frame).is_err(),
            "atlas+star_map union frame must be rejected"
        );

        // T-1131 (PR #184 review): the FIFTH shape's discriminator (`browse`)
        // must participate in the same union rejection — a well-formed
        // BrowseRequest smuggling another shape's discriminator alongside it
        // is rejected, not routed to whichever probe wins.
        #[derive(serde::Serialize)]
        struct BrowseAndStar {
            browse: bool,
            kind: crate::atlas::browse_proxy::BrowseEntityKind,
            query: crate::atlas::browse_proxy::BrowseQuery,
            star_map: bool,
        }
        let frame = rmp_serde::to_vec_named(&BrowseAndStar {
            browse: true,
            kind: crate::atlas::browse_proxy::BrowseEntityKind::StarSystem,
            query: crate::atlas::browse_proxy::BrowseQuery::Index {
                filter_system_id: None,
            },
            star_map: true,
        })
        .unwrap();
        assert!(
            decode_inbound(&frame).is_err(),
            "browse+star_map union frame must be rejected"
        );

        #[derive(serde::Serialize)]
        struct BrowseAndCity {
            browse: bool,
            kind: crate::atlas::browse_proxy::BrowseEntityKind,
            query: crate::atlas::browse_proxy::BrowseQuery,
            city_names: bool,
            body_id: String,
        }
        let frame = rmp_serde::to_vec_named(&BrowseAndCity {
            browse: true,
            kind: crate::atlas::browse_proxy::BrowseEntityKind::Body,
            query: crate::atlas::browse_proxy::BrowseQuery::Detail { id: "GJ1c".into() },
            city_names: true,
            body_id: "GJ1c".into(),
        })
        .unwrap();
        assert!(
            decode_inbound(&frame).is_err(),
            "browse+city_names union frame must be rejected"
        );
    }
}