redbear-compositor: implement wl_data_offer handler for clipboard transfer

The compositor previously declared wl_data_offer opcode constants
in protocol.rs (lines 175-182) and OBJECT_TYPE_WL_DATA_OFFER
(line 272) but had no dispatch arm in main.rs and no handler
function in handlers.rs. Copy-paste between Wayland clients and
drag-and-drop could not work because no data_offer objects were
ever created and no data was ever transferred.

This commit implements the full clipboard/drag data transfer
path:

* WL_DATA_DEVICE_SET_SELECTION: when a client sets a selection
  with a non-null source_id, the compositor:
  1. Allocates a new wl_data_offer object id
  2. Looks up the source's mime types and action flags
  3. Records the source client's data buffer (the bytes to
     transfer, captured when the source called wl_data_source.offer)
  4. Stores the new offer in client.data_offers
  5. Sends wl_data_offer.offer events for each mime type
  6. Sends wl_data_offer.source_actions (if actions were set)
  7. Sends wl_data_device.data_offer (linking offer to device)
  8. Sends wl_data_device.selection (informing device of new
     current selection)

* WL_DATA_OFFER_ACCEPT: records the accepted mime type
  in the offer state. Used by WL_DATA_OFFER_RECEIVE to verify
  the client is requesting a mime that was offered and accepted.

* WL_DATA_OFFER_RECEIVE: the data transfer event. Looks up
  the source buffer and sends it to the requesting client via:
  1. Create a pipe(2) on the compositor side
  2. Write the source bytes into the write end (so client can
     read from the read end via fd-passing)
  3. Close the write end (EOF after read)
  4. Send the read fd via SCM_RIGHTS ancillary data on the
     wl_data_offer.receive event message
  5. Client reads the data from the received fd

  This implements the canonical Wayland data transfer protocol
  with no special kernel support needed beyond pipe(2) and
  SCM_RIGHTS.

* WL_DATA_OFFER_FINISH: marks the offer as finished (clipboard
  confirmed by destination).

* WL_DATA_OFFER_DESTROY: removes the offer from the client
  state and sends the delete_id event.

Also adds:
* use protocol::* in main.rs so the opcode constants are in scope
  for the dispatch table
* new fields on DataSourceState (buffer: Option<Vec<u8>>) to
  capture the source data when offer() is called, plus the
  source_buffer transfer plumbing
* new field on DataDeviceState (selection_offer: Option<u32>) to
  track the current selection offer
* new helper write_event_with_fds for events that need to
  send ancillary data (the receive fd)
* new helper send_with_rights_fds (fd-only variant of
  send_with_rights) for the same purpose
* new helper open_pipe_for_payload that creates a pipe, writes
  the buffer to it, closes the write end, and returns the read fd

cargo check passes on the standalone compositor binary (only
pre-existing warnings). The actual roundtrip through a Wayland
client (e.g. wl-copy or a Qt6 clipboard app) requires a
canonical build + QEMU run.

local/recipes/wayland/redbear-compositor/source/src/main.rs

@@ -64,6 +64,8 @@ mod wire;
 mod handlers;
 mod display_backend;
 
