use super::*;

fn receive_render(receiver: &std::sync::mpsc::Receiver<Vec<u8>>, timeout: Duration) -> Vec<u8> {
    receiver.recv_timeout(timeout).unwrap()
}

#[tokio::test]
async fn client_shell_surface_sends_complete_placements_and_each_live_asset_once() {
    let (mut server, _control_rx, client_rx, pane_id) =
        retained_test_server_with_control(b"client shell graphics");
    let client = server.clients.get_mut(&1).unwrap();
    client.mode = ClientConnectionMode::ClientShell;
    client.render_state =
        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };
    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3, 4]);

    server.render_and_stream();
    let first = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(first) = first else {
        panic!("expected client shell pane surface");
    };
    assert_eq!(first.graphics.placements.len(), 1);
    assert_eq!(first.graphics.assets.len(), 1);
    assert_eq!(first.graphics.assets[0].data, vec![1, 2, 3, 4]);
    assert!(matches!(
        first.graphics.placements[0].asset.source,
        crate::protocol::SurfaceGraphicsSource::PaneLayer { .. }
    ));

    server.clients.get_mut(&1).unwrap().request_repaint();
    server.render_and_stream();
    let second = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(second) = second else {
        panic!("expected replacement client shell pane surface");
    };
    assert_eq!(second.graphics.placements, first.graphics.placements);
    assert!(second.graphics.assets.is_empty());

    // A post-commit typed surface.set(true) resets only this viewer's delivery cache, so the
    // already selected target receives the asset again without relying on graphics that may have
    // arrived during the frozen source frame.
    assert!(server.set_client_shell_surface_active(1, true).is_some());
    server.render_and_stream();
    let replay = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(replay) = replay else {
        panic!("expected post-commit graphics replay");
    };
    assert_eq!(replay.graphics.placements, first.graphics.placements);
    assert_eq!(replay.graphics.assets.len(), 1);
    assert_eq!(replay.graphics.assets[0].data, vec![1, 2, 3, 4]);
}

#[tokio::test]
async fn client_shell_asset_delivery_is_bounded_to_the_current_live_scene() {
    let (mut server, client_rx, pane_id) = retained_test_server(b"client shell graphics");
    let client = server.clients.get_mut(&1).unwrap();
    client.mode = ClientConnectionMode::ClientShell;
    client.render_state =
        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };
    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3, 4]);

    server.render_and_stream();
    let _first = receive_render(&client_rx, Duration::from_millis(100));
    server.app.pane_graphics.slots.clear();
    server.app.pane_graphics.mark_changed();
    server.clients.get_mut(&1).unwrap().request_repaint();
    server.render_and_stream();
    let removed = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(removed) = removed else {
        panic!("expected removed client shell scene");
    };
    assert!(removed.graphics.placements.is_empty());

    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3, 4]);
    server.clients.get_mut(&1).unwrap().request_repaint();
    server.render_and_stream();
    let restored = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(restored) = restored else {
        panic!("expected restored client shell scene");
    };
    assert_eq!(restored.graphics.assets.len(), 1);
    assert_eq!(restored.graphics.assets[0].data, vec![1, 2, 3, 4]);
}

#[tokio::test]
async fn first_kitty_image_updates_retained_surface_without_full_redraw() {
    let (mut server, client_rx, pane_id) = retained_test_server(b"text before image");
    let client = server.clients.get_mut(&1).unwrap();
    client.mode = ClientConnectionMode::ClientShell;
    client.render_state =
        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };
    server.render_and_stream();
    let ServerMessage::PaneSurface(initial) =
        read_server_message(receive_render(&client_rx, Duration::from_millis(100)))
    else {
        panic!("expected text-only baseline");
    };
    assert!(initial.graphics.placements.is_empty());

    write_shared_test_pane(
        &mut server,
        pane_id,
        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
    );
    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
    let ServerMessage::PaneSurface(repainted) =
        read_server_message(receive_render(&client_rx, Duration::from_millis(100)))
    else {
        panic!("expected image update without a full redraw");
    };
    assert_eq!(repainted.graphics.placements.len(), 1);
    assert_eq!(repainted.graphics.assets[0].data, [255, 0, 0, 255]);
    assert_eq!(repainted.panes[0].inner_rect, initial.panes[0].inner_rect);

    // Text changes while an image is visible must reuse its uploaded pixels.
    write_shared_test_pane(&mut server, pane_id, b"\rupdated text");
    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
    let ServerMessage::PaneSurface(text_update) =
        read_server_message(receive_render(&client_rx, Duration::from_millis(100)))
    else {
        panic!("expected retained text and image scene");
    };
    assert_eq!(
        text_update.graphics.placements,
        repainted.graphics.placements
    );
    assert!(text_update.graphics.assets.is_empty());
    assert!(frame_text(&text_update.frame).contains("updated text"));

    write_shared_test_pane(&mut server, pane_id, b"\x1b_Ga=d,d=A\x1b\\");
    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
    let ServerMessage::PaneSurface(deleted) =
        read_server_message(receive_render(&client_rx, Duration::from_millis(100)))
    else {
        panic!("expected image removal");
    };
    assert!(deleted.graphics.placements.is_empty());
    assert!(deleted.graphics.assets.is_empty());

    write_shared_test_pane(&mut server, pane_id, b"\rtext only again");
    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
    assert!(matches!(
        read_server_message(receive_render(&client_rx, Duration::from_millis(100))),
        ServerMessage::PaneSurfacePatch(_)
    ));

    // A full output queue must not mark unsent pixels as delivered.
    fill_render_lane(&server);
    write_shared_test_pane(
        &mut server,
        pane_id,
        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
    );
    assert!(server.render_retained_pane_surface_and_stream(&HashSet::from([pane_id])));
    assert!(server.clients[&1]
        .render_state
        .last_pane_surface()
        .unwrap()
        .graphics
        .placements
        .is_empty());
    let _ = receive_render(&client_rx, Duration::from_millis(100));
    server.render_and_stream();
    let ServerMessage::PaneSurface(recovered) =
        read_server_message(receive_render(&client_rx, Duration::from_millis(100)))
    else {
        panic!("expected deferred graphics recovery");
    };
    assert_eq!(recovered.graphics.assets[0].data, [255, 0, 0, 255]);
}

