settled-reach/server/tests/bridge_tcp.rs

//! Integration tests for TcpBridge over TCP localhost (D-030 Layer 2: IPC roundtrip).

use settled_reach_server::atlas::browse_proxy::{
    BrowseEntityKind, BrowseQuery, BrowseRequest, BrowseResponse, BrowseStatus,
};
use settled_reach_server::bridge::framing::{read_framed, write_framed};
use settled_reach_server::bridge::tcp::TcpBridge;
use settled_reach_server::bridge::types::*;
use settled_reach_server::bridge::{BridgeResource, Inbound, SimBridge};
use settled_reach_server::simulation::time::{DayPhase, TickRate};
use std::net::{TcpListener, TcpStream};
use std::thread;

#[test]
fn snapshot_roundtrip_over_tcp() {
    // Bind listener first — port is guaranteed ready before spawning threads
    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
    let server_addr = listener.local_addr().expect("failed to get local address");

    // Server thread: accept on pre-bound listener and send snapshot
    let server_handle = thread::spawn(move || {
        let bridge = TcpBridge::accept_on(listener).expect("failed to accept");

        let snapshot = ObserverSnapshot {
            tick: 42,
            game_time: GameTime {
                day: 0,
                time_of_day: 0,
                day_phase: DayPhase::Morning,
                tick_rate: TickRate::Full,
            },
            player_facing: FacingDirection::North,
            player_stance: MovementStance::default(),
            player_inventory: vec![],
            entities: vec![VisibleEntity {
                entity_id: 100,
                x: 10.5,
                y: 20.3,
                z: 0,
                kind: EntityKind::Npc,
                visibility: VisibilitySector::Forward,
                relationship: RelationshipState::Unknown,
                observation: EntityVisibility::Visible,
                tell_state: None,
            }],
            visible_tiles: vec![],
            nearby_interactions: vec![],
            current_monologue: None,
            pending_recognitions: vec![],
            dialogue_response: None,
            blocked_entities: vec![],
            scan_events: vec![],
            sound_events: vec![],
            follow_state: None,
            character_pressure: None,
            rng_seed: None,
            poi_list: vec![],
            examine_result: None,
            player_knowledge: None,
            save_result: None,
            triangle_crisis_events: vec![],
            state_hash: None,
            debug_response: None,
            sim_errors: vec![],
            current_ticker: None,
            settings_response: None,
            economy_snapshot: None,
            bookmark_catalog: None,
        };

        bridge
            .send_snapshot(&snapshot)
            .expect("failed to send snapshot");
    });

    // Client: connect and receive snapshot (no sleep needed — listener already bound)
    let stream = TcpStream::connect(server_addr).expect("failed to connect");
    let mut reader = std::io::BufReader::new(stream);

    let payload = read_framed(&mut reader)
        .expect("failed to read frame")
        .expect("unexpected EOF");

    let snapshot: ObserverSnapshot =
        rmp_serde::from_slice(&payload).expect("failed to deserialize");

    assert_eq!(snapshot.tick, 42);
    assert_eq!(snapshot.entities.len(), 1);
    assert_eq!(snapshot.entities[0].entity_id, 100);
    assert_eq!(snapshot.entities[0].x, 10.5);
    assert_eq!(snapshot.entities[0].y, 20.3);

    server_handle.join().expect("server thread panicked");
}

#[test]
fn input_roundtrip_over_tcp() {
    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
    let server_addr = listener.local_addr().expect("failed to get local address");

    let server_handle = thread::spawn(move || {
        let bridge = TcpBridge::accept_on(listener).expect("failed to accept");

        // Non-blocking socket: retry until data arrives or timeout.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        let inputs = loop {
            match bridge.receive() {
                Ok(Some(Inbound::Inputs(inputs))) if !inputs.is_empty() => break inputs,
                Ok(_) => {
                    assert!(
                        std::time::Instant::now() < deadline,
                        "timed out waiting for inputs"
                    );
                    thread::sleep(std::time::Duration::from_millis(1));
                }
                Err(e) => panic!("failed to receive inputs: {}", e),
            }
        };

        assert_eq!(inputs.len(), 2);
        assert_eq!(inputs[0].tick, 10);
        assert_eq!(inputs[1].tick, 11);

        inputs
    });

    // Client: connect and send inputs
    let stream = TcpStream::connect(server_addr).expect("failed to connect");
    let mut writer = std::io::BufWriter::new(stream);

    let inputs = vec![
        PlayerInput {
            tick: 10,
            action: PlayerAction::MoveNorth,
        },
        PlayerInput {
            tick: 11,
            action: PlayerAction::Interact {
                target_entity_id: None,
                verb: None,
            },
        },
    ];

    let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize");
    write_framed(&mut writer, &payload).expect("failed to write frame");

    // Drop writer to close connection and signal EOF to server
    drop(writer);

    let received_inputs = server_handle.join().expect("server thread panicked");

    match &received_inputs[0].action {
        PlayerAction::MoveNorth => {}
        _ => panic!("expected MoveNorth action"),
    }
    match &received_inputs[1].action {
        PlayerAction::Interact { .. } => {}
        _ => panic!("expected Interact action"),
    }
}

#[test]
fn tcp_bridge_eof_returns_error() {
    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
    let server_addr = listener.local_addr().expect("failed to get local address");

    let server_handle = thread::spawn(move || {
        let bridge = TcpBridge::accept_on(listener).expect("failed to accept");

        // Non-blocking socket: retry until we get Disconnected or timeout.
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        loop {
            match bridge.receive() {
                Ok(None) => {
                    // No data yet — client hasn't disconnected, retry
                    assert!(
                        std::time::Instant::now() < deadline,
                        "timed out waiting for EOF"
                    );
                    thread::sleep(std::time::Duration::from_millis(1));
                }
                Ok(Some(_)) => panic!("expected Disconnected error, got a message"),
                Err(settled_reach_server::bridge::BridgeError::Disconnected) => break,
                Err(e) => panic!("expected Disconnected error, got: {}", e),
            }
        }
    });

    // Client: connect and immediately disconnect without sending data
    let stream = TcpStream::connect(server_addr).expect("failed to connect");
    drop(stream);

    server_handle.join().expect("server thread panicked");
}

/// T-1045 regression: EOF *mid-frame* (peer dies after a partial write) must
/// escalate to `Disconnected` like clean EOF — not surface an Io error every
/// tick forever against a frame that can never complete.
#[test]
fn tcp_bridge_eof_mid_frame_escalates_to_disconnected() {
    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
    let server_addr = listener.local_addr().expect("failed to get local address");

    let server_handle = thread::spawn(move || {
        let bridge = TcpBridge::accept_on(listener).expect("failed to accept");

        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        loop {
            match bridge.receive() {
                Ok(None) => {
                    assert!(
                        std::time::Instant::now() < deadline,
                        "timed out waiting for mid-frame EOF"
                    );
                    thread::sleep(std::time::Duration::from_millis(1));
                }
                Ok(Some(_)) => panic!("expected Disconnected, got a message"),
                Err(settled_reach_server::bridge::BridgeError::Disconnected) => break,
                Err(e) => panic!("expected Disconnected, got: {}", e),
            }
        }
    });

    // Client: write 2 of the 4 length-prefix bytes, then die.
    use std::io::Write;
    let mut stream = TcpStream::connect(server_addr).expect("failed to connect");
    stream
        .write_all(&[0x00, 0x00])
        .expect("partial write failed");
    stream.flush().expect("flush failed");
    thread::sleep(std::time::Duration::from_millis(50));
    drop(stream);

    server_handle.join().expect("server thread panicked");
}

/// Build a raw frame (4-byte BE length prefix + payload) for split-write tests.
fn raw_frame(payload: &[u8]) -> Vec<u8> {
    let mut frame = (payload.len() as u32).to_be_bytes().to_vec();
    frame.extend_from_slice(payload);
    frame
}

/// Poll the non-blocking bridge until `count` input batches arrive.
fn collect_input_batches(bridge: &TcpBridge, count: usize) -> Vec<Vec<PlayerInput>> {
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    let mut batches = Vec::new();
    while batches.len() < count {
        match bridge.receive() {
            Ok(Some(Inbound::Inputs(inputs))) if !inputs.is_empty() => batches.push(inputs),
            Ok(_) => {
                assert!(
                    std::time::Instant::now() < deadline,
                    "timed out waiting for {} input batches (got {})",
                    count,
                    batches.len()
                );
                thread::sleep(std::time::Duration::from_millis(1));
            }
            Err(e) => panic!("failed to receive inputs: {}", e),
        }
    }
    batches
}

/// T-1045 regression: a frame delivered in two TCP writes (split at
/// `split_point(frame_len)`) must decode, and the NEXT frame must also decode
/// — i.e. a partial read mid-frame must not desync the stream. The server
/// polls receive() during the gap, so it observes WouldBlock mid-frame.
fn assert_split_frame_does_not_desync(split_point: impl Fn(usize) -> usize) {
    use std::io::Write;

    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
    let server_addr = listener.local_addr().expect("failed to get local address");

    let server_handle = thread::spawn(move || {
        let bridge = TcpBridge::accept_on(listener).expect("failed to accept");
        collect_input_batches(&bridge, 2)
    });

    let mut stream = TcpStream::connect(server_addr).expect("failed to connect");

    let inputs1 = vec![PlayerInput {
        tick: 10,
        action: PlayerAction::MoveNorth,
    }];
    let payload1 = rmp_serde::to_vec_named(&inputs1).expect("failed to serialize");
    let frame1 = raw_frame(&payload1);
    let split = split_point(frame1.len());
    assert!(split > 0 && split < frame1.len(), "split must be mid-frame");

    // First half, then a gap long enough for the server to poll mid-frame.
    stream
        .write_all(&frame1[..split])
        .expect("write first half");
    stream.flush().expect("flush first half");
    thread::sleep(std::time::Duration::from_millis(50));
    stream
        .write_all(&frame1[split..])
        .expect("write second half");
    stream.flush().expect("flush second half");

    // A second frame in a single write — decodes only if the stream is in sync.
    let inputs2 = vec![PlayerInput {
        tick: 11,
        action: PlayerAction::MoveSouth,
    }];
    let payload2 = rmp_serde::to_vec_named(&inputs2).expect("failed to serialize");
    stream
        .write_all(&raw_frame(&payload2))
        .expect("write second frame");
    stream.flush().expect("flush second frame");

    let batches = server_handle.join().expect("server thread panicked");
    assert_eq!(batches[0].len(), 1);
    assert_eq!(batches[0][0].tick, 10);
    assert!(matches!(batches[0][0].action, PlayerAction::MoveNorth));
    assert_eq!(batches[1].len(), 1);
    assert_eq!(batches[1][0].tick, 11);
    assert!(matches!(batches[1][0].action, PlayerAction::MoveSouth));
}

