1 files changed, 446 insertions, 0 deletions
diff --git a/src/server/handoff.rs b/src/server/handoff.rs
new file mode 100644
index 00000000..d38bdc29
--- /dev/null
+++ b/src/server/handoff.rs
@@ -0,0 +1,446 @@
+#[cfg(unix)]
+use std::io::{self, Read, Write};
+#[cfg(unix)]
+use std::os::fd::{AsRawFd, RawFd};
+#[cfg(unix)]
+use std::os::unix::net::{UnixListener, UnixStream};
+#[cfg(unix)]
+use std::os::unix::process::CommandExt;
+#[cfg(unix)]
+use std::path::{Path, PathBuf};
+#[cfg(unix)]
+use std::process::Command;
+#[cfg(unix)]
+use std::time::Duration;
+
+#[cfg(unix)]
+use serde::{Deserialize, Serialize};
+#[cfg(unix)]
+use tracing::{info, warn};
+
+#[cfg(unix)]
+const HANDOFF_VERSION: u32 = 1;
+#[cfg(unix)]
+const READY_TIMEOUT: Duration = Duration::from_secs(30);
+#[cfg(unix)]
+pub(crate) const MAX_FDS_PER_HANDOFF: usize = 64;
+#[cfg(unix)]
+pub(crate) const MAX_REPLAY_BYTES_PER_PANE: usize = 8 * 1024;
+#[cfg(unix)]
+pub(crate) const COMMIT_TIMEOUT: Duration = READY_TIMEOUT;
+
+#[cfg(unix)]
+#[derive(Serialize, Deserialize)]
+pub(crate) struct HandoffManifest {
+    pub version: u32,
+    pub source_version: String,
+    pub source_protocol: u32,
+    pub expected_version: Option<String>,
+    pub expected_protocol: Option<u32>,
+    pub snapshot: crate::persist::SessionSnapshot,
+    pub panes: Vec<HandoffPane>,
+}
+
+#[cfg(unix)]
+#[derive(Debug, Serialize, Deserialize)]
+pub(crate) struct HandoffPane {
+    pub pane_id: u32,
+    pub child_pid: u32,
+    pub rows: u16,
+    pub cols: u16,
+    pub cell_width_px: u32,
+    pub cell_height_px: u32,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub initial_history_ansi: Option<String>,
+}
+
+#[cfg(unix)]
+pub(crate) struct ReceivedHandoff {
+    pub manifest: HandoffManifest,
+    pub fds: Vec<RawFd>,
+    pub stream: UnixStream,
+}
+
+#[cfg(unix)]
+pub(crate) fn handoff_socket_path() -> PathBuf {
+    crate::session::data_dir().join(format!("herdr-handoff-{}.sock", std::process::id()))
+}
+
+#[cfg(unix)]
+pub(crate) fn spawn_handoff_import(
+    import_exe: Option<&Path>,
+    socket_path: &Path,
+    token: &str,
+) -> io::Result<u32> {
+    let fallback_exe;
+    let exe = if let Some(import_exe) = import_exe {
+        import_exe
+    } else {
+        fallback_exe = std::env::current_exe().map_err(|err| {
+            io::Error::new(
+                err.kind(),
+                format!("failed to determine herdr executable path: {err}"),
+            )
+        })?;
+        &fallback_exe
+    };
+    let mut command = Command::new(exe);
+    command
+        .arg("server")
+        .arg("--handoff-import")
+        .arg(socket_path)
+        .arg(token)
+        .process_group(0)
+        .stdin(std::process::Stdio::null())
+        .stdout(std::process::Stdio::null())
+        .stderr(std::process::Stdio::null());
+    let child = command.spawn().map_err(|err| {
+        io::Error::new(
+            err.kind(),
+            format!(
+                "failed to spawn handoff import server at {}: {err}",
+                exe.display()
+            ),
+        )
+    })?;
+    Ok(child.id())
+}
+
+#[cfg(unix)]
+pub(crate) fn bind_listener(socket_path: &Path) -> io::Result<UnixListener> {
+    let _ = std::fs::remove_file(socket_path);
+    let listener = UnixListener::bind(socket_path)?;
+    listener.set_nonblocking(true)?;
+    restrict_socket_permissions(socket_path)?;
+    Ok(listener)
+}
+
+#[cfg(unix)]
+pub(crate) fn accept_and_validate_on(
+    listener: UnixListener,
+    socket_path: &Path,
+    token: &str,
+    manifest: &HandoffManifest,
+) -> io::Result<UnixStream> {
+    let (mut stream, _) = accept_with_timeout(&listener, READY_TIMEOUT)?;
+    stream.set_nonblocking(false)?;
+    stream.set_read_timeout(Some(READY_TIMEOUT))?;
+    stream.set_write_timeout(Some(READY_TIMEOUT))?;
+    let token_line = read_line_unbuffered(&mut stream)?;
+    if token_line.trim_end() != token {
+        return Err(io::Error::new(
+            io::ErrorKind::PermissionDenied,
+            "handoff import token mismatch",
+        ));
+    }
+
+    serde_json::to_writer(&mut stream, manifest).map_err(io::Error::other)?;
+    stream.write_all(b"\n")?;
+    stream.flush()?;
+
+    stream.set_read_timeout(Some(READY_TIMEOUT))?;
+    let validated = read_line_unbuffered(&mut stream)?;
+    if validated.trim_end() != "validated" {
+        return Err(io::Error::other("handoff import did not validate manifest"));
+    }
+    let _ = std::fs::remove_file(socket_path);
+    Ok(stream)
+}
+
+#[cfg(unix)]
+pub(crate) fn send_fds_and_wait_restored(stream: &mut UnixStream, fds: &[RawFd]) -> io::Result<()> {
+    if fds.len() > MAX_FDS_PER_HANDOFF {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!("handoff supports at most {MAX_FDS_PER_HANDOFF} pane file descriptors at once"),
+        ));
+    }
+    send_fds(stream, fds)?;
+
+    stream.set_read_timeout(Some(READY_TIMEOUT))?;
+    let restored = read_line_unbuffered(&mut *stream)?;
+    if restored.trim_end() != "restored" {
+        return Err(io::Error::other(
+            "handoff import did not report restored runtimes",
+        ));
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+pub(crate) fn wait_ready(stream: &mut UnixStream) -> io::Result<()> {
+    stream.set_read_timeout(Some(READY_TIMEOUT))?;
+    let ready = read_line_unbuffered(&mut *stream)?;
+    if ready.trim_end() != "ready" {
+        return Err(io::Error::other("handoff import did not report ready"));
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+pub(crate) fn report_committed(stream: &mut UnixStream) -> io::Result<()> {
+    stream.write_all(b"committed\n")?;
+    stream.flush()
+}
+
+#[cfg(unix)]
+pub(crate) fn wait_owned_ack(stream: &mut UnixStream) {
+    if let Err(err) = stream.set_read_timeout(Some(READY_TIMEOUT)) {
+        warn!(err = %err, "failed to set handoff ownership ack timeout");
+        return;
+    }
+    match read_line_unbuffered(&mut *stream) {
+        Ok(owned) if owned.trim_end() == "owned" => {}
+        Ok(other) => {
+            warn!(
+                response = %other.trim_end(),
+                "handoff import sent unexpected ownership ack after commit"
+            );
+        }
+        Err(err) => {
+            warn!(err = %err, "handoff import ownership ack was not received after commit");
+        }
+    }
+}
+
+#[cfg(unix)]
+pub(crate) fn receive(socket_path: &Path, token: &str) -> io::Result<ReceivedHandoff> {
+    let mut stream = UnixStream::connect(socket_path)?;
+    stream.write_all(token.as_bytes())?;
+    stream.write_all(b"\n")?;
+    stream.flush()?;
+
+    let manifest_line = read_line_unbuffered(&mut stream)?;
+    let manifest: HandoffManifest =
+        serde_json::from_str(&manifest_line).map_err(io::Error::other)?;
+    if manifest.version != HANDOFF_VERSION {
+        return Err(io::Error::other(format!(
+            "unsupported handoff version {}",
+            manifest.version
+        )));
+    }
+    if manifest
+        .expected_protocol
+        .is_some_and(|protocol| protocol != crate::protocol::PROTOCOL_VERSION)
+    {
+        return Err(io::Error::other(format!(
+            "handoff expected protocol {}, but this server speaks protocol {}",
+            manifest.expected_protocol.unwrap_or_default(),
+            crate::protocol::PROTOCOL_VERSION
+        )));
+    }
+    if manifest
+        .expected_version
+        .as_deref()
+        .is_some_and(|version| version != env!("CARGO_PKG_VERSION"))
+    {
+        return Err(io::Error::other(format!(
+            "handoff expected herdr v{}, but this server is v{}",
+            manifest.expected_version.as_deref().unwrap_or("unknown"),
+            env!("CARGO_PKG_VERSION")
+        )));
+    }
+    stream.write_all(b"validated\n")?;
+    stream.flush()?;
+    let fds = recv_fds(&stream, manifest.panes.len())?;
+    Ok(ReceivedHandoff {
+        manifest,
+        fds,
+        stream,
+    })
+}
+
+#[cfg(unix)]
+pub(crate) fn report_restored(stream: &mut UnixStream) -> io::Result<()> {
+    stream.write_all(b"restored\n")?;
+    stream.flush()
+}
+
+#[cfg(unix)]
+pub(crate) fn report_ready(stream: &mut UnixStream) -> io::Result<()> {
+    stream.write_all(b"ready\n")?;
+    stream.flush()
+}
+
+#[cfg(unix)]
+pub(crate) fn wait_committed(stream: &mut UnixStream) -> io::Result<()> {
+    stream.set_read_timeout(Some(READY_TIMEOUT))?;
+    let committed = read_line_unbuffered(&mut *stream)?;
+    if committed.trim_end() != "committed" {
+        return Err(io::Error::other("handoff source did not commit"));
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+pub(crate) fn report_owned(stream: &mut UnixStream) -> io::Result<()> {
+    stream.write_all(b"owned\n")?;
+    stream.flush()
+}
+
+#[cfg(unix)]
+pub(crate) fn manifest_for(
+    snapshot: crate::persist::SessionSnapshot,
+    panes: Vec<HandoffPane>,
+    expected_protocol: Option<u32>,
+    expected_version: Option<String>,
+) -> HandoffManifest {
+    HandoffManifest {
+        version: HANDOFF_VERSION,
+        source_version: env!("CARGO_PKG_VERSION").to_string(),
+        source_protocol: crate::protocol::PROTOCOL_VERSION,
+        expected_version,
+        expected_protocol,
+        snapshot,
+        panes,
+    }
+}
+
+#[cfg(unix)]
+fn restrict_socket_permissions(path: &Path) -> io::Result<()> {
+    use std::os::unix::fs::PermissionsExt;
+
+    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))
+}
+
+#[cfg(unix)]
+fn accept_with_timeout(
+    listener: &UnixListener,
+    timeout: Duration,
+) -> io::Result<(UnixStream, std::os::unix::net::SocketAddr)> {
+    let deadline = std::time::Instant::now() + timeout;
+    loop {
+        match listener.accept() {
+            Ok(accepted) => return Ok(accepted),
+            Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
+                if std::time::Instant::now() >= deadline {
+                    return Err(io::Error::new(
+                        io::ErrorKind::TimedOut,
+                        "timed out waiting for handoff import connection",
+                    ));
+                }
+                std::thread::sleep(Duration::from_millis(25));
+            }
+            Err(err) if err.kind() == io::ErrorKind::Interrupted => {}
+            Err(err) => return Err(err),
+        }
+    }
+}
+
+#[cfg(unix)]
+fn read_line_unbuffered(stream: &mut UnixStream) -> io::Result<String> {
+    let mut bytes = Vec::new();
+    let mut byte = [0u8; 1];
+    loop {
+        let read = stream.read(&mut byte)?;
+        if read == 0 {
+            return Err(io::Error::new(
+                io::ErrorKind::UnexpectedEof,
+                "handoff stream closed while reading line",
+            ));
+        }
+        bytes.push(byte[0]);
+        if byte[0] == b'\n' {
+            return String::from_utf8(bytes)
+                .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err));
+        }
+        if bytes.len() > 16 * 1024 * 1024 {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "handoff line exceeded maximum size",
+            ));
+        }
+    }
+}
+
+#[cfg(unix)]
+fn send_fds(stream: &UnixStream, fds: &[RawFd]) -> io::Result<()> {
+    if fds.is_empty() {
+        return Ok(());
+    }
+    let byte = [b'F'];
+    let iov = [libc::iovec {
+        iov_base: byte.as_ptr() as *mut libc::c_void,
+        iov_len: byte.len(),
+    }];
+    let fd_bytes = std::mem::size_of_val(fds);
+    let mut control = vec![0u8; unsafe { libc::CMSG_SPACE(fd_bytes as u32) as usize }];
+    let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
+    msg.msg_iov = iov.as_ptr() as *mut libc::iovec;
+    msg.msg_iovlen = iov.len() as _;
+    msg.msg_control = control.as_mut_ptr() as *mut libc::c_void;
+    msg.msg_controllen = control.len() as _;
+
+    unsafe {
+        let cmsg = libc::CMSG_FIRSTHDR(&msg);
+        if cmsg.is_null() {
+            return Err(io::Error::other("failed to allocate fd control message"));
+        }
+        (*cmsg).cmsg_level = libc::SOL_SOCKET;
+        (*cmsg).cmsg_type = libc::SCM_RIGHTS;
+        (*cmsg).cmsg_len = libc::CMSG_LEN(fd_bytes as u32) as _;
+        std::ptr::copy_nonoverlapping(fds.as_ptr() as *const u8, libc::CMSG_DATA(cmsg), fd_bytes);
+        if libc::sendmsg(stream.as_raw_fd(), &msg, 0) < 0 {
+            return Err(io::Error::last_os_error());
+        }
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+fn recv_fds(stream: &UnixStream, expected: usize) -> io::Result<Vec<RawFd>> {
+    if expected == 0 {
+        return Ok(Vec::new());
+    }
+    let mut byte = [0u8; 1];
+    let mut iov = [libc::iovec {
+        iov_base: byte.as_mut_ptr() as *mut libc::c_void,
+        iov_len: byte.len(),
+    }];
+    let fd_bytes = expected * std::mem::size_of::<RawFd>();
+    let mut control = vec![0u8; unsafe { libc::CMSG_SPACE(fd_bytes as u32) as usize }];
+    let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
+    msg.msg_iov = iov.as_mut_ptr();
+    msg.msg_iovlen = iov.len() as _;
+    msg.msg_control = control.as_mut_ptr() as *mut libc::c_void;
+    msg.msg_controllen = control.len() as _;
+
+    let read = unsafe { libc::recvmsg(stream.as_raw_fd(), &mut msg, 0) };
+    if read < 0 {
+        return Err(io::Error::last_os_error());
+    }
+    if msg.msg_flags & libc::MSG_CTRUNC != 0 {
+        return Err(io::Error::other("handoff fd control message was truncated"));
+    }
+
+    let mut out = Vec::new();
+    unsafe {
+        let cmsg = libc::CMSG_FIRSTHDR(&msg);
+        if cmsg.is_null()
+            || (*cmsg).cmsg_level != libc::SOL_SOCKET
+            || (*cmsg).cmsg_type != libc::SCM_RIGHTS
+        {
+            return Err(io::Error::other("handoff fd message missing SCM_RIGHTS"));
+        }
+        let data_len = ((*cmsg).cmsg_len as usize).saturating_sub(libc::CMSG_LEN(0) as usize);
+        let count = data_len / std::mem::size_of::<RawFd>();
+        let data = libc::CMSG_DATA(cmsg) as *const RawFd;
+        for idx in 0..count {
+            out.push(*data.add(idx));
+        }
+    }
+    if out.len() != expected {
+        for fd in out {
+            let _ = unsafe { libc::close(fd) };
+        }
+        return Err(io::Error::other(format!(
+            "expected {expected} handoff fds, received fewer"
+        )));
+    }
+    Ok(out)
+}
+
+#[cfg(unix)]
+pub(crate) fn log_import_result(panes: usize) {
+    info!(panes, "handoff import ready");
+}