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

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::{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, 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,
        };
        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
        .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, CityNamesRequestBuffer,
        CityNamesResponseBuffer, ConnectionListener, HandshakeState, PendingConnections,
        ServerRunning, SnapshotBuffer, StarMapRequestBuffer, StarMapResponseBuffer,
    };
    use settled_reach_server::simulation::input::InputQueue;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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