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, HandshakeState, ServerRunning, StarMapRequestBuffer,
    };
    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: five frames back-to-back in one tick window — two input
    // batches plus one of EACH request shape (atlas, star-map, city-names:
    // the full D-225/T-949 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, 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,
        };
        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");
        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
        .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");
    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,
        HandshakeState, PendingConnections, ServerRunning, SnapshotBuffer, StarMapRequestBuffer,
        StarMapResponseBuffer,
    };
    use settled_reach_server::simulation::input::InputQueue;

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

    let mut world = bevy_ecs::world::World::new();
    world.insert_resource(ConnectionListener(Some(listener)));
    world.init_resource::<PendingConnections>();
    world.insert_resource(BridgeResource::default());
    world.init_resource::<InputQueue>();
    world.init_resource::<ServerRunning>();
    world.insert_resource(HandshakeState::Complete);
    world.init_resource::<SimErrorBuffer>();
    world.init_resource::<AtlasRequestBuffer>();
    world.init_resource::<AtlasResponseBuffer>();
    world.init_resource::<StarMapRequestBuffer>();
    world.init_resource::<StarMapResponseBuffer>();
    world.init_resource::<CityNamesRequestBuffer>();
    world.init_resource::<CityNamesResponseBuffer>();
    world.init_resource::<BrowseRequestBuffer>();
    world.init_resource::<BrowseResponseBuffer>();
    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);
}

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