#[test]
fn frame_split_inside_prefix_does_not_desync() {
    // Split inside the 4-byte length prefix.
    assert_split_frame_does_not_desync(|_| 2);
}

#[test]
fn frame_split_inside_payload_does_not_desync() {
    // Split midway through the payload (prefix is 4 bytes).
    assert_split_frame_does_not_desync(|frame_len| 4 + (frame_len - 4) / 2);
}

/// T-1045 drain: one receive_bridge_inputs run (= one 50 ms tick) must drain
/// every complete frame buffered on the socket, not one frame per tick.
#[test]
fn single_tick_drains_all_ready_inbound_frames() {
    use bevy_ecs::system::RunSystemOnce;
    use settled_reach_server::atlas::cascade::CascadeLayer;
    use settled_reach_server::atlas::layer_proxy::AtlasLayerRequest;
    use settled_reach_server::bridge::{
        receive_bridge_inputs, AtlasRequestBuffer, BridgeResource, BrowseRequestBuffer,
        CityNamesRequestBuffer, FeatureNamesRequestBuffer, HandshakeState, ServerRunning,
        StarMapRequestBuffer, StepCanvasRequestBuffer,
    };
    use settled_reach_server::simulation::input::InputQueue;

    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
    let server_addr = listener.local_addr().expect("failed to get local address");

    // Client: six frames back-to-back in one tick window — two input
    // batches plus one of EACH request shape (atlas, star-map, city-names,
    // feature-names: the full D-225/T-949/T-1169 demux surface over the real
    // framing/poll path — PR #176 review H6). Returns the stream so it stays
    // open until assertions complete (no EOF race).
    let client_handle = thread::spawn(move || {
        use settled_reach_server::atlas::atlas_data_proxy::{
            CityNamesRequest, FeatureNamesRequest, StarMapRequest,
        };

        let mut stream = TcpStream::connect(server_addr).expect("failed to connect");
        for tick in [20u64, 21] {
            let inputs = vec![PlayerInput {
                tick,
                action: PlayerAction::MoveNorth,
            }];
            let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize");
            write_framed(&mut stream, &payload).expect("write input frame");
        }
        let req = AtlasLayerRequest {
            body_id: "GJ1c".into(),
            up_to: CascadeLayer::Topography,
            window_center: None,
            window_n: 0,
            window_granularity_v2: None,
            window_min_wl_m: 0,
        };
        let payload = rmp_serde::to_vec_named(&req).expect("failed to serialize");
        write_framed(&mut stream, &payload).expect("write atlas frame");
        let sm = StarMapRequest { star_map: true };
        let payload = rmp_serde::to_vec_named(&sm).expect("failed to serialize star map");
        write_framed(&mut stream, &payload).expect("write star map frame");
        let cn = CityNamesRequest {
            city_names: true,
            body_id: "GJ1c".into(),
        };
        let payload = rmp_serde::to_vec_named(&cn).expect("failed to serialize city names");
        write_framed(&mut stream, &payload).expect("write city names frame");
        let fn_req = FeatureNamesRequest {
            feature_names: true,
            body_id: "GJ1c".into(),
        };
        let payload = rmp_serde::to_vec_named(&fn_req).expect("failed to serialize feature names");
        write_framed(&mut stream, &payload).expect("write feature names frame");
        stream
    });

    let bridge = TcpBridge::accept_on(listener).expect("failed to accept");
    let _stream = client_handle.join().expect("client thread panicked");
    // Writes are flushed and joined; small grace period for loopback delivery.
    thread::sleep(std::time::Duration::from_millis(100));

    let mut world = bevy_ecs::world::World::new();
    world.insert_resource(BridgeResource::new(bridge));
    world.init_resource::<InputQueue>();
    world.init_resource::<ServerRunning>();
    world.insert_resource(HandshakeState::Complete);
    world.init_resource::<SimErrorBuffer>();
    world.init_resource::<AtlasRequestBuffer>();
    world.init_resource::<StarMapRequestBuffer>();
    world.init_resource::<CityNamesRequestBuffer>();
    world.init_resource::<BrowseRequestBuffer>();
    world.init_resource::<StepCanvasRequestBuffer>();
    world.init_resource::<FeatureNamesRequestBuffer>();

    world
        .run_system_once(receive_bridge_inputs)
        .expect("receive_bridge_inputs failed to run");

    assert_eq!(
        world.resource::<InputQueue>().len(),
        2,
        "both input batches must drain in a single tick"
    );
    assert_eq!(
        world.resource::<AtlasRequestBuffer>().0.len(),
        1,
        "the atlas request must drain in the same tick"
    );
    assert_eq!(
        world.resource::<StarMapRequestBuffer>().0.len(),
        1,
        "the star-map request must drain in the same tick (H6: real wire path)"
    );
    let city_names = &world.resource::<CityNamesRequestBuffer>().0;
    assert_eq!(
        city_names.len(),
        1,
        "the city-names request must drain in the same tick (H6: real wire path)"
    );
    assert_eq!(city_names[0].1.body_id, "GJ1c");
    let feature_names = &world.resource::<FeatureNamesRequestBuffer>().0;
    assert_eq!(
        feature_names.len(),
        1,
        "the feature-names request must drain in the same tick (T-1169: real wire path)"
    );
    assert_eq!(feature_names[0].1.body_id, "GJ1c");
    assert!(
        world.resource::<ServerRunning>().0,
        "draining must not shut the server down"
    );
}

// -- D-254 §2 / T-1130: multi-connection bridge + ConnectionRole gate ---------
//
// Six behaviors below (a 7th — old-format StartupMessage decodes as Player —
// is a unit test in server/src/bridge/types.rs, next to the type itself).
// Each test drives the real wire path: a genuine TcpStream client performs
// the handshake/startup exchange a real Godot client (or PendingConnection's
// server-side counterpart) would, against the real non-blocking accept-loop
// and drain-loop systems via `run_system_once` — no mocks of the framing or
// role-gate logic itself.

/// Client-side test helper: perform one full handshake/startup exchange over
/// an already-connected stream, exactly as `sim_bridge.gd`'s live-mode path
/// does (read HandshakeMessage, write StartupMessage). Returns the stream so
/// the caller can continue driving it (send inputs, read responses, etc.).
fn client_handshake(mut stream: TcpStream, role: ConnectionRole) -> TcpStream {
    let handshake_payload = read_framed(&mut stream)
        .expect("failed to read handshake")
        .expect("unexpected EOF reading handshake");
    let _: HandshakeMessage =
        rmp_serde::from_slice(&handshake_payload).expect("failed to decode HandshakeMessage");

    let startup = StartupMessage {
        world_seed: 12345,
        role,
    };
    let payload = rmp_serde::to_vec_named(&startup).expect("failed to serialize StartupMessage");
    write_framed(&mut stream, &payload).expect("failed to write StartupMessage");
    stream
}

/// Drive `accept_new_connections` for up to `max_ticks` schedule passes, or
/// until `bridge.readers().len() + bridge.has_player() as usize` (checked via
/// the passed predicate) is satisfied — matches this test file's existing
/// wall-clock-deadline-over-fixed-sleep hardening (see `collect_input_batches`
/// above): connection promotion depends on TCP delivery timing, not a fixed
/// tick count, so poll until true or time out rather than guessing a sleep.
fn drive_accept_loop_until(
    world: &mut bevy_ecs::world::World,
    condition: impl Fn(&bevy_ecs::world::World) -> bool,
) {
    use bevy_ecs::system::RunSystemOnce;
    use settled_reach_server::bridge::accept_new_connections;

    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    while !condition(world) {
        assert!(
            std::time::Instant::now() < deadline,
            "timed out waiting for accept-loop condition"
        );
        world
            .run_system_once(accept_new_connections)
            .expect("accept_new_connections failed to run");
        thread::sleep(std::time::Duration::from_millis(1));
    }
}