#[tokio::test]
async fn retained_unicode_image_arrives_after_fragmented_upload_without_reupload() {
    let (mut server, client_rx, pane_id) = retained_test_server(b"\x1b[?1049h");
    let client = server.clients.get_mut(&1).unwrap();
    client.mode = ClientConnectionMode::ClientShell;
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };
    server.render_and_stream();
    let _ = receive_render(&client_rx, Duration::from_millis(100));
    let sources = HashSet::from([pane_id]);
    // Yazi-style virtual placement: uploading the image and drawing its Unicode cell
    // can happen in separate PTY reads, with no text dirty rows when upload completes.
    for bytes in [
        b"\x1b_Ga=t,f=32,t=d,i=1193046,s=1,v=1,q=2;/wAA/w".as_slice(),
        b"==\x1b\\",
        b"\x1b_Ga=p,U=1,i=1193046,c=1,r=1,q=2\x1b\\",
    ] {
        write_shared_test_pane(&mut server, pane_id, bytes);
        assert!(server.render_retained_pane_surface_and_stream(&sources));
        for frame in client_rx.try_iter() {
            if let ServerMessage::PaneSurface(surface) = read_server_message(frame) {
                assert!(surface.graphics.placements.is_empty());
            }
        }
    }
    write_shared_test_pane(
        &mut server,
        pane_id,
        "\x1b[2;3H\x1b[38;2;18;52;86m\u{10eeee}\u{0305}\u{0305}\x1b[0m".as_bytes(),
    );
    assert!(server.render_retained_pane_surface_and_stream(&sources));
    let ServerMessage::PaneSurface(surface) =
        read_server_message(receive_render(&client_rx, Duration::from_millis(100)))
    else {
        panic!("virtual image must arrive without a tab switch");
    };
    assert_eq!(surface.graphics.placements.len(), 1);
    assert_eq!(surface.graphics.assets[0].data, [255, 0, 0, 255]);
    // Retransmission removes placements; recreating the virtual placement must
    // invalidate the delivered asset without needing another text update.
    write_shared_test_pane(
        &mut server,
        pane_id,
        b"\x1b_Ga=t,f=32,t=d,i=1193046,s=1,v=1,q=2;AP8A/w==\x1b\\",
    );
    assert!(server.render_retained_pane_surface_and_stream(&sources));
    let ServerMessage::PaneSurface(removed) =
        read_server_message(receive_render(&client_rx, Duration::from_millis(100)))
    else {
        panic!("retransmission must remove the virtual placement");
    };
    assert!(removed.graphics.placements.is_empty());
    write_shared_test_pane(
        &mut server,
        pane_id,
        b"\x1b_Ga=p,U=1,i=1193046,c=1,r=1,q=2\x1b\\",
    );
    assert!(server.render_retained_pane_surface_and_stream(&sources));
    let ServerMessage::PaneSurface(replaced) =
        read_server_message(receive_render(&client_rx, Duration::from_millis(100)))
    else {
        panic!("updated image pixels must arrive");
    };
    assert_eq!(replaced.graphics.assets[0].data, [0, 255, 0, 255]);
    assert_ne!(
        surface.graphics.assets[0].key,
        replaced.graphics.assets[0].key
    );
}

#[tokio::test]
#[ignore = "manual retained text/image scaling profile"]
async fn render_scale_profile_retained_graphics() {
    use ratatui::layout::Direction;
    for retained in [false, true] {
        for with_image in [false, true] {
            for count in [1, 15] {
                let (mut server, client_rx, root) = retained_test_server(b"populated terminal\r\n");
                let mut pane_ids = vec![root];
                for index in 1..count {
                    let workspace = &mut server.app.state.workspaces[0];
                    workspace.tabs[0]
                        .layout
                        .focus_pane(pane_ids[(index - 1) / 2]);
                    let id = workspace.test_split(if index % 2 == 0 {
                        Direction::Vertical
                    } else {
                        Direction::Horizontal
                    });
                    workspace.insert_test_runtime(
                        id,
                        crate::terminal::TerminalRuntime::test_with_screen_bytes(
                            80,
                            24,
                            b"populated terminal\r\n",
                        ),
                    );
                    pane_ids.push(id);
                }
                let client = server.clients.get_mut(&1).unwrap();
                client.mode = ClientConnectionMode::ClientShell;
                client.cell_size = crate::kitty_graphics::HostCellSize {
                    width_px: 10,
                    height_px: 20,
                };
                if with_image {
                    write_shared_test_pane(
                        &mut server,
                        root,
                        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
                    );
                }
                server.render_and_stream();
                let _ = receive_render(&client_rx, Duration::from_millis(100));
                let sources = pane_ids.iter().copied().collect();
                let mut samples = Vec::new();
                for sample in 0..110 {
                    for id in &pane_ids {
                        write_shared_test_pane(
                            &mut server,
                            *id,
                            format!("\x1b[H{sample:03}").as_bytes(),
                        );
                    }
                    let started = Instant::now();
                    if retained {
                        assert!(server.render_retained_pane_surface_and_stream(&sources));
                    } else {
                        server.render_and_stream();
                    }
                    let elapsed = started.elapsed();
                    for frame in client_rx.try_iter() {
                        if let ServerMessage::PaneSurface(surface) = read_server_message(frame) {
                            assert!(surface.graphics.assets.is_empty());
                        }
                    }
                    if sample >= 10 {
                        samples.push(elapsed);
                    }
                }
                samples.sort_unstable();
                println!(
                "retained={retained} 80x24 panes={count} image={with_image} median_us={} p95_us={}",
                samples[50].as_micros(),
                samples[94].as_micros()
            );
            }
        }
    }
}

#[tokio::test]
async fn client_shell_surface_projects_terminal_kitty_images_from_authoritative_runtime() {
    let (mut server, client_rx, _pane_id) =
        retained_test_server(b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\");
    let client = server.clients.get_mut(&1).unwrap();
    client.mode = ClientConnectionMode::ClientShell;
    client.render_state =
        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };

    server.render_and_stream();
    let message = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(surface) = message else {
        panic!("expected client shell pane surface");
    };
    assert_eq!(surface.graphics.placements.len(), 1);
    assert_eq!(surface.graphics.assets.len(), 1);
    assert!(matches!(
        surface.graphics.placements[0].asset.source,
        crate::protocol::SurfaceGraphicsSource::Terminal {
            target: crate::protocol::SurfaceGraphicsTarget::Pane { .. },
            image_id: 7,
        }
    ));
    assert_eq!(surface.graphics.assets[0].data, vec![255, 0, 0, 255]);
}