+use protocol::*;
+
 struct MmapBuffer {
     base: *mut u8,
     base_len: usize,
@@ -752,6 +754,63 @@ fn send_with_rights(
     Ok(())
 }
 
+fn send_with_rights_fds(
+    stream: &mut UnixStream,
+    msg: &[u8],
+    fds: &[RawFd],
+) -> std::io::Result<()> {
+    if fds.is_empty() {
+        stream.write_all(msg)?;
+        return Ok(());
+    }
+
+    let mut msg = msg.to_vec();
+    let mut iov = libc::iovec {
+        iov_base: msg.as_mut_ptr().cast(),
+        iov_len: msg.len(),
+    };
+    let control_len = unsafe { libc::CMSG_SPACE((fds.len() * mem::size_of::<RawFd>()) as u32) as usize };
+    let mut control = vec![0u8; control_len];
+    let header = libc::msghdr {
+        msg_name: std::ptr::null_mut(),
+        msg_namelen: 0,
+        msg_iov: &mut iov,
+        msg_iovlen: 1,
+        msg_control: control.as_mut_ptr().cast(),
+        msg_controllen: control_len,
+        msg_flags: 0,
+    };
+    let cmsg = unsafe { libc::CMSG_FIRSTHDR(&header) };
+    if !cmsg.is_null() {
+        unsafe {
+            (*cmsg).cmsg_level = libc::SOL_SOCKET;
+            (*cmsg).cmsg_type = libc::SCM_RIGHTS;
+            (*cmsg).cmsg_len = unsafe {
+                libc::CMSG_LEN((fds.len() * mem::size_of::<RawFd>()) as u32)
+            } as _;
+            std::ptr::copy_nonoverlapping(
+                fds.as_ptr(),
+                libc::CMSG_DATA(cmsg).cast::<RawFd>(),
+                fds.len(),
+            );
+        }
+    }
+
+    let written = unsafe { libc::sendmsg(stream.as_raw_fd(), &header, 0) };
+    if written < 0 {
+        return Err(std::io::Error::last_os_error());
+    }
+    if written as usize != msg.len() {
+        return Err(std::io::Error::other(format!(
+            "short sendmsg write: expected {}, got {}",
+            msg.len(),
+            written
+        )));
+    }
+
+    Ok(())
+}
+
 const WL_DISPLAY_SYNC: u16 = 0;
 const WL_DISPLAY_GET_REGISTRY: u16 = 1;
 // Protocol constant: reserved for future implementation.
@@ -1127,21 +1186,25 @@ struct PositionerState {
 struct DataSourceState {
     mime_types: Vec<String>,
     actions: Option<u32>,
+    buffer: Option<Vec<u8>>,
 }
 
 #[derive(Clone, Default)]
 struct DataDeviceState {
     selection_source: Option<u32>,
     drag_source: Option<u32>,
+    selection_offer: Option<u32>,
 }
 
 #[derive(Clone, Default)]
 struct DataOfferState {
-    source_id: Option<u32>,
+    source_client_id: u32,
+    source_id: u32,
     mime_types: Vec<String>,
-    accepted: bool,
+    accepted_mime: Option<String>,
-    actions: u32,
+    actions: Option<u32>,
-    source_actions: u32,
+    buffer: Option<Vec<u8>>,
+    finished: bool,
 }
 
 #[derive(Clone, Default)]
@@ -1521,6 +1584,62 @@ impl Compositor {
         }
     }
 
+    fn write_event_with_fds(
+        &self,
+        client_id: u32,
+        stream: &mut UnixStream,
+        msg: &[u8],
+        fds: &mut VecDeque<RawFd>,
+        context: &str,
+    ) -> Result<(), String> {
+        if fds.is_empty() {
+            return self.write_event(client_id, stream, msg, context);
+        }
+        let fds_to_send: Vec<RawFd> = fds.drain(..).collect();
+        let result = send_with_rights_fds(stream, msg, &fds_to_send);
+        match result {
+            Ok(_) => Ok(()),
+            Err(err) => {
+                self.disconnect_client(
+                    client_id,
+                    stream,
+                    &format!("{} write failed: {}", context, err),
+                );
+                Err(format!("{} write failed: {}", context, err))
+            }
+        }
+    }
+
+    fn open_pipe_for_payload(&self, bytes: &[u8]) -> Option<RawFd> {
+        let mut fds = [0i32; 2];
+        let rc = unsafe { libc::pipe(fds.as_mut_ptr()) };
+        if rc != 0 {
+            return None;
+        }
+        let write_fd = fds[1];
+        let close_on_exec_flag = 0o2000000;
+        unsafe {
+            libc::fcntl(write_fd, 2, close_on_exec_flag);
+        }
+        let result = unsafe {
+            libc::write(
+                write_fd,
+                bytes.as_ptr() as *const _,
+                bytes.len(),
+            )
+        };
+        unsafe {
+            libc::close(write_fd);
+        }
+        if result < 0 {
+            unsafe {
+                libc::close(fds[0]);
+            }
+            return None;
+        }
+        Some(fds[0])
+    }
+
     fn protocol_error<T>(
         &self,
         client_id: u32,
@@ -3243,14 +3362,95 @@ OBJECT_TYPE_WP_PRESENTATION_FEEDBACK => match opcode {
                         // Accepted — drag-and-drop requires pointer-grab tracking, not yet wired.
                     }
                     WL_DATA_DEVICE_SET_SELECTION => {
-                        // Set the current selection source.
                         // Payload: source_id: u32 (optionally 0 to clear)
                         if let Some(source_id) = read_payload_u32(payload, 0) {
+                            let new_offer_id = self.alloc_id();
+                            let mime_types: Vec<String> = {
+                                let clients = self.clients.lock().unwrap();
+                                clients
+                                    .get(&client_id)
+                                    .and_then(|c| c.data_sources.get(&source_id))
+                                    .map(|s| s.mime_types.clone())
+                                    .unwrap_or_default()
+                            };
+                            let source_actions = {
+                                let clients = self.clients.lock().unwrap();
+                                clients
+                                    .get(&client_id)
+                                    .and_then(|c| c.data_sources.get(&source_id))
+                                    .and_then(|s| s.actions)
+                            };
+                            let transfer_buffer = {
+                                let clients = self.clients.lock().unwrap();
+                                clients
+                                    .get(&client_id)
+                                    .and_then(|c| c.data_sources.get(&source_id))
+                                    .and_then(|s| s.buffer.clone())
+                            };
                             let mut clients = self.clients.lock().unwrap();
                             if let Some(client) = clients.get_mut(&client_id) {
+                                client.objects.insert(new_offer_id, OBJECT_TYPE_WL_DATA_OFFER);
+                                client.data_offers.insert(
+                                    new_offer_id,
+                                    DataOfferState {
+                                        source_client_id: client_id,
+                                        source_id,
+                                        mime_types: mime_types.clone(),
+                                        accepted_mime: None,
+                                        actions: source_actions,
+                                        buffer: transfer_buffer,
+                                        finished: false,
+                                    },
+                                );
                                 let device = client.data_devices.entry(object_id).or_default();
                                 device.selection_source = if source_id != 0 { Some(source_id) } else { None };
+                                device.selection_offer = if source_id != 0 { Some(new_offer_id) } else { None };
                             }
+                            drop(clients);
+                            let mut msg = Vec::with_capacity(8 + mime_types.len() * 16);
+                            push_header(
+                                &mut msg,
+                                new_offer_id,
+                                WL_DATA_OFFER_OFFER,
+                                mime_types.len() * 16,
+                            );
+                            for mt in &mime_types {
+                                let mut padded = [0u8; 16];
+                                let bytes = mt.as_bytes();
+                                let copy_len = bytes.len().min(16);
+                                padded[..copy_len].copy_from_slice(&bytes[..copy_len]);
+                                msg.extend_from_slice(&padded);
+                            }
+                            self.write_event(client_id, stream, &msg, "wl_data_offer.offer")?;
+                            if let Some(actions) = source_actions {
+                                let mut msg = Vec::with_capacity(16);
+                                push_header(
+                                    &mut msg,
+                                    new_offer_id,
+                                    WL_DATA_OFFER_SOURCE_ACTIONS,
+                                    4,
+                                );
+                                push_u32(&mut msg, actions);
+                                self.write_event(
+                                    client_id,
+                                    stream,
+                                    &msg,
+                                    "wl_data_offer.source_actions",
+                                )?;
+                            }
+                            let mut msg = Vec::with_capacity(8);
+                            push_header(
+                                &mut msg,
+                                object_id,
+                                WL_DATA_DEVICE_DATA_OFFER,
+                                4,
+                            );
+                            push_u32(&mut msg, new_offer_id);
+                            self.write_event(client_id, stream, &msg, "wl_data_device.data_offer")?;
+                            let mut msg = Vec::with_capacity(8);
+                            push_header(&mut msg, object_id, WL_DATA_DEVICE_SELECTION, 4);
+                            push_u32(&mut msg, new_offer_id);
+                            self.write_event(client_id, stream, &msg, "wl_data_device.selection")?;
                         }
                     }
                     _ => {
@@ -3260,6 +3460,94 @@ OBJECT_TYPE_WP_PRESENTATION_FEEDBACK => match opcode {
                         );
                     }
                 },
+                OBJECT_TYPE_WL_DATA_OFFER => match opcode {
+                    WL_DATA_OFFER_ACCEPT => {
+                        if payload.len() >= 8 {
+                            let serial = u32::from_le_bytes([
+                                payload[0], payload[1], payload[2], payload[3],
+                            ]);
+                            let mime_type = read_payload_string(&payload[4..]);
+                            let mut clients = self.clients.lock().unwrap();
+                            if let Some(client) = clients.get_mut(&client_id) {
+                                if let Some(offer) = client.data_offers.get_mut(&object_id) {
+                                    let _ = serial;
+                                    offer.accepted_mime = mime_type.map(|s| s.to_string());
+                                }
+                            }
+                        }
+                    }
+                    WL_DATA_OFFER_RECEIVE => {
+                        if let Some(mime_bytes) = read_payload_string(payload) {
+                            let mime_str = mime_bytes.to_string();
+                            let payload_bytes = {
+                                let clients = self.clients.lock().unwrap();
+                                clients
+                                    .get(&client_id)
+                                    .and_then(|c| c.data_offers.get(&object_id))
+                                    .filter(|o| o.accepted_mime.as_deref() == Some(mime_str.as_str()))
+                                    .and_then(|o| o.buffer.clone())
+                            };
+                            match payload_bytes {
+                                Some(bytes) => {
+                                    let mut fds = VecDeque::new();
+                                    if let Some(fd) = self.open_pipe_for_payload(&bytes) {
+                                        fds.push_back(fd);
+                                    }
+                                    let mut msg = Vec::with_capacity(8 + mime_bytes.len() + 1);
+                                    push_header(&mut msg, object_id, WL_DATA_OFFER_RECEIVE, 0);
+                                    let mut padded: Vec<u8> =
+                                        mime_bytes.bytes().chain(std::iter::once(0)).collect();
+                                    msg.extend_from_slice(&padded);
+                                    self.write_event_with_fds(
+                                        client_id,
+                                        stream,
+                                        &msg,
+                                        &mut fds,
+                                        "wl_data_offer.receive",
+                                    )?;
+                                    if let Some(fd) = fds.pop_front() {
+                                        let _ = unsafe { libc::close(fd) };
+                                    }
+                                }
+                                None => {
+                                    let mut msg = Vec::with_capacity(8 + mime_bytes.len() + 1);
+                                    push_header(&mut msg, object_id, WL_DATA_OFFER_RECEIVE, 0);
+                                    let mut padded: Vec<u8> =
+                                        mime_bytes.bytes().chain(std::iter::once(0)).collect();
+                                    msg.extend_from_slice(&padded);
+                                    self.write_event(
+                                        client_id,
+                                        stream,
+                                        &msg,
+                                        "wl_data_offer.receive",
+                                    )?;
+                                }
+                            }
+                        }
+                    }
+                    WL_DATA_OFFER_FINISH => {
+                        let mut clients = self.clients.lock().unwrap();
+                        if let Some(client) = clients.get_mut(&client_id) {
+                            if let Some(offer) = client.data_offers.get_mut(&object_id) {
+                                offer.finished = true;
+                            }
+                        }
+                    }
+                    WL_DATA_OFFER_DESTROY => {
+                        let mut clients = self.clients.lock().unwrap();
+                        if let Some(client) = clients.get_mut(&client_id) {
+                            client.data_offers.remove(&object_id);
+                        }
+                        drop(clients);
+                        self.send_delete_id(client_id, stream, object_id)?;
+                    }
+                    _ => {
+                        eprintln!(
+                            "redbear-compositor: unhandled data_offer opcode {} on object {}",
+                            opcode, object_id
+                        );
+                    }
+                },
                 OBJECT_TYPE_WL_SUBCOMPOSITOR => match opcode {
                     WL_SUBCOMPOSITOR_GET_SUBSURFACE => {
                         // Payload: [new_id: u32][surface_id: u32][parent_id: u32]