/// Build a `bevy_ecs::World` wired exactly like `BridgePlugin` wires it for
/// the systems under test here (accept-loop + drain-loop + response
/// buffers), bound to a fresh OS-assigned port.
fn new_multi_connection_world() -> (bevy_ecs::world::World, std::net::SocketAddr) {
    use settled_reach_server::bridge::{
        AtlasRequestBuffer, AtlasResponseBuffer, BridgeResource, BrowseRequestBuffer,
        BrowseResponseBuffer, CityNamesRequestBuffer, CityNamesResponseBuffer, ConnectionListener,
        FeatureNamesRequestBuffer, FeatureNamesResponseBuffer, HandshakeState, PendingConnections,
        ServerRunning, SnapshotBuffer, StarMapRequestBuffer, StarMapResponseBuffer,
        StepCanvasRequestBuffer, StepCanvasResponseBuffer,
    };
    use settled_reach_server::simulation::input::InputQueue;

    let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
    let addr = listener.local_addr().expect("failed to get local address");
    listener
        .set_nonblocking(true)
        .expect("failed to set listener non-blocking");

    let mut world = bevy_ecs::world::World::new();
    world.insert_resource(ConnectionListener(Some(listener)));
    world.init_resource::<PendingConnections>();
    world.insert_resource(BridgeResource::default());
    world.init_resource::<InputQueue>();
    world.init_resource::<ServerRunning>();
    world.insert_resource(HandshakeState::Complete);
    world.init_resource::<SimErrorBuffer>();
    world.init_resource::<AtlasRequestBuffer>();
    world.init_resource::<AtlasResponseBuffer>();
    world.init_resource::<StarMapRequestBuffer>();
    world.init_resource::<StarMapResponseBuffer>();
    world.init_resource::<CityNamesRequestBuffer>();
    world.init_resource::<CityNamesResponseBuffer>();
    world.init_resource::<BrowseRequestBuffer>();
    world.init_resource::<BrowseResponseBuffer>();
    world.init_resource::<StepCanvasRequestBuffer>();
    world.init_resource::<StepCanvasResponseBuffer>();
    world.init_resource::<FeatureNamesRequestBuffer>();
    world.init_resource::<FeatureNamesResponseBuffer>();
    world.init_resource::<SnapshotBuffer>();
    (world, addr)
}

/// T-1130 behavior 2: a Reader connection's handshake succeeds — it reaches
/// `BridgeResource`'s reader collection — without any character-spawn
/// concept ever entering the picture. There is no `PlayerCharacter`-query
/// assertion here because that's the correct proof of the D-254 §2
/// guarantee: character-spawn is a `main.rs`/world-setup concern entirely
/// disjoint from connection acceptance (see `main.rs`'s D-254 §2 scope-note
/// comment at the `setup_proof_room`/`setup_gauntlet` call site) — a Reader
/// reaching `BridgeResource.readers` never touches that code path at all,
/// which this test demonstrates by never invoking it and the connection
/// still working end-to-end.
#[test]
fn reader_handshake_succeeds_and_installs_as_reader() {
    use settled_reach_server::bridge::BridgeResource;

    let (mut world, addr) = new_multi_connection_world();

    let client_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect");
        client_handshake(stream, ConnectionRole::Reader)
    });

    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });

    let _stream = client_handle.join().expect("client thread panicked");

    let bridge = world.resource::<BridgeResource>();
    assert!(
        !bridge.has_player(),
        "a Reader-only connection must never be installed as Player"
    );
    assert_eq!(bridge.reader_count(), 1, "exactly one reader installed");
}

/// T-1130 behavior 3: an atlas/star-map/city-names response addressed to a
/// Reader's `ConnectionId` reaches that Reader over the wire — the
/// per-connection response-tagging plumbing (D-254 §2) actually delivers,
/// not just tags in-memory. Drives `send_star_map_responses` directly
/// (bypassing `serve_star_map_requests`/the atlas proxy, which are already
/// covered elsewhere) to isolate exactly the tagging/routing behavior this
/// ticket adds.
#[test]
fn reader_receives_tagged_star_map_response() {
    use bevy_ecs::system::RunSystemOnce;
    use settled_reach_server::atlas::atlas_data_proxy::{StarMapResponse, StarMapStatus};
    use settled_reach_server::bridge::{
        send_star_map_responses, BridgeResource, StarMapResponseBuffer,
    };

    let (mut world, addr) = new_multi_connection_world();

    let client_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect");
        client_handshake(stream, ConnectionRole::Reader)
    });

    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let mut stream = client_handle.join().expect("client thread panicked");

    let reader_id = world
        .resource::<BridgeResource>()
        .reader_ids()
        .first()
        .copied()
        .expect("reader must be installed");

    world.resource_mut::<StarMapResponseBuffer>().0.push((
        reader_id,
        StarMapResponse {
            status: StarMapStatus::Ready,
            data: Some(serde_json::json!({"nodes": [], "edges": []})),
        },
    ));

    world
        .run_system_once(send_star_map_responses)
        .expect("send_star_map_responses failed to run");

    let payload = read_framed(&mut stream)
        .expect("failed to read response frame")
        .expect("unexpected EOF reading response");
    let resp: StarMapResponse = rmp_serde::from_slice(&payload).expect("failed to decode");
    assert_eq!(resp.status, StarMapStatus::Ready);
}

/// T-1130 behavior 4: while a Player streams `ObserverSnapshot` every tick,
/// a concurrently-connected Reader receives NOTHING on its socket — not a
/// filtered snapshot, not an empty one, nothing at all (D-254 §2: "not even
/// filtered" is structural, this test proves it holds over the wire with
/// both connections genuinely live at once, not just by code inspection).
#[test]
fn reader_never_receives_observer_snapshot_while_player_streams() {
    use bevy_ecs::system::RunSystemOnce;
    use settled_reach_server::bridge::{send_bridge_snapshot, BridgeResource, SnapshotBuffer};

    let (mut world, addr) = new_multi_connection_world();

    // Player connects first (mirrors main.rs's structural first-connection
    // path in spirit, though this test drives it through the SAME
    // accept-loop the Reader below uses — the accept-loop must handle a
    // Player exactly as well as a Reader, not just readers-after-a-
    // pre-existing-Player).
    let player_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect (player)");
        client_handshake(stream, ConnectionRole::Player)
    });
    drive_accept_loop_until(&mut world, |w| w.resource::<BridgeResource>().has_player());
    let mut player_stream = player_handle.join().expect("player thread panicked");

    let reader_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect (reader)");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let mut reader_stream = reader_handle.join().expect("reader thread panicked");
    // Non-blocking so a read that would otherwise hang forever (the whole
    // point being tested — nothing ever arrives) returns WouldBlock instead.
    reader_stream
        .set_nonblocking(true)
        .expect("failed to set reader stream non-blocking");

    world.resource_mut::<SnapshotBuffer>().snapshot = Some(sample_snapshot(7));
    world
        .run_system_once(send_bridge_snapshot)
        .expect("send_bridge_snapshot failed to run");

    // Player DID get the snapshot — establishes the positive control so a
    // trivially-broken send path (e.g. send_bridge_snapshot silently
    // no-op'ing for everyone) can't masquerade as "reader correctly got
    // nothing".
    let payload = read_framed(&mut player_stream)
        .expect("failed to read player frame")
        .expect("unexpected EOF reading player snapshot");
    let snapshot: ObserverSnapshot = rmp_serde::from_slice(&payload).expect("failed to decode");
    assert_eq!(snapshot.tick, 7);

    // Reader got NOTHING — not WouldBlock-then-eventually-something, a
    // sustained absence over a real wall-clock window.
    let check_until = std::time::Instant::now() + std::time::Duration::from_millis(300);
    while std::time::Instant::now() < check_until {
        let mut probe = [0u8; 1];
        match std::io::Read::read(&mut reader_stream, &mut probe) {
            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
            Ok(0) => panic!("reader stream unexpectedly closed"),
            other => panic!(
                "reader received unexpected data/result while player streamed: {:?}",
                other
            ),
        }
        thread::sleep(std::time::Duration::from_millis(10));
    }
}

/// T-1130 behavior 5: a Reader sending the forbidden `Vec<PlayerInput>`
/// shape is dropped (not forwarded to `InputQueue`) on each offense, and
/// disconnected once it crosses the strike threshold — never on the first
/// offense (D-254 §2: "log + drop on first offense, repeated = disconnect").
#[test]
fn forbidden_reader_input_is_dropped_then_disconnected() {
    use bevy_ecs::system::RunSystemOnce;
    use settled_reach_server::bridge::{receive_bridge_inputs, BridgeResource};
    use settled_reach_server::simulation::input::InputQueue;

    let (mut world, addr) = new_multi_connection_world();

    let client_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let mut stream = client_handle.join().expect("client thread panicked");

    let send_one_input_frame = |stream: &mut TcpStream| {
        let inputs = vec![PlayerInput {
            tick: 1,
            action: PlayerAction::MoveNorth,
        }];
        let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize");
        write_framed(stream, &payload).expect("failed to write input frame");
    };

    // Strikes 1 and 2: the reader survives, but nothing reaches InputQueue.
    for strike in 1..=2 {
        send_one_input_frame(&mut stream);
        // Wall-clock settle for TCP delivery, then drain — mirrors this
        // file's existing pattern (see single_tick_drains_all_ready_inbound_frames).
        thread::sleep(std::time::Duration::from_millis(50));
        world
            .run_system_once(receive_bridge_inputs)
            .expect("receive_bridge_inputs failed to run");
        assert_eq!(
            world.resource::<InputQueue>().len(),
            0,
            "forbidden reader input must never reach InputQueue (strike {})",
            strike
        );
        assert_eq!(
            world.resource::<BridgeResource>().reader_count(),
            1,
            "reader must survive strike {} (below disconnect threshold)",
            strike
        );
    }

    // Strike 3 crosses READER_VIOLATION_DISCONNECT_THRESHOLD (3) — the
    // reader is disconnected.
    send_one_input_frame(&mut stream);
    thread::sleep(std::time::Duration::from_millis(50));
    world
        .run_system_once(receive_bridge_inputs)
        .expect("receive_bridge_inputs failed to run");
    assert_eq!(
        world.resource::<InputQueue>().len(),
        0,
        "forbidden reader input must never reach InputQueue, even on the disconnecting strike"
    );
    assert_eq!(
        world.resource::<BridgeResource>().reader_count(),
        0,
        "reader must be disconnected after crossing the violation threshold"
    );
}