#[tokio::test]
async fn client_shell_delivers_equal_pixels_for_distinct_terminal_image_ids() {
    let (mut server, client_rx, _pane_id) = retained_test_server(
        b"\x1b_Ga=T,f=32,t=d,i=7,p=3,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\\x1b_Ga=T,f=32,t=d,i=8,p=4,s=1,v=1,c=1,r=1,q=2;/wAA/w==\x1b\\",
    );
    let client = server.clients.get_mut(&1).unwrap();
    client.mode = ClientConnectionMode::ClientShell;
    client.render_state =
        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };

    server.render_and_stream();
    let message = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(surface) = message else {
        panic!("expected client shell pane surface");
    };
    assert_eq!(surface.graphics.placements.len(), 2);
    assert_eq!(surface.graphics.assets.len(), 2);
    assert_ne!(
        surface.graphics.assets[0].key,
        surface.graphics.assets[1].key
    );
    assert_eq!(
        surface.graphics.assets[0].data,
        surface.graphics.assets[1].data
    );

    server.clients.get_mut(&1).unwrap().request_repaint();
    server.render_and_stream();
    let message = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(surface) = message else {
        panic!("expected replacement client shell pane surface");
    };
    assert_eq!(surface.graphics.placements.len(), 2);
    assert!(surface.graphics.assets.is_empty());
}

#[tokio::test]
async fn full_client_shell_render_lane_does_not_commit_graphics_delivery() {
    let (mut server, client_rx, pane_id) = retained_test_server(b"client shell graphics");
    let client = server.clients.get_mut(&1).unwrap();
    client.mode = ClientConnectionMode::ClientShell;
    client.render_state =
        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };
    set_graphics_layer(&mut server, pane_id, vec![5, 6, 7, 8]);

    fill_render_lane(&server);
    server.render_and_stream();
    let _older = receive_render(&client_rx, Duration::from_millis(100));
    server.render_and_stream();
    let message = read_server_message(receive_render(&client_rx, Duration::from_millis(100)));
    let ServerMessage::PaneSurface(surface) = message else {
        panic!("expected client shell pane surface");
    };
    assert_eq!(surface.graphics.assets.len(), 1);
    assert_eq!(surface.graphics.assets[0].data, vec![5, 6, 7, 8]);
}

fn graphics_key(pane_id: crate::layout::PaneId) -> crate::app::pane_graphics::Key {
    (pane_id, api::schema::PANE_GRAPHICS_PRIMARY_LAYER_ID.into())
}

fn active_gate() -> std::sync::Arc<std::sync::atomic::AtomicBool> {
    std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true))
}

fn set_graphics_layer(server: &mut HeadlessServer, pane_id: crate::layout::PaneId, data: Vec<u8>) {
    set_named_graphics_layer(
        server,
        pane_id,
        api::schema::PANE_GRAPHICS_PRIMARY_LAYER_ID,
        data,
        0,
    );
}

fn set_named_graphics_layer(
    server: &mut HeadlessServer,
    pane_id: crate::layout::PaneId,
    layer_id: &str,
    data: Vec<u8>,
    z_index: i32,
) {
    let key = (pane_id, layer_id.into());
    let host_image_id = server.app.pane_graphics.reserve_image_id(&key).unwrap();
    let layer = crate::app::pane_graphics::Layer::inline(
        api::schema::PaneGraphicsFormat::Png,
        1,
        1,
        data,
        Default::default(),
        z_index,
    );
    server.app.pane_graphics.slots.insert(
        key,
        crate::app::pane_graphics::Slot::test(host_image_id, Some(layer)),
    );
}

fn set_stream_owner(server: &mut HeadlessServer, pane_id: crate::layout::PaneId, owner: &str) {
    let key = graphics_key(pane_id);
    if let Some(slot) = server.app.pane_graphics.slots.get_mut(&key) {
        slot.stream_owner = Some(owner.into());
        slot.stream_active = Some(active_gate());
    } else {
        let host_image_id = server.app.pane_graphics.reserve_image_id(&key).unwrap();
        let mut slot = crate::app::pane_graphics::Slot::test(host_image_id, None);
        slot.stream_owner = Some(owner.into());
        slot.stream_active = Some(active_gate());
        server.app.pane_graphics.slots.insert(key, slot);
    }
}

fn fill_render_lane(server: &HeadlessServer) {
    let queued = HeadlessServer::frame_server_message(&ServerMessage::ReloadSoundConfig)
        .expect("dummy frame");
    server.clients[&1]
        .writer
        .as_ref()
        .unwrap()
        .test_fill_render(queued);
}

fn stream_set_message(
    id: &str,
    pane_id: &str,
    owner: &str,
    data: Vec<u8>,
) -> (api::ApiRequestMessage, std::sync::mpsc::Receiver<String>) {
    let (respond_to, response_rx) = std::sync::mpsc::channel();
    (
        api::ApiRequestMessage {
            request: api::schema::Request {
                id: id.into(),
                method: api::schema::Method::PaneGraphicsStreamSet(
                    api::schema::PaneGraphicsSetParams {
                        pane_id: pane_id.into(),
                        layer_id: None,
                        z_index: 0,
                        owner: owner.into(),
                        format: api::schema::PaneGraphicsFormat::Png,
                        image_width: 1,
                        image_height: 1,
                        data: Some(data),
                        data_base64: String::new(),
                        placement: api::schema::PaneGraphicsPlacementParams::default(),
                    },
                ),
            },
            respond_to,
            response_write_complete: None,
            stream_active: None,
        },
        response_rx,
    )
}

#[cfg(unix)]
fn sparse_direct_frame(
    server: &HeadlessServer,
    name: &str,
    image_width: u32,
    image_height: u32,
) -> String {
    use std::os::unix::fs::OpenOptionsExt as _;

    let path = server
        .app
        .pane_graphics_files
        .source_directory()
        .unwrap()
        .join(name);
    let file = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .mode(0o600)
        .open(&path)
        .unwrap();
    file.set_len(u64::from(image_width) * u64::from(image_height) * 4)
        .unwrap();
    path.to_string_lossy().into_owned()
}

#[cfg(unix)]
fn direct_stream_message(
    id: &str,
    pane_id: &str,
    owner: &str,
    path: String,
    image_width: u32,
    image_height: u32,
    sequence: u64,
) -> (api::ApiRequestMessage, std::sync::mpsc::Receiver<String>) {
    let (respond_to, response_rx) = std::sync::mpsc::channel();
    (
        api::ApiRequestMessage {
            request: api::schema::Request {
                id: id.into(),
                method: api::schema::Method::PaneGraphicsStreamDirect(
                    api::schema::PaneGraphicsDirectParams {
                        pane_id: pane_id.into(),
                        layer_id: None,
                        z_index: 0,
                        owner: owner.into(),
                        image_width,
                        image_height,
                        format: api::schema::PaneGraphicsFormat::Rgba,
                        path,
                        sequence,
                        revision: 1,
                        placement: Default::default(),
                    },
                ),
            },
            respond_to,
            response_write_complete: None,
            stream_active: None,
        },
        response_rx,
    )
}

#[tokio::test]
async fn pixel_mouse_activation_follows_child_1016_without_graphics_demand() {
    let (mut server, _client_rx, _pane_id) =
        retained_test_server(b"\x1b[?1003h\x1b[?1006h\x1b[?1016h");
    let (writer, control_rx, _render_rx) = test_client_writer();
    let client = server.clients.get_mut(&1).unwrap();
    client.writer = Some(writer);
    client.direct_graphics = false;
    client.pixel_mouse = true;
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };
    client.host_mouse_capture_active = None;
    client.host_sgr_pixels_active = None;
    server.app.direct_graphics_available = false;

    server.stream_host_mouse_capture_mode();
    assert!(matches!(
        read_server_message(control_rx.recv_timeout(Duration::from_millis(100)).unwrap()),
        ServerMessage::MouseCapture {
            enabled: true,
            sgr_pixels: true
        }
    ));
}

#[tokio::test]
async fn graphics_pruning_preserves_live_panes_and_removes_closed_panes() {
    let (mut server, _client_rx, pane_id) = retained_test_server(b"aaaa");
    set_graphics_layer(&mut server, pane_id, vec![1, 2, 3]);

    assert!(!server
        .app
        .pane_graphics
        .retain_live_panes(&server.app.state));
    assert!(server
        .app
        .pane_graphics
        .slots
        .contains_key(&graphics_key(pane_id)));

    server.app.state.workspaces.clear();
    assert!(server
        .app
        .pane_graphics
        .retain_live_panes(&server.app.state));
    assert!(server.app.pane_graphics.slots.is_empty());
}

#[test]
fn stream_open_gate_is_owned_by_the_layer_and_cancels_on_removal() {
    let mut server = test_headless_server();
    server.app.state.kitty_graphics_enabled = true;
    let workspace = crate::workspace::Workspace::test_new("gated");
    let pane_id = workspace.tabs[0].root_pane;
    let public = format!("{}:p1", workspace.id);
    server.app.state.workspaces = vec![workspace];
    server.app.state.active = Some(0);
    let active = active_gate();
    let (respond_to, response_rx) = std::sync::mpsc::channel();

    server.handle_api_request_with_shutdown_check(api::ApiRequestMessage {
        request: api::schema::Request {
            id: "open-gated".into(),
            method: api::schema::Method::PaneGraphicsStreamOpen(
                api::schema::PaneGraphicsStreamParams {
                    pane_id: public.clone(),
                    layer_id: None,
                    z_index: 0,
                    owner: "worker-1".into(),
                },
            ),
        },
        respond_to,
        response_write_complete: None,
        stream_active: Some(active.clone()),
    });
    assert!(
        serde_json::from_str::<api::schema::SuccessResponse>(&response_rx.recv().unwrap()).is_ok()
    );
    let (frame, frame_response) =
        stream_set_message("gated-frame", &public, "worker-1", vec![1, 2, 3]);
    assert_eq!(
        server.handle_api_request_with_render_impact(frame),
        RenderImpact::Graphics
    );
    assert!(frame_response.recv().is_ok());
    assert!(active.load(std::sync::atomic::Ordering::Acquire));
    active.store(false, std::sync::atomic::Ordering::Release);
    let (delayed, delayed_response) =
        stream_set_message("delayed-frame", &public, "worker-1", vec![4, 5, 6]);
    assert_eq!(
        server.handle_api_request_with_render_impact(delayed),
        RenderImpact::None
    );
    let error: api::schema::ErrorResponse =
        serde_json::from_str(&delayed_response.recv().unwrap()).unwrap();
    assert_eq!(error.error.code, "stream_closed");
    assert!(server
        .app
        .pane_graphics
        .slots
        .remove(&graphics_key(pane_id))
        .is_some());
    assert!(!active.load(std::sync::atomic::Ordering::Acquire));
}

#[test]
fn stream_set_has_graphics_only_render_impact() {
    let mut server = test_headless_server();
    let workspace = crate::workspace::Workspace::test_new("graphics");
    let pane_id = workspace.tabs[0].root_pane;
    let public_pane_id = format!("{}:p1", workspace.id);
    server.app.state.workspaces = vec![workspace];
    server.app.state.active = Some(0);
    server.app.state.selected = 0;
    server.app.state.kitty_graphics_enabled = true;
    set_stream_owner(&mut server, pane_id, "owner-a");

    let (request, response_rx) =
        stream_set_message("wrong-owner", &public_pane_id, "owner-b", vec![1, 2, 3]);
    assert_eq!(
        server.handle_api_request_with_render_impact(request),
        RenderImpact::None
    );
    assert!(serde_json::from_str::<api::schema::ErrorResponse>(
        &response_rx
            .recv_timeout(Duration::from_millis(100))
            .unwrap()
    )
    .is_ok());

    let (request, response_rx) =
        stream_set_message("stream-frame", &public_pane_id, "owner-a", vec![1, 2, 3]);
    assert_eq!(
        server.handle_api_request_with_render_impact(request),
        RenderImpact::Graphics
    );
    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
        &response_rx
            .recv_timeout(Duration::from_millis(100))
            .unwrap()
    )
    .is_ok());

    server
        .app
        .event_tx
        .try_send(AppEvent::UpdateReady {
            version: "9.9.9".into(),
            install_command: "herdr update".into(),
        })
        .unwrap();
    let (request, _response_rx) = stream_set_message(
        "stream-frame-with-internal-event",
        &public_pane_id,
        "owner-a",
        vec![4, 5, 6],
    );
    assert_eq!(
        server.handle_api_request_with_render_impact(request),
        RenderImpact::Full
    );

    server.app.pane_graphics.clear();
    let (respond_to, _response_rx) = std::sync::mpsc::channel();
    let impact = server.handle_api_request_with_render_impact(api::ApiRequestMessage {
        request: api::schema::Request {
            id: "direct-frame".into(),
            method: api::schema::Method::PaneGraphicsSet(api::schema::PaneGraphicsSetParams {
                pane_id: public_pane_id,
                layer_id: None,
                z_index: 0,
                owner: String::new(),
                format: api::schema::PaneGraphicsFormat::Png,
                image_width: 1,
                image_height: 1,
                data: Some(vec![1, 2, 3]),
                data_base64: String::new(),
                placement: api::schema::PaneGraphicsPlacementParams::default(),
            }),
        },
        respond_to,
        response_write_complete: None,
        stream_active: None,
    });
    assert_eq!(impact, RenderImpact::Full);
}