/// T-1130 behavior 6 (THE CRITICAL FIX): a Reader disconnecting must leave
/// a running Player session completely unaffected — `ServerRunning` stays
/// true, and the Player connection keeps working (proven by successfully
/// sending it a snapshot AFTER the reader is gone, not just by inspecting
/// the flag).
#[test]
fn reader_disconnect_does_not_affect_running_player_session() {
    use bevy_ecs::system::RunSystemOnce;
    use settled_reach_server::bridge::{
        receive_bridge_inputs, send_bridge_snapshot, BridgeResource, ServerRunning, SnapshotBuffer,
    };

    let (mut world, addr) = new_multi_connection_world();

    let player_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect (player)");
        client_handshake(stream, ConnectionRole::Player)
    });
    drive_accept_loop_until(&mut world, |w| w.resource::<BridgeResource>().has_player());
    let mut player_stream = player_handle.join().expect("player thread panicked");

    let reader_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect (reader)");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let reader_stream = reader_handle.join().expect("reader thread panicked");

    // The reader disconnects (drop closes the TCP connection — clean EOF).
    drop(reader_stream);

    // Drain: the server observes the reader's EOF and removes it.
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        world
            .run_system_once(receive_bridge_inputs)
            .expect("receive_bridge_inputs failed to run");
        if world.resource::<BridgeResource>().reader_count() == 0 {
            break;
        }
        assert!(
            std::time::Instant::now() < deadline,
            "timed out waiting for reader disconnect to be observed"
        );
        thread::sleep(std::time::Duration::from_millis(10));
    }

    assert!(
        world.resource::<ServerRunning>().0,
        "a reader's disconnect must never flip ServerRunning"
    );
    assert!(
        world.resource::<BridgeResource>().has_player(),
        "the player connection must still be installed after the reader disconnects"
    );

    // Prove the player session is still genuinely functional, not just
    // structurally present: send it a snapshot and read it back.
    world.resource_mut::<SnapshotBuffer>().snapshot = Some(sample_snapshot(99));
    world
        .run_system_once(send_bridge_snapshot)
        .expect("send_bridge_snapshot failed to run");
    let payload = read_framed(&mut player_stream)
        .expect("failed to read post-disconnect player frame")
        .expect("unexpected EOF — player session was affected by reader disconnect");
    let snapshot: ObserverSnapshot = rmp_serde::from_slice(&payload).expect("failed to decode");
    assert_eq!(
        snapshot.tick, 99,
        "player session must remain fully functional after reader disconnect"
    );
    assert!(
        world.resource::<ServerRunning>().0,
        "sending to the still-live player must not flip ServerRunning either"
    );
}

/// T-1130 behavior 7: a second connection attempting `role: Player` while a
/// Player is already connected gets a clean, immediate rejection (the
/// connection closes right after its handshake completes) — NOT the
/// original starvation bug (silent hang forever), and NOT a crash or
/// disruption to the existing Player's session.
#[test]
fn second_player_attempt_is_cleanly_rejected() {
    use bevy_ecs::system::RunSystemOnce;
    use settled_reach_server::bridge::BridgeResource;

    let (mut world, addr) = new_multi_connection_world();

    let first_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect (first player)");
        client_handshake(stream, ConnectionRole::Player)
    });
    drive_accept_loop_until(&mut world, |w| w.resource::<BridgeResource>().has_player());
    let _first_stream = first_handle.join().expect("first player thread panicked");

    let second_handle = thread::spawn(move || {
        let mut stream = TcpStream::connect(addr).expect("failed to connect (second player)");
        // Read handshake and send StartupMessage{role: Player} exactly like
        // a normal client — the rejection happens AFTER this, not by
        // refusing the handshake itself (the connection is genuinely
        // accepted and handshaken; it's the ROLE promotion that's refused).
        stream = client_handshake(stream, ConnectionRole::Player);
        // The server closes the connection right after — prove it's a
        // clean disconnect (not the original silent-hang bug) by reading
        // until EOF. read_framed blocks here (this stream is never set
        // non-blocking), which is deliberate: it's this thread's own proof
        // that the disconnect actually happens — if the original starvation
        // bug were still present, this call would hang forever with no
        // internal deadline of its own. The OUTER test loop below bounds
        // total test time via `second_handle.is_finished()` polling against
        // ITS OWN 5s deadline, so a regression here still fails the test in
        // bounded time rather than hanging the test suite.
        match read_framed(&mut stream) {
            Ok(None) => {} // clean EOF — the expected rejection signal
            Ok(Some(_)) => panic!("second player attempt must never receive a message"),
            Err(_) => {} // connection reset also counts as "rejected, not hung"
        }
    });

    // Drive the accept-loop until the second connection has been processed
    // (it will never appear as a reader OR a second player — reader_count
    // stays 0 and has_player stays true throughout, which is exactly the
    // rejection this test verifies).
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    while !second_handle.is_finished() {
        assert!(
            std::time::Instant::now() < deadline,
            "timed out waiting for second player rejection to complete"
        );
        world
            .run_system_once(settled_reach_server::bridge::accept_new_connections)
            .expect("accept_new_connections failed to run");
        thread::sleep(std::time::Duration::from_millis(5));
    }
    second_handle.join().expect("second player thread panicked");

    assert!(
        world.resource::<BridgeResource>().has_player(),
        "the original player connection must be untouched by the rejected second attempt"
    );
    assert_eq!(
        world.resource::<BridgeResource>().reader_count(),
        0,
        "a rejected second-Player attempt must never be silently installed as a reader"
    );
}

// ─────────────────────────────────────────────────────────────────────────
// T-1131: data browser (D-254 §4) end-to-end over the real TCP wire.
//
// These tests exercise the FULL pipeline — demux (decode_inbound) ->
// receive_bridge_inputs (drain + connection-tag) -> BrowseRequestBuffer ->
// serve_browse_requests (the atlas-plugin proxy system) -> BrowseResponseBuffer
// -> send_browse_responses -> the socket — not just the buffer-push shortcut
// `reader_receives_tagged_star_map_response` uses above, since T-1131's own
// ticket text calls for the connection-tagging proof specifically over
// requests that actually round-trip through the demux.
// ─────────────────────────────────────────────────────────────────────────