#[test]
fn rejected_or_stale_requests_do_not_schedule_rendering() {
    let mut server = test_headless_server();
    let workspace = crate::workspace::Workspace::test_new("graphics");
    let pane_id = workspace.tabs[0].root_pane;
    let public_pane_id = format!("{}:p1", workspace.id);
    server.app.state.workspaces = vec![workspace];
    server.app.state.active = Some(0);
    server.app.state.selected = 0;
    server.app.state.kitty_graphics_enabled = false;

    let (respond_to, response_rx) = std::sync::mpsc::channel();
    let changed = server.handle_api_request_with_shutdown_check(api::ApiRequestMessage {
        request: api::schema::Request {
            id: "disabled-set".into(),
            method: api::schema::Method::PaneGraphicsSet(api::schema::PaneGraphicsSetParams {
                pane_id: public_pane_id.clone(),
                layer_id: None,
                z_index: 0,
                owner: String::new(),
                format: api::schema::PaneGraphicsFormat::Png,
                image_width: 1,
                image_height: 1,
                data: Some(vec![1, 2, 3]),
                data_base64: String::new(),
                placement: api::schema::PaneGraphicsPlacementParams::default(),
            }),
        },
        respond_to,
        response_write_complete: None,
        stream_active: None,
    });
    assert!(!changed);
    let response = response_rx
        .recv_timeout(Duration::from_millis(100))
        .unwrap();
    assert_eq!(
        serde_json::from_str::<api::schema::ErrorResponse>(&response)
            .unwrap()
            .error
            .code,
        "feature_disabled"
    );

    server.app.state.kitty_graphics_enabled = true;
    set_stream_owner(&mut server, pane_id, "current-owner");
    let (respond_to, response_rx) = std::sync::mpsc::channel();
    let impact = server.handle_api_request_with_render_impact(api::ApiRequestMessage {
        request: api::schema::Request {
            id: "stale-close".into(),
            method: api::schema::Method::PaneGraphicsStreamClose(
                api::schema::PaneGraphicsStreamParams {
                    pane_id: public_pane_id,
                    layer_id: None,
                    z_index: 0,
                    owner: "stale-owner".into(),
                },
            ),
        },
        respond_to,
        response_write_complete: None,
        stream_active: None,
    });
    assert_eq!(impact, RenderImpact::None);
    assert_eq!(
        server
            .app
            .pane_graphics
            .slots
            .get(&graphics_key(pane_id))
            .and_then(|slot| slot.stream_owner.as_deref()),
        Some("current-owner")
    );
    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
        &response_rx
            .recv_timeout(Duration::from_millis(100))
            .unwrap()
    )
    .is_ok());
}

#[cfg(unix)]
#[tokio::test]
async fn direct_graphics_availability_follows_foreground_client_with_background_clients() {
    let (mut server, _client_rx, _pane_id) = retained_test_server(b"direct eligibility");
    let foreground = server.clients.get_mut(&1).unwrap();
    foreground.direct_graphics = true;
    foreground.pixel_mouse = true;

    let (background_writer, _background_control_rx, _background_render_rx) = test_client_writer();
    server.clients.insert(
        2,
        ClientConnection::new(
            (80, 24),
            crate::kitty_graphics::HostCellSize::default(),
            2,
            RenderEncoding::SemanticFrame,
            Some(background_writer),
        ),
    );

    server.sync_foreground_client_state();
    assert!(server.direct_graphics_available());
    assert!(server.app.direct_graphics_available);

    server.foreground_client_id = Some(2);
    server.sync_foreground_client_state();
    assert!(!server.direct_graphics_available());
    assert!(!server.app.direct_graphics_available);
}