/// Build a fixture `systems.db`-shaped file with one row in each of the six
/// v1 browse tables, wired as a `BrowseReaderResource` into `world`. A
/// self-contained minimal fixture local to THIS file — `browse_reader`'s own
/// (larger, column-exhaustive) fixture lives behind `#[cfg(test)]` in the
/// library crate, which is only compiled for `cargo test --lib`, not for a
/// separate integration-test binary linking against the built library (a
/// cross-crate `#[cfg(test)]` visibility boundary — confirmed by attempting
/// the reuse first). This file only needs to prove the WIRE plumbing (demux
/// -> serve -> send), not the SQL correctness `browse_reader`'s own unit
/// tests already verify column-by-column, so a minimal one-row-per-table
/// fixture is the right scope here, not a duplicate of the exhaustive one.
fn wire_browse_reader(world: &mut bevy_ecs::world::World) {
    use settled_reach_server::atlas::browse_reader::{BrowseReader, BrowseReaderResource};

    let db_path = std::env::temp_dir().join(format!(
        "sr_browse_tcp_fixture_{}_{}.db",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    let _ = std::fs::remove_file(&db_path);
    {
        let conn = rusqlite::Connection::open(&db_path).expect("create fixture db");
        conn.execute_batch(
            "CREATE TABLE star_systems (
                 system_id TEXT PRIMARY KEY, proper_name TEXT, system_name TEXT,
                 star_type TEXT, spectral_class TEXT, dist_ly REAL,
                 geographic_sector TEXT, geographic_band TEXT, political_zone TEXT,
                 habitable_planet_count INTEGER, inhabited_planet_count INTEGER,
                 asteroid_belt INTEGER, gas_giant INTEGER, habitability_profile TEXT,
                 earth_alignment TEXT, earth_proximity TEXT, earth_tension TEXT,
                 stability_index INTEGER, system_volatility TEXT, cultural_corridor TEXT,
                 currency_zone TEXT
             );
             CREATE TABLE system_economy (
                 system_id TEXT PRIMARY KEY, economic_tier INTEGER, population INTEGER,
                 economic_base_primary TEXT, economic_base_secondary TEXT
             );
             CREATE TABLE system_factions (
                 system_id TEXT PRIMARY KEY, governance_type TEXT, dominant_faction TEXT
             );
             CREATE TABLE system_culture (
                 system_id TEXT PRIMARY KEY, cultural_register TEXT,
                 atmospheric_tone TEXT, primary_archetype TEXT
             );
             CREATE TABLE bodies (
                 body_id TEXT PRIMARY KEY, system_id TEXT NOT NULL, parent_body_id TEXT,
                 body_type TEXT NOT NULL, orbit_index INTEGER, proper_name TEXT,
                 mass_class TEXT, atmosphere TEXT, surface_gravity REAL,
                 orbital_period_days REAL, rotation_period_hours REAL, planet_class TEXT,
                 hydrosphere TEXT, biosphere_class TEXT, inhabited INTEGER NOT NULL DEFAULT 0,
                 population INTEGER, economic_role TEXT, founding_age_years INTEGER,
                 settlement_pattern TEXT, cultural_corridor TEXT, industrial_corridor TEXT,
                 body_radius_km REAL, axial_tilt_deg REAL
             );
             CREATE TABLE stations (
                 station_id TEXT PRIMARY KEY, system_id TEXT NOT NULL, orbits_body_id TEXT,
                 station_type TEXT NOT NULL, proper_name TEXT, population INTEGER,
                 economic_role TEXT, governance_type TEXT, docking_class TEXT,
                 has_gate_infrastructure INTEGER DEFAULT 0, district_count INTEGER
             );
             CREATE TABLE corporations (
                 corp_id TEXT PRIMARY KEY, proper_name TEXT NOT NULL, corp_type TEXT NOT NULL,
                 scope TEXT, headquarters_system TEXT, headquarters_body TEXT,
                 specialization TEXT, parent_corp TEXT, notes TEXT, behavioral_archetype TEXT,
                 supply_chain_role TEXT, shadow_economy_access INTEGER DEFAULT 0,
                 corp_specialization TEXT, hq_placement TEXT
             );
             CREATE TABLE corp_presence (
                 corp_id TEXT NOT NULL, location_id TEXT NOT NULL, location_type TEXT NOT NULL,
                 primary_operation TEXT, PRIMARY KEY (corp_id, location_id)
             );
             CREATE TABLE corp_financial_state (corp_id TEXT PRIMARY KEY, health_metric REAL NOT NULL DEFAULT 1.0);
             CREATE TABLE commodities (
                 commodity_id TEXT PRIMARY KEY, name TEXT NOT NULL, tier TEXT NOT NULL,
                 elasticity TEXT NOT NULL, base_price REAL NOT NULL, bulk_class TEXT,
                 unit TEXT, production_ubiquity TEXT, demand_model TEXT,
                 commission_certifiable INTEGER DEFAULT 0, compact_contested INTEGER DEFAULT 0,
                 shadow_viable INTEGER DEFAULT 0, panic_threshold_weeks INTEGER DEFAULT 0, description TEXT
             );
             CREATE TABLE production_chains (
                 chain_id TEXT PRIMARY KEY, output_commodity_id TEXT NOT NULL,
                 output_quantity REAL NOT NULL DEFAULT 1.0, location_bound INTEGER DEFAULT 0, description TEXT
             );
             CREATE TABLE chain_inputs (
                 chain_id TEXT NOT NULL, input_commodity_id TEXT NOT NULL, quantity REAL NOT NULL,
                 PRIMARY KEY (chain_id, input_commodity_id)
             );
             CREATE TABLE trait_templates (
                 tag TEXT PRIMARY KEY, label TEXT NOT NULL, cultural_description TEXT,
                 corridor_pool TEXT NOT NULL DEFAULT 'baseline', geographic_sector TEXT,
                 bulk_class_gate TEXT, production_ubiquity_gate TEXT,
                 min_prosperity_bps INTEGER NOT NULL DEFAULT 0, base_weight INTEGER NOT NULL DEFAULT 10000,
                 weight_mods TEXT, zone_affinity TEXT, allow_tags TEXT, block_tags TEXT,
                 era_scope TEXT, visual_bundle TEXT
             );",
        )
        .expect("create fixture tables");

        conn.execute(
            "INSERT INTO star_systems (system_id, proper_name, star_type) VALUES ('GJ-1', 'Aldren', 'M')",
            [],
        )
        .expect("insert star system");
        conn.execute(
            "INSERT INTO bodies (body_id, system_id, body_type, proper_name, inhabited)
             VALUES ('GJ1c', 'GJ-1', 'planet', 'Aldren Prime', 1)",
            [],
        )
        .expect("insert body");
        conn.execute(
            "INSERT INTO stations (station_id, system_id, station_type, proper_name)
             VALUES ('GJ1c-S1', 'GJ-1', 'commercial', 'Aldren Orbital')",
            [],
        )
        .expect("insert station");
        conn.execute(
            "INSERT INTO corporations (corp_id, proper_name, corp_type, headquarters_system)
             VALUES ('gate-corporation', 'Gate Corporation', 'corporation', 'GJ-1')",
            [],
        )
        .expect("insert corp");
        conn.execute(
            "INSERT INTO commodities (commodity_id, name, tier, elasticity, base_price)
             VALUES ('fusion_fuel', 'Fusion Fuel', 'intermediate', 'inelastic', 42.5)",
            [],
        )
        .expect("insert commodity");
        conn.execute(
            "INSERT INTO trait_templates (tag, label) VALUES ('frontier_utilitarian', 'Frontier Utilitarian')",
            [],
        )
        .expect("insert trait template");
    }

    let reader = BrowseReader::open(&db_path).expect("open fixture browse db");
    world.insert_resource(BrowseReaderResource(reader));
}

/// Drive one full server tick's worth of browse plumbing: `receive_bridge_inputs`
/// (drains the socket into `BrowseRequestBuffer`, connection-tagged),
/// `serve_browse_requests` (the atlas-plugin proxy — reads `BrowseReaderResource`,
/// fills `BrowseResponseBuffer`), and `send_browse_responses` (flushes back
/// out to the originating connection). Matches how `BridgePlugin` +
/// `GenerationPlugin` actually schedule these three systems in `PreInput`/
/// `PostSnapshot`, just run directly rather than through a full `App`.
fn drive_browse_tick(world: &mut bevy_ecs::world::World) {
    use bevy_ecs::system::RunSystemOnce;
    use settled_reach_server::atlas::plugin::serve_browse_requests;
    use settled_reach_server::bridge::{receive_bridge_inputs, send_browse_responses};

    world
        .run_system_once(receive_bridge_inputs)
        .expect("receive_bridge_inputs failed to run");
    world
        .run_system_once(serve_browse_requests)
        .expect("serve_browse_requests failed to run");
    world
        .run_system_once(send_browse_responses)
        .expect("send_browse_responses failed to run");
}

/// Drive `drive_browse_tick` repeatedly until a framed `BrowseResponse`
/// arrives on `stream`, or a wall-clock deadline expires (this file's
/// established hardening pattern — see `drive_accept_loop_until`/
/// `collect_input_batches` — rather than a single tick or a fixed sleep).
///
/// A single `drive_browse_tick` call races the client thread's write
/// actually landing in the server's non-blocking socket buffer before
/// `receive_bridge_inputs` polls it — `stream` here is a genuinely blocking
/// client-side socket (only the SERVER's accepted connections are toggled
/// non-blocking, by `TcpBridge::from_connected_stream`/D-254 §2's
/// accept-loop design), so `set_read_timeout` + a bounded retry loop is the
/// correct fix, not a longer single wait.
fn drive_browse_tick_until_response(
    world: &mut bevy_ecs::world::World,
    stream: &mut TcpStream,
) -> BrowseResponse {
    stream
        .set_read_timeout(Some(std::time::Duration::from_millis(50)))
        .expect("failed to set read timeout");
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        drive_browse_tick(world);
        match read_framed(stream) {
            Ok(Some(payload)) => {
                return rmp_serde::from_slice(&payload).expect("failed to decode BrowseResponse");
            }
            Ok(None) => panic!("unexpected EOF reading browse response"),
            Err(e)
                if e.kind() == std::io::ErrorKind::WouldBlock
                    || e.kind() == std::io::ErrorKind::TimedOut =>
            {
                assert!(
                    std::time::Instant::now() < deadline,
                    "timed out waiting for a browse response"
                );
            }
            Err(e) => panic!("failed to read browse response frame: {e}"),
        }
    }
}

fn send_browse_index_request(stream: &mut TcpStream, kind: BrowseEntityKind) {
    let req = BrowseRequest {
        browse: true,
        kind,
        query: BrowseQuery::Index {
            filter_system_id: None,
        },
    };
    let payload = rmp_serde::to_vec_named(&req).expect("failed to encode BrowseRequest");
    write_framed(stream, &payload).expect("failed to write BrowseRequest");
}

fn send_browse_detail_request(stream: &mut TcpStream, kind: BrowseEntityKind, id: &str) {
    let req = BrowseRequest {
        browse: true,
        kind,
        query: BrowseQuery::Detail { id: id.to_string() },
    };
    let payload = rmp_serde::to_vec_named(&req).expect("failed to encode BrowseRequest");
    write_framed(stream, &payload).expect("failed to write BrowseRequest");
}

/// T-1131: index + detail round-trip for all six D-254 §4 v1 entity kinds,
/// over one Reader connection, through the full demux->serve->send pipeline.
/// Parameterized-style — one test iterating all six kinds, per the ticket's
/// own suggested test shape.
#[test]
fn browse_index_and_detail_round_trip_for_all_six_kinds_over_tcp() {
    let (mut world, addr) = new_multi_connection_world();
    wire_browse_reader(&mut world);

    let client_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let mut stream = client_handle.join().expect("client thread panicked");

    let cases: &[(BrowseEntityKind, &str)] = &[
        (BrowseEntityKind::StarSystem, "GJ-1"),
        (BrowseEntityKind::Body, "GJ1c"),
        (BrowseEntityKind::Station, "GJ1c-S1"),
        (BrowseEntityKind::Corporation, "gate-corporation"),
        (BrowseEntityKind::Commodity, "fusion_fuel"),
        (BrowseEntityKind::TraitTemplate, "frontier_utilitarian"),
    ];

    for &(kind, id) in cases {
        send_browse_index_request(&mut stream, kind);
        let index_resp = drive_browse_tick_until_response(&mut world, &mut stream);
        assert_eq!(index_resp.kind, kind, "index response kind echo");
        assert_eq!(index_resp.status, BrowseStatus::Ready, "index({kind:?})");
        let rows = index_resp.index.expect("index populated");
        assert!(
            rows.iter().any(|r| r.id == id),
            "index({kind:?}) should list {id}"
        );

        send_browse_detail_request(&mut stream, kind, id);
        let detail_resp = drive_browse_tick_until_response(&mut world, &mut stream);
        assert_eq!(detail_resp.kind, kind, "detail response kind echo");
        assert_eq!(detail_resp.status, BrowseStatus::Ready, "detail({kind:?})");
        assert!(
            detail_resp.detail.is_some(),
            "detail({kind:?}) should be populated"
        );
    }
}

/// T-1131: a Reader (not just a Player) can send `BrowseRequest` and get a
/// `Ready` response — proves the permitted-message-matrix row (D-254 §2:
/// atlas/star-map/city-names/browse all say "yes" for Reader) actually holds
/// for the new request type specifically, not just by code inspection.
#[test]
fn reader_can_browse() {
    let (mut world, addr) = new_multi_connection_world();
    wire_browse_reader(&mut world);

    let client_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let mut stream = client_handle.join().expect("client thread panicked");

    send_browse_index_request(&mut stream, BrowseEntityKind::Commodity);
    let resp = drive_browse_tick_until_response(&mut world, &mut stream);
    assert_eq!(resp.status, BrowseStatus::Ready);
}

/// T-1131: a Player (not just a Reader) can also browse — the permitted-
/// message-matrix row says "yes" for both roles, and this is the other half
/// of that proof.
#[test]
fn player_can_browse() {
    let (mut world, addr) = new_multi_connection_world();
    wire_browse_reader(&mut world);

    let client_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect");
        client_handshake(stream, ConnectionRole::Player)
    });
    drive_accept_loop_until(&mut world, |w| w.resource::<BridgeResource>().has_player());
    let mut stream = client_handle.join().expect("client thread panicked");

    send_browse_index_request(&mut stream, BrowseEntityKind::TraitTemplate);
    let resp = drive_browse_tick_until_response(&mut world, &mut stream);
    assert_eq!(resp.status, BrowseStatus::Ready);
}

/// T-1131: two concurrently-connected Readers each get back only their OWN
/// browse response — the connection-tagging plumbing D-254 §2 established
/// for atlas/star-map/city-names must hold for browse too, proven with two
/// simultaneously-live sockets asking for DIFFERENT things so a crossed wire
/// would be immediately visible (not just "a response arrived", but "the
/// WRONG response arrived").
#[test]
fn two_readers_do_not_cross_browse_responses() {
    let (mut world, addr) = new_multi_connection_world();
    wire_browse_reader(&mut world);

    let reader_a_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect (reader A)");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let mut stream_a = reader_a_handle.join().expect("reader A thread panicked");

    let reader_b_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect (reader B)");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 2
    });
    let mut stream_b = reader_b_handle.join().expect("reader B thread panicked");

    // A asks about star systems, B asks about commodities — deliberately
    // different kinds so a crossed response is unmistakable (not just "wrong
    // data", but "wrong KIND"). Both requests are in flight before any
    // response is expected, so `drive_browse_tick` is retried until BOTH
    // streams have something readable (rather than draining/reading one
    // stream to completion before the other's request has even arrived).
    send_browse_index_request(&mut stream_a, BrowseEntityKind::StarSystem);
    send_browse_index_request(&mut stream_b, BrowseEntityKind::Commodity);

    stream_a
        .set_read_timeout(Some(std::time::Duration::from_millis(50)))
        .expect("failed to set read timeout (A)");
    stream_b
        .set_read_timeout(Some(std::time::Duration::from_millis(50)))
        .expect("failed to set read timeout (B)");
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    let mut resp_a: Option<BrowseResponse> = None;
    let mut resp_b: Option<BrowseResponse> = None;
    while resp_a.is_none() || resp_b.is_none() {
        drive_browse_tick(&mut world);
        if resp_a.is_none() {
            if let Ok(Some(payload)) = read_framed(&mut stream_a) {
                resp_a = Some(rmp_serde::from_slice(&payload).expect("decode A"));
            }
        }
        if resp_b.is_none() {
            if let Ok(Some(payload)) = read_framed(&mut stream_b) {
                resp_b = Some(rmp_serde::from_slice(&payload).expect("decode B"));
            }
        }
        assert!(
            std::time::Instant::now() < deadline,
            "timed out waiting for both readers' browse responses (A={}, B={})",
            resp_a.is_some(),
            resp_b.is_some()
        );
    }
    let resp_a = resp_a.expect("resp_a set by loop exit condition");
    let resp_b = resp_b.expect("resp_b set by loop exit condition");

    assert_eq!(
        resp_a.kind,
        BrowseEntityKind::StarSystem,
        "reader A must get back ITS OWN request's kind, not reader B's"
    );
    assert_eq!(
        resp_b.kind,
        BrowseEntityKind::Commodity,
        "reader B must get back ITS OWN request's kind, not reader A's"
    );
}

/// T-1131: a `Detail` request for an id that doesn't exist in that kind's
/// table comes back `NotFound` over the wire (not an error, not a hang, not
/// silently dropped).
#[test]
fn browse_detail_unknown_id_is_not_found_over_tcp() {
    let (mut world, addr) = new_multi_connection_world();
    wire_browse_reader(&mut world);

    let client_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let mut stream = client_handle.join().expect("client thread panicked");

    send_browse_detail_request(&mut stream, BrowseEntityKind::Corporation, "no-such-corp");
    let resp = drive_browse_tick_until_response(&mut world, &mut stream);
    assert_eq!(resp.status, BrowseStatus::NotFound);
    assert!(resp.detail.is_none());
}

/// T-1131: an `Index` request against a kind with zero rows still comes back
/// `Ready` with an empty list, matching the existing city-names convention
/// (`CityNamesStatus`'s "unknown body -> Ready, empty" pattern) — over the
/// real wire, not just at the reader layer (see
/// `browse_reader::tests::index_stations_empty_table_is_empty_vec` for that
/// half of the proof).
#[test]
fn browse_index_empty_table_is_ready_with_empty_list_over_tcp() {
    use settled_reach_server::atlas::browse_reader::{BrowseReader, BrowseReaderResource};

    let (mut world, addr) = new_multi_connection_world();

    // A fixture db with the stations table present but genuinely empty — a
    // distinct db from wire_browse_reader's shared fixture (which always has
    // one station), built directly here so this test controls the "empty"
    // precondition explicitly rather than relying on incidental fixture state.
    let db_path =
        std::env::temp_dir().join(format!("sr_browse_tcp_empty_{}.db", std::process::id()));
    let _ = std::fs::remove_file(&db_path);
    {
        let conn = rusqlite::Connection::open(&db_path).expect("create empty fixture db");
        conn.execute_batch(
            "CREATE TABLE stations (
                 station_id TEXT PRIMARY KEY, system_id TEXT NOT NULL, orbits_body_id TEXT,
                 station_type TEXT NOT NULL, proper_name TEXT, population INTEGER,
                 economic_role TEXT, governance_type TEXT, docking_class TEXT,
                 has_gate_infrastructure INTEGER DEFAULT 0, district_count INTEGER
             );",
        )
        .expect("create empty stations table");
    }
    let reader = BrowseReader::open(&db_path).expect("open empty fixture db");
    world.insert_resource(BrowseReaderResource(reader));

    let client_handle = thread::spawn(move || {
        let stream = TcpStream::connect(addr).expect("failed to connect");
        client_handshake(stream, ConnectionRole::Reader)
    });
    drive_accept_loop_until(&mut world, |w| {
        w.resource::<BridgeResource>().reader_count() == 1
    });
    let mut stream = client_handle.join().expect("client thread panicked");

    send_browse_index_request(&mut stream, BrowseEntityKind::Station);
    let resp = drive_browse_tick_until_response(&mut world, &mut stream);
    assert_eq!(resp.status, BrowseStatus::Ready);
    assert_eq!(resp.index.unwrap().len(), 0);

    let _ = std::fs::remove_file(&db_path);
}

// -----------------------------------------------------------------------
// ConnectionId(0) district_window delivery regression (live-verified bug,
// coordinator repro 2026-07-22): a fresh server's FIRST connection (whatever
// role) never received a district_window response over the wire, while a
// second connection to the same (or a fresh) server was served correctly.
// Every existing window test in layer_proxy.rs's unit-test module calls
// `test_conn_id() -> ConnectionId(1)` — NEVER ConnectionId(0) — so this class
// of bug had zero unit-test coverage. This test drives the REAL
// id-assignment path (`BridgeResource::default()` + `insert_reader`, the
// exact path `main.rs`'s first-connection handling and
// `accept_new_connections`'s reader promotion both use) rather than a
// hand-picked id, over the real TCP wire, through the real
// receive->serve->drain->send tick-phase pipeline.
// -----------------------------------------------------------------------