#[cfg(unix)]
#[test]
fn direct_graphics_routing_prefers_the_target_stream_owner() {
    let (mut server, first_key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    add_direct_client(&mut server, 7);
    add_direct_client(&mut server, 8);
    server.clients.get_mut(&8).unwrap().last_activity = 2;
    let (transfer_id, image_id) = direct_ids(&server, &first_key);
    assert!(server.complete_direct_graphics(7, transfer_id, image_id, true));
    assert_eq!(response_rx.recv().unwrap(), "ack");

    let second_key = (first_key.0, "second".into());
    let second_layer = crate::app::pane_graphics::Layer {
        format: crate::api::schema::PaneGraphicsFormat::Rgba,
        image_width: 1,
        image_height: 1,
        backing: crate::app::pane_graphics::Backing::Resident {
            len: 4,
            client_id: 8,
        },
        data_fingerprint: 2,
        render: Default::default(),
        z_index: 0,
    };
    server.app.pane_graphics.slots.insert(
        second_key.clone(),
        crate::app::pane_graphics::Slot::test((1 << 31) | 901, Some(second_layer)),
    );

    assert_eq!(server.direct_graphics_client_for_key(&first_key), Some(7));
    assert_eq!(server.direct_graphics_client_for_key(&second_key), Some(8));
    assert_eq!(
        server.direct_graphics_client_for_key(&(first_key.0, "new".into())),
        Some(8)
    );
}

#[cfg(unix)]
#[test]
fn resident_direct_stream_survives_non_direct_client_becoming_foreground() {
    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    add_direct_client(&mut server, 7);
    server.foreground_client_id = Some(7);
    server.sync_foreground_client_state();
    let (transfer_id, image_id) = direct_ids(&server, &key);
    assert!(server.complete_direct_graphics(7, transfer_id, image_id, true));
    assert_eq!(response_rx.recv().unwrap(), "ack");

    let (writer, _control_rx, _render_rx) = test_client_writer();
    server.clients.insert(
        8,
        ClientConnection::new(
            (80, 24),
            crate::kitty_graphics::HostCellSize::default(),
            2,
            RenderEncoding::SemanticFrame,
            Some(writer),
        ),
    );
    server.foreground_client_id = Some(8);
    server.sync_foreground_client_state();

    assert!(server.direct_graphics_available());
    assert!(server.app.direct_graphics_available);
    assert_eq!(
        server.app.pane_graphics.slots[&key]
            .layer
            .as_ref()
            .and_then(crate::app::pane_graphics::Layer::resident_client),
        Some(7)
    );

    server.remove_client_and_resize_if_needed(7);
    assert!(!server.app.pane_graphics.slots.contains_key(&key));
    assert!(!server.direct_graphics_available());
    assert!(!server.app.direct_graphics_available);
}

#[cfg(unix)]
#[tokio::test]
async fn client_shell_direct_graphics_uploads_without_server_authored_coordinates() {
    let (mut server, client_rx, pane_id) = retained_test_server(b"client shell direct");
    server.app.state.kitty_graphics_enabled = true;
    let client = server.clients.get_mut(&1).unwrap();
    client.mode = ClientConnectionMode::ClientShell;
    client.render_state =
        crate::server::render_stream::ClientRenderState::new(RenderEncoding::SemanticFrame);
    client.cell_size = crate::kitty_graphics::HostCellSize {
        width_px: 10,
        height_px: 20,
    };
    client.direct_graphics = true;
    client.pixel_mouse = true;
    server.app.direct_graphics_available = true;
    let (background_writer, _background_control_rx, background_render_rx) = test_client_writer();
    server.clients.insert(
        2,
        ClientConnection::new(
            (80, 24),
            crate::kitty_graphics::HostCellSize {
                width_px: 10,
                height_px: 20,
            },
            2,
            RenderEncoding::SemanticFrame,
            Some(background_writer),
        ),
    );
    set_stream_owner(&mut server, pane_id, "browser");
    let public_pane_id = server.app.public_pane_id(0, pane_id).unwrap();
    let path = sparse_direct_frame(&server, "client-shell-direct.rgba", 1, 1);
    let (message, response_rx) =
        direct_stream_message("shell-direct", &public_pane_id, "browser", path, 1, 1, 1);

    assert_eq!(
        server.handle_pane_graphics_stream_frame(message),
        RenderImpact::None
    );
    let (transfer_id, image_id, asset) = match read_server_message(
        client_rx
            .recv_timeout(Duration::from_secs(1))
            .expect("client shell direct upload"),
    ) {
        ServerMessage::GraphicsFile {
            transfer_id,
            image_id,
            leading,
            control,
            surface_asset: Some(asset),
            ..
        } => {
            assert!(leading.is_empty());
            assert!(control.starts_with("a=t,"), "{control}");
            assert!(!control.contains("\u{1b}["), "{control}");
            (transfer_id, image_id, asset)
        }
        other => panic!("expected client shell graphics file, got {other:?}"),
    };
    assert!(background_render_rx.try_recv().is_err());
    assert_eq!(
        image_id,
        crate::kitty_graphics::surface::host_image_id(&server.client_shell_boot_id, &asset)
    );
    let (pending, _) = crate::server::client_shell_graphics::collect(
        &server.app,
        &[],
        &[],
        None,
        Some(crate::ui::TabSurfaceTarget {
            workspace_index: 0,
            tab_index: 0,
        }),
        crate::kitty_graphics::HostCellSize {
            width_px: 10,
            height_px: 20,
        },
        &crate::kitty_graphics::surface::DeliveryCache::default(),
        1,
    );
    assert_eq!(pending.retained_assets, vec![asset.clone()]);

    server.start_direct_graphics_response(1, transfer_id, image_id);
    assert!(server.complete_direct_graphics(1, transfer_id, image_id, true));
    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
        &response_rx.recv_timeout(Duration::from_secs(1)).unwrap()
    )
    .is_ok());

    server.foreground_client_id = Some(2);
    server.sync_foreground_client_state();
    let next_path = sparse_direct_frame(&server, "client-shell-direct-next.rgba", 1, 1);
    let (next_message, next_response_rx) = direct_stream_message(
        "shell-direct-next",
        &public_pane_id,
        "browser",
        next_path,
        1,
        1,
        2,
    );
    assert_eq!(
        server.handle_pane_graphics_stream_frame(next_message),
        RenderImpact::None
    );
    let (next_transfer_id, next_image_id, next_asset) = match read_server_message(
        client_rx
            .recv_timeout(Duration::from_secs(1))
            .expect("sticky owner direct upload"),
    ) {
        ServerMessage::GraphicsFile {
            transfer_id,
            image_id,
            surface_asset: Some(asset),
            ..
        } => (transfer_id, image_id, asset),
        other => panic!("expected sticky owner graphics file, got {other:?}"),
    };
    assert!(background_render_rx.try_recv().is_err());
    server.start_direct_graphics_response(1, next_transfer_id, next_image_id);
    assert!(server.complete_direct_graphics(1, next_transfer_id, next_image_id, true));
    assert!(serde_json::from_str::<api::schema::SuccessResponse>(
        &next_response_rx
            .recv_timeout(Duration::from_secs(1))
            .unwrap()
    )
    .is_ok());
    assert!(server.app.direct_graphics_available);

    server.clients.get_mut(&1).unwrap().request_repaint();
    server.render_and_stream();
    let surface = read_server_message(
        client_rx
            .recv_timeout(Duration::from_secs(1))
            .expect("resident client shell scene"),
    );
    let ServerMessage::PaneSurface(surface) = surface else {
        panic!("expected client shell pane surface");
    };
    assert_eq!(surface.graphics.placements.len(), 1);
    assert!(surface.graphics.assets.is_empty());
    assert_eq!(surface.graphics.placements[0].asset, next_asset);
    assert_eq!(surface.graphics.retained_assets, vec![next_asset.clone()]);

    let background_surface = read_server_message(
        background_render_rx
            .recv_timeout(Duration::from_secs(1))
            .expect("non-owning client shell scene"),
    );
    let ServerMessage::PaneSurface(background_surface) = background_surface else {
        panic!("expected non-owning client shell pane surface");
    };
    assert!(background_surface.graphics.assets.is_empty());
    assert!(background_surface.graphics.placements.is_empty());
    assert!(background_surface.graphics.retained_assets.is_empty());

    let (hidden, _) = crate::server::client_shell_graphics::collect(
        &server.app,
        &[],
        &[],
        None,
        Some(crate::ui::TabSurfaceTarget {
            workspace_index: 0,
            tab_index: 0,
        }),
        crate::kitty_graphics::HostCellSize {
            width_px: 10,
            height_px: 20,
        },
        &crate::kitty_graphics::surface::DeliveryCache::default(),
        1,
    );
    assert!(hidden.placements.is_empty());
    assert_eq!(hidden.retained_assets, vec![next_asset]);
}