mod connection_zero_window_delivery {
    use super::*;
    use bevy_app::App;
    use rusqlite::Connection;
    use settled_reach_server::atlas::body_params_reader::{
        BodyParamsReader, BodyParamsReaderResource,
    };
    use settled_reach_server::atlas::cascade::CascadeLayer;
    use settled_reach_server::atlas::layer_proxy::{
        AtlasLayerRequest, AtlasLayerStatus, WindowGranularity,
    };
    use settled_reach_server::atlas::source_resolver::{
        BodySourceResolver, BodySourceResolverResource,
    };
    use settled_reach_server::atlas::GenerationPlugin;
    use settled_reach_server::bridge::{
        BridgePlugin, ConnectionId, ConnectionListener, HandshakeState, PendingConnections,
    };
    use settled_reach_server::simulation::SimulationPlugin;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU32, Ordering};

    static SEQ: AtomicU32 = AtomicU32::new(0);

    /// Same fixture shape as `layer_proxy.rs`'s own `write_tiny_heightmap`
    /// (not exported for cross-crate reuse — duplicated intentionally rather
    /// than widening that module's visibility for one integration test).
    fn write_tiny_heightmap(path: &Path) {
        use std::io::BufWriter;
        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let file = std::fs::File::create(path).unwrap();
        let mut enc = png::Encoder::new(BufWriter::new(file), 32, 16);
        enc.set_color(png::ColorType::Grayscale);
        enc.set_depth(png::BitDepth::Sixteen);
        let mut w = enc.write_header().unwrap();
        let data: Vec<u8> = (0..32u32 * 16)
            .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes())
            .collect();
        w.write_image_data(&data).unwrap();
    }

    /// Same fixture shape as `layer_proxy.rs`'s `resolver_and_params_reader`
    /// (duplicated for the same cross-crate-visibility reason as above).
    fn resolver_and_params_reader(
        body_id: &str,
    ) -> (BodySourceResolver, BodyParamsReader, PathBuf) {
        const REL: &str = "wiki/star-systems/GJ-1/bodies/GJ1c/heightmap.png";
        let n = SEQ.fetch_add(1, Ordering::Relaxed);
        let db = std::env::temp_dir().join(format!("sr_connzero_{}_{n}.db", std::process::id()));
        let _ = std::fs::remove_file(&db);
        let conn = Connection::open(&db).unwrap();
        conn.execute_batch(
            "CREATE TABLE star_systems (
                 system_id TEXT PRIMARY KEY,
                 spectral_class TEXT,
                 star_type TEXT
             );
             CREATE TABLE bodies (
                 body_id TEXT PRIMARY KEY,
                 system_id TEXT,
                 terrain_reference TEXT,
                 hydrosphere TEXT,
                 atmosphere TEXT,
                 planet_class TEXT,
                 body_radius_km REAL,
                 orbital_period_days REAL,
                 axial_tilt_deg REAL
             );",
        )
        .unwrap();
        conn.execute(
            "INSERT INTO star_systems (system_id, spectral_class, star_type) VALUES ('GJ-1', 'G', 'main_sequence')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO bodies (body_id, system_id, terrain_reference, hydrosphere, atmosphere, planet_class, body_radius_km, orbital_period_days, axial_tilt_deg)
             VALUES (?1, 'GJ-1', ?2, 'ocean', 'breathable', 'temperate', 6371.0, 365.25, 23.5)",
            rusqlite::params![body_id, REL],
        )
        .unwrap();
        drop(conn);

        let root = std::env::temp_dir().join(format!("sr_connzeroroot_{}_{n}", std::process::id()));
        write_tiny_heightmap(&root.join(REL));

        let resolver = BodySourceResolver::open(&db, vec![root.clone()]).unwrap();
        let params_reader = BodyParamsReader::open(&db).unwrap();
        (resolver, params_reader, root)
    }

    /// A district-window `AtlasLayerRequest` at Region granularity — the
    /// exact rung the coordinator's live repro used (six Region tile
    /// requests on body entry).
    fn region_window_request(body_id: &str, center: (i32, i32)) -> AtlasLayerRequest {
        AtlasLayerRequest {
            body_id: body_id.to_string(),
            up_to: CascadeLayer::Topography,
            window_center: Some(center),
            window_n: 4,
            window_granularity_v2: Some(WindowGranularity::Region),
            window_min_wl_m: 0,
        }
    }

    /// Build a real `App`, replicating `main.rs`'s ACTUAL bootstrap sequence
    /// for the first connection — not an approximation. `BridgePlugin` never
    /// `init_resource`s `BridgeResource` itself (confirmed: no
    /// `init_resource`/`insert_resource::<BridgeResource>` anywhere in
    /// `bridge/mod.rs`'s `Plugin::build`) — `main.rs` constructs it AFTER a
    /// single BLOCKING `TcpListener::accept()` + `TcpBridge::accept_on()` +
    /// handshake/startup exchange, pre-populated with that first connection
    /// already installed via `insert_player`/`insert_reader`, and only
    /// THEN starts the tick loop with a SEPARATE non-blocking listener clone
    /// for `accept_new_connections` to handle connections 2+. This function
    /// reproduces exactly that two-listener-handle, blocking-then-async
    /// split (`main.rs` lines ~104-145, ~368-377) rather than the
    /// all-non-blocking shortcut the first draft of this test used (which
    /// masked the real bootstrap entirely and hit "BridgeResource does not
    /// exist" — the WRONG failure, not the bug under investigation).
    ///
    /// `role` selects Reader (the standalone Atlas companion's real role,
    /// matching the coordinator's live repro) or Player.
    fn accept_first_connection_and_build_app(
        resolver: BodySourceResolver,
        params_reader: settled_reach_server::atlas::body_params_reader::BodyParamsReader,
        role: ConnectionRole,
    ) -> (App, std::net::SocketAddr, TcpStream) {
        let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind");
        let addr = listener.local_addr().expect("failed to get local address");
        // try_clone BEFORE any accept, mirroring main.rs exactly: one handle
        // for the blocking first accept, a second (later set non-blocking)
        // for the async per-tick accept-loop — the same underlying socket.
        let listener_for_loop = listener.try_clone().expect("failed to clone listener");

        let client_handle = thread::spawn(move || {
            let stream = TcpStream::connect(addr).expect("failed to connect");
            client_handshake(stream, role)
        });

        // Blocking accept — genuinely blocks this thread until the client
        // above connects, exactly like main.rs's listener.accept() call.
        let bridge = TcpBridge::accept_on(listener).expect("failed to accept first connection");
        // main.rs's real sequence (lines ~145-161): send the protocol
        // handshake, THEN read the client's StartupMessage — BEFORE the
        // client thread's client_handshake() (which reads the handshake
        // first) can complete. Omitting these two calls was the first
        // draft's bug: both sides deadlocked waiting on each other with
        // nothing ever written first.
        bridge.send_handshake().expect("failed to send handshake");
        let _startup = bridge
            .receive_startup()
            .expect("failed to receive startup message");
        let client_stream = client_handle.join().expect("client thread panicked");

        let mut bridge_resource = BridgeResource::default();
        let first_id = match role {
            ConnectionRole::Player => bridge_resource.insert_player(bridge),
            ConnectionRole::Reader => bridge_resource.insert_reader(bridge),
        };
        assert_eq!(
            first_id,
            ConnectionId(0),
            "sanity: the first-ever insert on a fresh BridgeResource must be id 0"
        );

        // Full main.rs plugin composition (not a hand-picked subset) — a
        // missing resource from any of these panicked the first draft
        // (KnowledgePlugin/NpcPlugin/StorytellerPlugin/SettingsPlugin/
        // BookmarkPlugin's systems are NOT gated behind Option<Res<...>> the
        // way GenerationPlugin's atlas-reader resources are), so matching
        // main.rs exactly is the correct fix, not narrowing the plugin set.
        let mut app = App::new();
        app.add_plugins(SimulationPlugin { seed: 42 });
        app.add_plugins(BridgePlugin);
        app.add_plugins(settled_reach_server::knowledge::KnowledgePlugin);
        app.add_plugins(settled_reach_server::npc::NpcPlugin);
        app.add_plugins(settled_reach_server::storyteller::StorytellerPlugin);
        app.add_plugins(settled_reach_server::settings::SettingsPlugin);
        app.add_plugins(settled_reach_server::bookmark::BookmarkPlugin::default());
        app.add_plugins(GenerationPlugin);
        // main.rs unconditionally calls setup_proof_room()/setup_gauntlet()
        // AFTER building every plugin above — even for a Reader-only spawn
        // (D-254 §1's own comment on that call site: "this call is NOT
        // gated on the first connection's role ... A Reader-only spawned
        // server therefore has an inert, unpiloted PlayerCharacter entity").
        // setup_proof_room isn't `pub` (private to the main.rs binary crate)
        // so it can't be called directly from an integration test; inserting
        // WalkabilityMap alone (the specific resource several PreInput/
        // Movement-phase systems require unconditionally, confirmed by this
        // test's first draft panicking with it absent) is the minimal
        // equivalent for a Reader-only session that spawns no character —
        // matching game_loop.rs's own established minimal pattern.
        app.insert_resource(
            settled_reach_server::simulation::movement::WalkabilityMap::new(32, 32, 1),
        );
        app.insert_resource(bridge_resource);
        app.insert_resource(HandshakeState::Complete);
        // Non-blocking for the tick loop's accept_new_connections, exactly
        // as main.rs's listener_for_loop.set_nonblocking(true) does.
        listener_for_loop
            .set_nonblocking(true)
            .expect("failed to set accept-loop listener non-blocking");
        app.insert_resource(ConnectionListener(Some(listener_for_loop)));
        app.world_mut()
            .resource_mut::<PendingConnections>()
            .0
            .clear();
        app.insert_resource(BodySourceResolverResource(resolver));
        app.insert_resource(BodyParamsReaderResource(params_reader));
        (app, addr, client_stream)
    }

    /// Drive `app.update()` (one full real tick — every registered system,
    /// in the real `TickPhase` order) repeatedly, checking the accept-loop
    /// condition after each — mirrors `drive_accept_loop_until`'s pattern but
    /// against a full `App` rather than a hand-picked system.
    fn drive_app_accept_loop_until(app: &mut App, condition: impl Fn(&App) -> bool) {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while !condition(app) {
            assert!(
                std::time::Instant::now() < deadline,
                "timed out waiting for accept-loop condition"
            );
            app.update();
            thread::sleep(std::time::Duration::from_millis(1));
        }
    }

    /// Drive `app.update()` ticks until a `Ready` `AtlasLayerResponse`
    /// carrying a populated `district_window` arrives on `stream`, or a
    /// wall-clock deadline expires — mirrors
    /// `drive_browse_tick_until_response`'s established pattern (bounded
    /// retry loop over a genuinely blocking client-side socket, not a fixed
    /// sleep/tick count) but must also tolerate intermediate `Pending`
    /// frames (the D-225 poll-and-recheck-cache contract: the FIRST response
    /// for a cold window is `Ready` with `district_window: None`, not the
    /// final answer) by reading and discarding them until the populated one
    /// lands.
    fn drive_app_until_window_response(
        app: &mut App,
        stream: &mut TcpStream,
        body_id: &str,
        center: (i32, i32),
    ) -> settled_reach_server::atlas::layer_proxy::AtlasLayerResponse {
        stream
            .set_read_timeout(Some(std::time::Duration::from_millis(50)))
            .expect("failed to set read timeout");
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
        loop {
            app.update();
            match read_framed(stream) {
                Ok(Some(payload)) => {
                    let resp: settled_reach_server::atlas::layer_proxy::AtlasLayerResponse =
                        rmp_serde::from_slice(&payload)
                            .expect("failed to decode AtlasLayerResponse");
                    if resp.district_window.is_some() {
                        return resp;
                    }
                    // Pending (cold cache) or Ready-with-None (derive still
                    // in flight) — re-request and keep polling, exactly the
                    // client's real re-poll behavior (atlas_window_request.gd's
                    // on_response()/_schedule_retry(): the server never pushes
                    // a second response on its own — D-225's poll-and-recheck-
                    // cache contract requires the CLIENT to re-send). Omitting
                    // this re-send was this test's own first-draft bug: the
                    // request was consumed by serve_atlas_requests within the
                    // SAME app.update() it arrived in (receive -> serve ->
                    // send all run inside one PreInput/PostSnapshot pass), so
                    // every later tick correctly had nothing left to serve —
                    // not a production hang, a missing re-poll in the harness.
                    assert!(
                        std::time::Instant::now() < deadline,
                        "timed out waiting for a populated district_window \
                         (last status: {:?})",
                        resp.status
                    );
                    send_region_window_request(stream, body_id, center);
                }
                Ok(None) => panic!("unexpected EOF reading atlas response"),
                Err(e)
                    if e.kind() == std::io::ErrorKind::WouldBlock
                        || e.kind() == std::io::ErrorKind::TimedOut =>
                {
                    assert!(
                        std::time::Instant::now() < deadline,
                        "timed out waiting for an atlas response frame at all"
                    );
                }
                Err(e) => panic!("failed to read atlas response frame: {e}"),
            }
        }
    }

    fn send_region_window_request(stream: &mut TcpStream, body_id: &str, center: (i32, i32)) {
        let req = region_window_request(body_id, center);
        let payload = rmp_serde::to_vec_named(&req).expect("failed to encode AtlasLayerRequest");
        write_framed(stream, &payload).expect("failed to write AtlasLayerRequest");
    }

    /// **THE regression test.** A fresh server's FIRST connection ever
    /// accepted (Reader role, matching the standalone Atlas companion) must
    /// receive its district_window response — proven via the REAL
    /// `insert_reader`-assigned id, not `ConnectionId(1)` the way every
    /// existing window unit test does. If this test alone is added without
    /// re-requesting after the initial Pending (i.e. treats the first
    /// response as final), it would have passed even on a broken server —
    /// the assertion on `resp.status == Ready` AND `district_window.is_some()`
    /// together, reached only via `drive_window_tick_until_response`'s
    /// re-poll loop, is what actually exercises the async derive-then-cache
    /// path a live client depends on.
    #[test]
    fn first_connection_ever_accepted_receives_district_window() {
        let (resolver, params_reader, _root) = resolver_and_params_reader("GJ1c");
        let (mut app, _addr, mut stream) =
            accept_first_connection_and_build_app(resolver, params_reader, ConnectionRole::Reader);

        // Confirm this really is the id-zero path before asserting delivery
        // — a false pass here (e.g. BridgeResource pre-seeded elsewhere)
        // would silently defeat the whole point of the test. (Also asserted
        // inside accept_first_connection_and_build_app; re-checked here at
        // the point of use for a self-contained failure message.)
        let conn_id = app
            .world()
            .resource::<BridgeResource>()
            .reader_ids()
            .first()
            .copied()
            .expect("reader must be installed");
        assert_eq!(
            conn_id,
            ConnectionId(0),
            "this test only proves what it claims to prove if the reader under \
             test genuinely got the FIRST id a fresh BridgeResource ever assigns"
        );

        send_region_window_request(&mut stream, "GJ1c", (0, 0));
        let resp = drive_app_until_window_response(&mut app, &mut stream, "GJ1c", (0, 0));

        assert_eq!(resp.status, AtlasLayerStatus::Ready);
        assert_eq!(resp.body_id, "GJ1c");
        let window = resp
            .district_window
            .expect("district_window must be populated (checked above; re-asserted for clarity)");
        assert_eq!(window.granularity_v2, WindowGranularity::Region);
        assert!(
            !window.morphology.is_empty(),
            "the delivered window must carry real derived cell data, not an empty payload"
        );
    }

    /// The EXACT shape of the coordinator's live repro: six Region tile
    /// requests fired on body entry (the standalone Atlas companion's real
    /// window-tile-set fan-out, not a single request) on connection 0, all
    /// six must be delivered — not just the first, in case a many-in-flight
    /// scenario surfaces an ordering issue a single-request test can't see
    /// (e.g. coalescing/supersede logic dropping later requests, or the
    /// per-tick MAX_READER_INBOUND_FRAMES_PER_TICK cap interacting badly
    /// with six queued frames).
    #[test]
    fn first_connection_receives_all_six_region_tiles() {
        let (resolver, params_reader, _root) = resolver_and_params_reader("GJ1c");
        let (mut app, _addr, mut stream) =
            accept_first_connection_and_build_app(resolver, params_reader, ConnectionRole::Reader);

        let conn_id = app
            .world()
            .resource::<BridgeResource>()
            .reader_ids()
            .first()
            .copied()
            .expect("reader must be installed");
        assert_eq!(conn_id, ConnectionId(0));

        // Six distinct region-tile centers, matching the design doc's tiled
        // orbital-view worked example shape (a 3x2 or similar tile grid
        // around the entry point) — distinct centers so each is a genuinely
        // separate cache entry, not accidental coalescing onto one.
        let centers: [(i32, i32); 6] = [
            (0, 0),
            (12739, -3200),
            (12739, 3200),
            (0, -3200),
            (0, 3200),
            (6400, -3200),
        ];

        for &center in &centers {
            send_region_window_request(&mut stream, "GJ1c", center);
        }

        let mut received: std::collections::BTreeSet<(i32, i32)> =
            std::collections::BTreeSet::new();
        stream
            .set_read_timeout(Some(std::time::Duration::from_millis(50)))
            .expect("failed to set read timeout");
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15);
        while received.len() < centers.len() {
            assert!(
                std::time::Instant::now() < deadline,
                "timed out with only {}/{} tiles delivered: {:?}",
                received.len(),
                centers.len(),
                received
            );
            app.update();
            match read_framed(&mut stream) {
                Ok(Some(payload)) => {
                    let resp: settled_reach_server::atlas::layer_proxy::AtlasLayerResponse =
                        rmp_serde::from_slice(&payload)
                            .expect("failed to decode AtlasLayerResponse");
                    if let Some(window) = resp.district_window {
                        received.insert(window.center);
                    } else {
                        // Pending/Ready-with-None for one of the six — re-send
                        // ALL not-yet-received centers (mirrors the real
                        // client's per-tile independent re-poll).
                        for &center in centers.iter().filter(|c| !received.contains(c)) {
                            send_region_window_request(&mut stream, "GJ1c", center);
                        }
                    }
                }
                Ok(None) => panic!("unexpected EOF reading atlas response"),
                Err(e)
                    if e.kind() == std::io::ErrorKind::WouldBlock
                        || e.kind() == std::io::ErrorKind::TimedOut => {}
                Err(e) => panic!("failed to read atlas response frame: {e}"),
            }
        }

        assert_eq!(
            received.len(),
            centers.len(),
            "all six region tiles must be delivered to connection 0"
        );
    }

    /// The delivery-order counterpart: a SECOND connection's request must
    /// ALSO be served correctly (this already passed in the coordinator's
    /// live repro — included here as a same-file regression guard so a
    /// future fix to the id-0 case can't accidentally break id-1 while
    /// fixing id-0, and so this file documents the FULL observed shape of
    /// the bug, not just half of it).
    #[test]
    fn second_connection_also_receives_district_window() {
        let (resolver, params_reader, _root) = resolver_and_params_reader("GJ1c");
        // First connection occupies ConnectionId(0) via the SAME real
        // blocking-accept bootstrap main.rs uses (see
        // accept_first_connection_and_build_app's doc) but sends no
        // requests — isolates "is it specifically the SECOND id that works"
        // from "does a second connection existing change anything for the
        // first".
        let (mut app, addr, _first_stream) =
            accept_first_connection_and_build_app(resolver, params_reader, ConnectionRole::Reader);

        let second_handle = thread::spawn(move || {
            let stream = TcpStream::connect(addr).expect("failed to connect");
            client_handshake(stream, ConnectionRole::Reader)
        });
        drive_app_accept_loop_until(&mut app, |a| {
            a.world().resource::<BridgeResource>().reader_count() == 2
        });
        let mut second_stream = second_handle.join().expect("client thread panicked");

        let second_id = app
            .world()
            .resource::<BridgeResource>()
            .reader_ids()
            .get(1)
            .copied()
            .expect("second reader must be installed");
        assert_eq!(second_id, ConnectionId(1));

        send_region_window_request(&mut second_stream, "GJ1c", (0, 0));
        let resp = drive_app_until_window_response(&mut app, &mut second_stream, "GJ1c", (0, 0));

        assert_eq!(resp.status, AtlasLayerStatus::Ready);
        assert!(resp.district_window.is_some());
    }
}

/// Minimal `ObserverSnapshot` for the tests above — same shape as
/// `snapshot_roundtrip_over_tcp`'s at the top of this file, parameterized
/// only by `tick` (the one field these tests assert on).
fn sample_snapshot(tick: u64) -> ObserverSnapshot {
    ObserverSnapshot {
        tick,
        game_time: GameTime {
            day: 0,
            time_of_day: 0,
            day_phase: DayPhase::Morning,
            tick_rate: TickRate::Full,
        },
        player_facing: FacingDirection::North,
        player_stance: MovementStance::default(),
        player_inventory: vec![],
        entities: vec![],
        visible_tiles: vec![],
        nearby_interactions: vec![],
        current_monologue: None,
        pending_recognitions: vec![],
        dialogue_response: None,
        blocked_entities: vec![],
        scan_events: vec![],
        sound_events: vec![],
        follow_state: None,
        character_pressure: None,
        rng_seed: None,
        poi_list: vec![],
        examine_result: None,
        player_knowledge: None,
        save_result: None,
        triangle_crisis_events: vec![],
        state_hash: None,
        debug_response: None,
        sim_errors: vec![],
        current_ticker: None,
        settings_response: None,
        economy_snapshot: None,
        bookmark_catalog: None,
    }
}