#[cfg(unix)]
fn direct_gate_server(
    data: &[u8],
) -> (
    HeadlessServer,
    crate::app::pane_graphics::Key,
    std::sync::mpsc::Receiver<String>,
) {
    direct_gate_server_with_file(data.len(), Some(data))
}

#[cfg(unix)]
fn direct_gate_server_with_file(
    len: usize,
    data: Option<&[u8]>,
) -> (
    HeadlessServer,
    crate::app::pane_graphics::Key,
    std::sync::mpsc::Receiver<String>,
) {
    use std::io::Write as _;
    use std::os::unix::fs::OpenOptionsExt as _;
    let mut server = test_headless_server();
    let workspace = crate::workspace::Workspace::test_new("direct-gate");
    let pane_id = workspace.tabs[0].root_pane;
    server.app.state.workspaces = vec![workspace];
    server.app.state.active = Some(0);
    let key = graphics_key(pane_id);
    let path = server
        .app
        .pane_graphics_files
        .source_directory()
        .unwrap()
        .join("gate-frame");
    let mut file = std::fs::OpenOptions::new()
        .write(true)
        .create_new(true)
        .mode(0o600)
        .open(&path)
        .unwrap();
    if let Some(data) = data {
        file.write_all(data).unwrap();
    } else {
        file.set_len(len as u64).unwrap();
    }
    drop(file);
    let lease = server.app.pane_graphics_files.lease(&path, len).unwrap();
    let (respond_to, response_rx) = std::sync::mpsc::channel();
    let layer =
        crate::app::pane_graphics::Layer::direct(1, 1, lease.clone(), Default::default(), 0);
    let mut slot = crate::app::pane_graphics::Slot::test((1 << 31) | 900, Some(layer));
    slot.stream_owner = Some("owner".into());
    slot.stream_active = Some(active_gate());
    slot.direct_gate = Some(crate::app::pane_graphics::DirectGate {
        transfer_id: lease.fingerprint(),
        image_id: (1 << 31) | 900,
        client_id: 7,
        deadline: std::time::Instant::now() + Duration::from_secs(1),
        written: true,
        success_response: "ack".into(),
        respond_to,
    });
    server.app.pane_graphics.slots.insert(key.clone(), slot);
    (server, key, response_rx)
}

#[cfg(unix)]
fn direct_ids(server: &HeadlessServer, key: &crate::app::pane_graphics::Key) -> (u64, u32) {
    let slot = &server.app.pane_graphics.slots[key];
    let gate = slot.direct_gate.as_ref().unwrap();
    (gate.transfer_id, gate.image_id)
}

#[cfg(unix)]
fn add_direct_client(server: &mut HeadlessServer, client_id: u64) {
    let (writer, control_rx, render_rx) = test_client_writer();
    std::mem::forget((control_rx, render_rx));
    let mut client = ClientConnection::new(
        (80, 24),
        crate::kitty_graphics::HostCellSize {
            width_px: 10,
            height_px: 20,
        },
        1,
        RenderEncoding::SemanticFrame,
        Some(writer),
    );
    client.direct_graphics = true;
    client.pixel_mouse = true;
    server.clients.insert(client_id, client);
}

#[cfg(unix)]
#[test]
fn terminal_response_deadline_starts_only_after_client_flush() {
    let (mut server, key, _response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    let slot = server.app.pane_graphics.slots.get_mut(&key).unwrap();
    let gate = slot.direct_gate.as_mut().unwrap();
    gate.written = false;
    let (transfer_id, image_id) = (gate.transfer_id, slot.host_image_id);
    assert!(!server.complete_direct_graphics(7, transfer_id, image_id, true));
    assert!(!server.start_direct_graphics_response(7, transfer_id, image_id));
    let gate = server.app.pane_graphics.slots[&key]
        .direct_gate
        .as_ref()
        .unwrap();
    assert!(gate.written && gate.deadline > std::time::Instant::now());
}

#[cfg(unix)]
#[test]
fn outer_timeout_covers_both_direct_phases_and_cancellation_blocks_late_results() {
    assert!(
        crate::app::pane_graphics::DIRECT_OUTER_TIMEOUT
            > crate::app::pane_graphics::DIRECT_DELIVERY_TIMEOUT
                + crate::app::pane_graphics::DIRECT_RESPONSE_TIMEOUT
    );
    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    let slot = server.app.pane_graphics.slots.get_mut(&key).unwrap();
    slot.stream_active
        .as_ref()
        .unwrap()
        .store(false, std::sync::atomic::Ordering::Release);
    let (transfer_id, image_id) = (
        slot.direct_gate.as_ref().unwrap().transfer_id,
        slot.host_image_id,
    );

    assert!(!server.complete_direct_graphics(7, transfer_id, image_id, true));
    assert!(response_rx.try_recv().is_err());
    assert!(server.app.pane_graphics.slots[&key]
        .layer
        .as_ref()
        .unwrap()
        .direct_lease()
        .is_some());
}

#[cfg(unix)]
#[test]
fn matching_terminal_ok_releases_producer_and_acknowledges() {
    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    let (transfer_id, image_id) = direct_ids(&server, &key);

    assert!(server.complete_direct_graphics(7, transfer_id, image_id, true));

    assert_eq!(response_rx.recv().unwrap(), "ack");
    let layer = server.app.pane_graphics.slots[&key].layer.as_ref().unwrap();
    assert!(layer.terminal_only());
    assert!(layer.direct_lease().is_none());
}

#[cfg(unix)]
#[test]
fn unwritten_direct_full_falls_back_without_stickiness_but_disconnect_retires() {
    for error in [
        std::sync::mpsc::TrySendError::Full(Vec::new()),
        std::sync::mpsc::TrySendError::Disconnected(Vec::new()),
    ] {
        let should_ack = matches!(error, std::sync::mpsc::TrySendError::Full(_));
        let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
        add_direct_client(&mut server, 7);
        let gate = server
            .app
            .pane_graphics
            .slots
            .get_mut(&key)
            .and_then(|slot| slot.direct_gate.take())
            .unwrap();
        let result = server.handle_unwritten_direct_failure(
            &key,
            gate.success_response,
            gate.respond_to,
            error,
        );
        let inline = server
            .app
            .pane_graphics
            .slots
            .get(&key)
            .and_then(|slot| slot.layer.as_ref()?.inline_data())
            .is_some();
        assert_eq!(
            (
                result,
                response_rx.try_recv().ok().as_deref() == Some("ack"),
                inline,
                server.clients[&7].direct_graphics,
            ),
            (should_ack, should_ack, should_ack, true)
        );
    }
}

#[cfg(unix)]
#[test]
fn eligibility_loss_cancels_the_queued_direct_upload() {
    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    let (writer, control_rx, _render_rx) = test_client_writer();
    let mut client = ClientConnection::new(
        (80, 24),
        crate::kitty_graphics::HostCellSize {
            width_px: 10,
            height_px: 20,
        },
        1,
        RenderEncoding::SemanticFrame,
        Some(writer),
    );
    client.direct_graphics = true;
    client.pixel_mouse = true;
    server.clients.insert(7, client);
    let (transfer_id, image_id) = direct_ids(&server, &key);

    server.retire_all_direct_graphics();

    assert!(matches!(
        read_server_message(
            control_rx
                .recv_timeout(Duration::from_millis(100))
                .expect("direct retirement")
        ),
        ServerMessage::GraphicsTransmissionRetired {
            transfer_id: retired_transfer,
            image_id: retired_image,
        } if retired_transfer == transfer_id && retired_image == image_id
    ));
    assert!(!server.app.pane_graphics.slots.contains_key(&key));
    assert!(response_rx.recv().is_err());
}

#[cfg(unix)]
#[test]
fn client_loss_retires_only_its_direct_stream() {
    let (mut pending, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    pending.retire_direct_graphics_for_client(8);
    assert!(pending.app.pane_graphics.slots.contains_key(&key));
    pending.retire_direct_graphics_for_client(7);
    assert!(!pending.app.pane_graphics.slots.contains_key(&key));
    assert!(response_rx.recv().is_err());

    let (mut resident, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    let slot = resident.app.pane_graphics.slots.get(&key).unwrap();
    assert!(resident.complete_direct_graphics(
        7,
        slot.direct_gate.as_ref().unwrap().transfer_id,
        slot.host_image_id,
        true,
    ));
    assert_eq!(response_rx.recv().unwrap(), "ack");
    resident.retire_direct_graphics_for_client(8);
    assert!(resident.app.pane_graphics.slots.contains_key(&key));
    resident.retire_direct_graphics_for_client(7);
    assert!(!resident.app.pane_graphics.slots.contains_key(&key));
}

#[cfg(unix)]
#[test]
fn pane_removal_cancels_the_pending_client_upload() {
    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    let (writer, control_rx, _render_rx) = test_client_writer();
    let mut client = ClientConnection::new(
        (80, 24),
        crate::kitty_graphics::HostCellSize {
            width_px: 10,
            height_px: 20,
        },
        1,
        RenderEncoding::SemanticFrame,
        Some(writer),
    );
    client.direct_graphics = true;
    client.pixel_mouse = true;
    server.clients.insert(7, client);
    let (transfer_id, image_id) = direct_ids(&server, &key);
    server.app.state.workspaces.clear();

    assert!(server.retain_live_pane_graphics());

    assert!(matches!(
        read_server_message(
            control_rx
                .recv_timeout(Duration::from_millis(100))
                .expect("pane removal retirement")
        ),
        ServerMessage::GraphicsTransmissionRetired {
            transfer_id: retired_transfer,
            image_id: retired_image,
        } if retired_transfer == transfer_id && retired_image == image_id
    ));
    assert!(!server.app.pane_graphics.slots.contains_key(&key));
    assert!(response_rx.recv().is_err());
}

#[cfg(unix)]
#[test]
fn pane_removal_and_shutdown_drop_direct_without_ack() {
    let setups: [fn(&mut HeadlessServer); 2] = [
        |server| server.app.state.workspaces.clear(),
        |server| server.shutting_down = true,
    ];
    for setup in setups {
        let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
        let (transfer_id, image_id) = direct_ids(&server, &key);
        setup(&mut server);
        assert!(!server.complete_direct_graphics(7, transfer_id, image_id, true));
        assert!(response_rx.recv().is_err());
        assert!(!server.app.pane_graphics.slots.contains_key(&key));
    }
}

#[cfg(unix)]
#[test]
fn timeout_retires_stream_without_producer_ack() {
    let (mut server, key, response_rx) = direct_gate_server(&[1, 2, 3, 4]);
    add_direct_client(&mut server, 7);
    server
        .app
        .pane_graphics
        .slots
        .get_mut(&key)
        .unwrap()
        .direct_gate
        .as_mut()
        .unwrap()
        .deadline = std::time::Instant::now() - Duration::from_millis(1);

    assert!(server.expire_direct_graphics(std::time::Instant::now()));

    assert!(response_rx.recv().is_err());
    assert!(!server.app.pane_graphics.slots.contains_key(&key));
    assert!(!server.clients[&7].direct_graphics);
    assert!(server.clients[&7].pixel_mouse);
}